diff --git a/docs/checks.json b/docs/checks.json index 50d8657f..0d266218 100644 --- a/docs/checks.json +++ b/docs/checks.json @@ -1221,6 +1221,57 @@ "requires_human_review_regardless_of_patch": false, "suggested_patch_kind": "manual" }, + { + "autofix_safe": false, + "category": "verify", + "default_severity": "medium", + "description": "The PR changes the config file that binds a dynamic toolkit's authority; static analysis cannot diff the effective tool list.", + "docs_url": "https://github.com/ThreeMoonsLab/agents-shipgate/blob/main/docs/checks.md#ship-cap-config-binding-changed", + "dynamic_default": false, + "evidence_fields": [ + "kind", + "provider", + "constructor", + "binding", + "config_path", + "changed_file", + "source_ref" + ], + "fires_when": "A VerificationContext and a base report are both present, the same factory site is config-bound on both sides, and the bound config file is in the PR's changed files.", + "floor_severity": null, + "id": "SHIP-CAP-CONFIG-BINDING-CHANGED", + "mvp_tier": "lifecycle", + "rationale": "A factory whose authority-bearing argument is bound from a config read hides its effective tool surface from static enumeration on both sides of the diff, so a config edit (e.g. adding \"refund\" to an actions list) produces no capability change in the generic diff. The config delta must be reviewed instead of passing invisibly.", + "recommendation": "Review the config delta for added capabilities, or declare an explicit local tool inventory for the toolkit so future changes are statically comparable.", + "requires_human_review": true, + "requires_human_review_regardless_of_patch": false, + "suggested_patch_kind": "manual" + }, + { + "autofix_safe": false, + "category": "verify", + "default_severity": "high", + "description": "The PR removes the config binding from a dynamic toolkit factory present on both sides of the diff, so the toolkit may fall back to its everything-enabled defaults.", + "docs_url": "https://github.com/ThreeMoonsLab/agents-shipgate/blob/main/docs/checks.md#ship-cap-config-binding-removed", + "dynamic_default": false, + "evidence_fields": [ + "kind", + "provider", + "constructor", + "binding", + "base_config_path", + "source_ref" + ], + "fires_when": "A VerificationContext and a base report are both present, the base bound the factory's authority-bearing argument to a config read, and the head constructs the same factory site with no binding at all. An unreadable (unknown) head binding never fires this check.", + "floor_severity": "high", + "id": "SHIP-CAP-CONFIG-BINDING-REMOVED", + "mvp_tier": "lifecycle", + "rationale": "A restriction-bearing config binding is the only statically-provable restraint on a toolkit whose tools load through a runtime factory. Removing the binding expands authority while looking like cleanup, and neither side's tools are statically enumerable, so the generic capability diff sees nothing. A coding agent must not self-approve that expansion; without this check the diff degrades to silent insufficient_evidence parity between base and head.", + "recommendation": "A human must review the removal; restore the config binding or an explicit literal allowlist, or declare an explicit local tool inventory for the toolkit so the mounted surface is statically enumerable.", + "requires_human_review": true, + "requires_human_review_regardless_of_patch": false, + "suggested_patch_kind": "manual" + }, { "autofix_safe": false, "category": "codex_boundary", @@ -2693,7 +2744,7 @@ "id": "SHIP-SCOPE-TOOLKIT-UNBOUNDED", "mvp_tier": "core", "rationale": "A recognized toolkit constructor (e.g. stripe_agent_toolkit) called with no\nconfiguration allowlist mounts its full tool surface \u2014 refund/cancel/dispute \u2014\nwhich the static extractor cannot enumerate. The unbounded grant must be\nreviewed rather than pass silently into insufficient_evidence.", - "recommendation": "Pass an explicit configuration allowlist (resource:verb actions) to the toolkit constructor so the agent mounts only the tools it needs.", + "recommendation": "Pass an explicit configuration allowlist (resource:verb actions) to the toolkit constructor so the agent mounts only the tools it needs. If the toolkit must stay dynamic, declare an explicit local tool inventory for the tools it mounts so the surface is statically reviewable.", "requires_human_review": true, "requires_human_review_regardless_of_patch": true, "suggested_patch_kind": "manual" diff --git a/docs/checks.md b/docs/checks.md index 9efb734e..caf21677 100644 --- a/docs/checks.md +++ b/docs/checks.md @@ -154,6 +154,8 @@ baseline summary and do not fail CI. | `SHIP-VERIFY-AGENT-INSTRUCTIONS-WEAKENED` | medium | The PR edits agent-instruction trust roots and weakening cannot be statically disproven; routed to human review. | | `SHIP-VERIFY-TRIGGER-CATALOG-DRIFT` | medium | The PR changes the trigger catalog that decides when Shipgate runs; routed to human review to rule out gate evasion. | | `SHIP-VERIFY-CAPABILITY-SCOPE-BROADENED` | critical | The PR removes or broadens a dynamically-loaded toolkit's least-privilege configuration bound (e.g. a `stripe_agent_toolkit` allowlist), silently expanding the toolkit surface; blocks rather than degrading to insufficient_evidence. | +| `SHIP-CAP-CONFIG-BINDING-REMOVED` | high | The PR removes the config binding from a dynamic toolkit factory present on both sides of the diff, so the toolkit may fall back to its everything-enabled defaults; routed to human review instead of silent insufficient_evidence parity. | +| `SHIP-CAP-CONFIG-BINDING-CHANGED` | medium | The PR changes the config file that binds a dynamic toolkit's authority; static analysis cannot diff the effective tool list, so the config delta is routed to review. | ## Check Details @@ -948,6 +950,30 @@ coding agent cannot self-approve the broadening; a human must re-apply a least-privilege configuration or explicitly approve the expanded surface. A narrowing (bound added or tightened) emits nothing. +### SHIP-CAP-CONFIG-BINDING-REMOVED + +Config-bound capability detection: a dynamic toolkit factory whose +authority-bearing constructor argument is *bound from a config read* +(`json`/`yaml`/`toml` load, `os.environ`, pydantic settings) hides its +effective tool surface from static enumeration on both sides of a verify +diff. When the same factory site appears on both sides and the head removes +the binding entirely, the toolkit may fall back to its everything-enabled +defaults — authority expands while looking like cleanup, and the generic +capability diff sees nothing. Fires `high` in verify mode and routes to +human review instead of silent `insufficient_evidence` parity. Fail-safe by +design: a head binding that is merely *unreadable* (`unknown`) never fires +this check — it degrades to a source warning; a head that moves to a +literal allowlist is the safe direction and emits nothing. + +### SHIP-CAP-CONFIG-BINDING-CHANGED + +Companion review item: the same factory site is config-bound on both sides +of the diff and the PR changes the bound config file itself (e.g. adding +`"refund"` to an actions list). Static analysis cannot diff the effective +tool list, so the config delta is routed to review at `medium` with the +inventory remedy. Fires only when the binding's literal config path matches +a changed file — env- or settings-bound factories never guess a match. + ### SHIP-MCP-ENV-SECRET-PASSTHROUGH Fires when a statically parsed MCP server passes through secret-like diff --git a/docs/checks/scope.yaml b/docs/checks/scope.yaml index 64ffd0bf..e224e608 100644 --- a/docs/checks/scope.yaml +++ b/docs/checks/scope.yaml @@ -57,5 +57,7 @@ checks: - binding - source_ref recommendation: Pass an explicit configuration allowlist (resource:verb actions) - to the toolkit constructor so the agent mounts only the tools it needs. + to the toolkit constructor so the agent mounts only the tools it needs. If the + toolkit must stay dynamic, declare an explicit local tool inventory for the + tools it mounts so the surface is statically reviewable. requires_human_review_regardless_of_patch: true diff --git a/docs/checks/verify.yaml b/docs/checks/verify.yaml index 7b42c115..10c501e0 100644 --- a/docs/checks/verify.yaml +++ b/docs/checks/verify.yaml @@ -163,3 +163,56 @@ checks: recommendation: A human must review and re-apply a least-privilege toolkit configuration or explicitly approve the broadened scope; do not remove the allowlist to expand agent capability without review. +- id: SHIP-CAP-CONFIG-BINDING-CHANGED + default_severity: medium + mvp_tier: lifecycle + requires_human_review: true + description: The PR changes the config file that binds a dynamic toolkit's + authority; static analysis cannot diff the effective tool list. + rationale: A factory whose authority-bearing argument is bound from a config + read hides its effective tool surface from static enumeration on both sides + of the diff, so a config edit (e.g. adding "refund" to an actions list) + produces no capability change in the generic diff. The config delta must be + reviewed instead of passing invisibly. + fires_when: A VerificationContext and a base report are both present, the same + factory site is config-bound on both sides, and the bound config file is in + the PR's changed files. + evidence_fields: + - kind + - provider + - constructor + - binding + - config_path + - changed_file + - source_ref + recommendation: Review the config delta for added capabilities, or declare an + explicit local tool inventory for the toolkit so future changes are + statically comparable. +- id: SHIP-CAP-CONFIG-BINDING-REMOVED + default_severity: high + floor_severity: high + mvp_tier: lifecycle + requires_human_review: true + description: The PR removes the config binding from a dynamic toolkit factory + present on both sides of the diff, so the toolkit may fall back to its + everything-enabled defaults. + rationale: A restriction-bearing config binding is the only statically-provable + restraint on a toolkit whose tools load through a runtime factory. Removing + the binding expands authority while looking like cleanup, and neither side's + tools are statically enumerable, so the generic capability diff sees nothing. + A coding agent must not self-approve that expansion; without this check the + diff degrades to silent insufficient_evidence parity between base and head. + fires_when: A VerificationContext and a base report are both present, the base + bound the factory's authority-bearing argument to a config read, and the head + constructs the same factory site with no binding at all. An unreadable + (unknown) head binding never fires this check. + evidence_fields: + - kind + - provider + - constructor + - binding + - base_config_path + - source_ref + recommendation: A human must review the removal; restore the config binding or + an explicit literal allowlist, or declare an explicit local tool inventory + for the toolkit so the mounted surface is statically enumerable. diff --git a/llms-full.txt b/llms-full.txt index 8781470d..e3114829 100644 --- a/llms-full.txt +++ b/llms-full.txt @@ -1665,6 +1665,8 @@ baseline summary and do not fail CI. | `SHIP-VERIFY-AGENT-INSTRUCTIONS-WEAKENED` | medium | The PR edits agent-instruction trust roots and weakening cannot be statically disproven; routed to human review. | | `SHIP-VERIFY-TRIGGER-CATALOG-DRIFT` | medium | The PR changes the trigger catalog that decides when Shipgate runs; routed to human review to rule out gate evasion. | | `SHIP-VERIFY-CAPABILITY-SCOPE-BROADENED` | critical | The PR removes or broadens a dynamically-loaded toolkit's least-privilege configuration bound (e.g. a `stripe_agent_toolkit` allowlist), silently expanding the toolkit surface; blocks rather than degrading to insufficient_evidence. | +| `SHIP-CAP-CONFIG-BINDING-REMOVED` | high | The PR removes the config binding from a dynamic toolkit factory present on both sides of the diff, so the toolkit may fall back to its everything-enabled defaults; routed to human review instead of silent insufficient_evidence parity. | +| `SHIP-CAP-CONFIG-BINDING-CHANGED` | medium | The PR changes the config file that binds a dynamic toolkit's authority; static analysis cannot diff the effective tool list, so the config delta is routed to review. | ## Check Details @@ -2459,6 +2461,30 @@ coding agent cannot self-approve the broadening; a human must re-apply a least-privilege configuration or explicitly approve the expanded surface. A narrowing (bound added or tightened) emits nothing. +### SHIP-CAP-CONFIG-BINDING-REMOVED + +Config-bound capability detection: a dynamic toolkit factory whose +authority-bearing constructor argument is *bound from a config read* +(`json`/`yaml`/`toml` load, `os.environ`, pydantic settings) hides its +effective tool surface from static enumeration on both sides of a verify +diff. When the same factory site appears on both sides and the head removes +the binding entirely, the toolkit may fall back to its everything-enabled +defaults — authority expands while looking like cleanup, and the generic +capability diff sees nothing. Fires `high` in verify mode and routes to +human review instead of silent `insufficient_evidence` parity. Fail-safe by +design: a head binding that is merely *unreadable* (`unknown`) never fires +this check — it degrades to a source warning; a head that moves to a +literal allowlist is the safe direction and emits nothing. + +### SHIP-CAP-CONFIG-BINDING-CHANGED + +Companion review item: the same factory site is config-bound on both sides +of the diff and the PR changes the bound config file itself (e.g. adding +`"refund"` to an actions list). Static analysis cannot diff the effective +tool list, so the config delta is routed to review at `medium` with the +inventory remedy. Fires only when the binding's literal config path matches +a changed file — env- or settings-bound factories never guess a match. + ### SHIP-MCP-ENV-SECRET-PASSTHROUGH Fires when a statically parsed MCP server passes through secret-like diff --git a/src/agents_shipgate/checks/_verify_common.py b/src/agents_shipgate/checks/_verify_common.py index e16c948c..174e7ed6 100644 --- a/src/agents_shipgate/checks/_verify_common.py +++ b/src/agents_shipgate/checks/_verify_common.py @@ -25,11 +25,15 @@ from __future__ import annotations +from collections import defaultdict + from agents_shipgate.core.context import ScanContext +from agents_shipgate.core.domain import ToolkitScopeBound from agents_shipgate.core.globbing import glob_match from agents_shipgate.core.lenses.effective_policy import ( build_effective_policy_snapshot, ) +from agents_shipgate.core.toolkit_scope import bound_from_policy_fact, policy_key_for from agents_shipgate.schemas.capability_change import EffectivePolicy from agents_shipgate.schemas.common import ( SourceReference, @@ -100,6 +104,67 @@ def base_effective_policy(context: ScanContext) -> EffectivePolicy | None: return getattr(reference, "effective_policy", None) +def base_toolkit_bounds(context: ScanContext) -> dict[str, ToolkitScopeBound]: + """BASE-side toolkit scope bounds, decoded from the ``--diff-from`` + reference report's carried ``toolkit_scope_bound`` policy facts. + + Empty when there is no diff reference or the base carried no bounds — + callers degrade safely (no base, no comparison). + """ + reference = context.diff_reference + facts = getattr(reference, "facts", None) if reference is not None else None + if facts is None: + return {} + bounds: dict[str, ToolkitScopeBound] = {} + for fact in facts.policies: + bound = bound_from_policy_fact(fact) + if bound is not None: + bounds[policy_key_for(bound)] = bound + return bounds + + +def head_toolkit_bounds(context: ScanContext) -> dict[str, ToolkitScopeBound]: + """HEAD-side toolkit scope bounds from the live extraction.""" + return {policy_key_for(bound): bound for bound in context.toolkit_bounds} + + +def pair_toolkit_bounds( + base: dict[str, ToolkitScopeBound], head: dict[str, ToolkitScopeBound] +) -> list[tuple[ToolkitScopeBound, ToolkitScopeBound]]: + """Pair base↔head bounds for comparison, keeping instances distinct. + + First by exact ``provider:binding`` key, so multiple toolkit instances of a + provider do not collapse. Then a leftover fallback: for a provider with + exactly one unmatched bound on each side, pair them — this re-pairs a + *renamed* single toolkit so a rename-plus-weaken cannot slip through. + Ambiguous leftovers (more than one unmatched per side) are left unpaired + (fail safe). + """ + pairs: list[tuple[ToolkitScopeBound, ToolkitScopeBound]] = [] + matched_head: set[str] = set() + base_left: list[ToolkitScopeBound] = [] + for key in sorted(base): + if key in head: + pairs.append((base[key], head[key])) + matched_head.add(key) + else: + base_left.append(base[key]) + head_left = [head[key] for key in sorted(head) if key not in matched_head] + + base_by_provider: dict[str, list[ToolkitScopeBound]] = defaultdict(list) + head_by_provider: dict[str, list[ToolkitScopeBound]] = defaultdict(list) + for bound in base_left: + base_by_provider[bound.provider].append(bound) + for bound in head_left: + head_by_provider[bound.provider].append(bound) + for provider in sorted(base_by_provider): + bl = base_by_provider[provider] + hl = head_by_provider.get(provider, []) + if len(bl) == 1 and len(hl) == 1: + pairs.append((bl[0], hl[0])) + return pairs + + def verify_finding( context: ScanContext, *, @@ -138,8 +203,11 @@ def verify_finding( __all__ = [ "SEVERITY_RANK", "base_effective_policy", + "base_toolkit_bounds", "changed_files", "head_effective_policy", + "head_toolkit_bounds", + "pair_toolkit_bounds", "touched", "verification_active", "verify_finding", diff --git a/src/agents_shipgate/checks/registry.py b/src/agents_shipgate/checks/registry.py index 99de9500..865502e4 100644 --- a/src/agents_shipgate/checks/registry.py +++ b/src/agents_shipgate/checks/registry.py @@ -31,6 +31,7 @@ verify_baseline_waiver, verify_capability_scope, verify_ci_gate, + verify_config_binding, verify_policy, verify_trigger_drift, ) @@ -77,6 +78,7 @@ verify_agent_instructions.run, verify_trigger_drift.run, verify_capability_scope.run, + verify_config_binding.run, ] diff --git a/src/agents_shipgate/checks/toolkit_bounds.py b/src/agents_shipgate/checks/toolkit_bounds.py index acba84d6..3b80f12f 100644 --- a/src/agents_shipgate/checks/toolkit_bounds.py +++ b/src/agents_shipgate/checks/toolkit_bounds.py @@ -33,6 +33,13 @@ def run(context: ScanContext) -> list[Finding]: # An explicit resource:verb allowlist was declared — least # privilege, exactly what we want. Nothing to flag. continue + if bound.config_binding in {"config", "unknown"}: + # A configuration IS passed — it lives in config (or is not + # statically readable), so "mounted without a scope bound" would + # be a false claim. The config-binding detections + # (SHIP-CAP-CONFIG-BINDING-*) and the unknown-binding source + # warning own those states. + continue findings.append( Finding( check_id=CHECK_ID, @@ -62,7 +69,11 @@ def run(context: ScanContext) -> list[Finding]: recommendation=( f"Pass an explicit `configuration` allowlist (resource:verb " f"actions) to the {bound.constructor} constructor so the agent " - "mounts only the tools it needs, not the full toolkit surface." + "mounts only the tools it needs, not the full toolkit surface. " + "If the toolkit must stay dynamic, declare an explicit local " + "tool inventory for the tools it mounts at " + f"{bound.source_ref}:{bound.source_line} so the surface is " + "statically reviewable." ), ) ) diff --git a/src/agents_shipgate/checks/verify_capability_scope.py b/src/agents_shipgate/checks/verify_capability_scope.py index c50e0d30..c042fa6b 100644 --- a/src/agents_shipgate/checks/verify_capability_scope.py +++ b/src/agents_shipgate/checks/verify_capability_scope.py @@ -28,12 +28,15 @@ from __future__ import annotations -from collections import defaultdict - -from agents_shipgate.checks._verify_common import verification_active, verify_finding +from agents_shipgate.checks._verify_common import ( + base_toolkit_bounds, + head_toolkit_bounds, + pair_toolkit_bounds, + verification_active, + verify_finding, +) from agents_shipgate.core.context import ScanContext from agents_shipgate.core.domain import ToolkitScopeBound -from agents_shipgate.core.toolkit_scope import bound_from_policy_fact, policy_key_for from agents_shipgate.schemas.report import Finding CHECK_ID = "SHIP-VERIFY-CAPABILITY-SCOPE-BROADENED" @@ -42,76 +45,22 @@ def run(context: ScanContext) -> list[Finding]: if not verification_active(context): return [] - base_bounds = _base_bounds(context) + base_bounds = base_toolkit_bounds(context) if not base_bounds: # No base toolkit bound to compare against (no diff reference, a # pre-feature base, or a base with no recognized toolkit). Degrade # safely: the dynamic-toolkit surface still reads as low-confidence # evidence elsewhere, so this is never a silent pass. return [] - head_bounds = _head_bounds(context) + head_bounds = head_toolkit_bounds(context) findings: list[Finding] = [] - for base_bound, head_bound in _pair_bounds(base_bounds, head_bounds): + for base_bound, head_bound in pair_toolkit_bounds(base_bounds, head_bounds): weakening = _classify(base_bound, head_bound) if weakening is not None: findings.append(_finding(context, base_bound, head_bound, weakening)) return findings -def _base_bounds(context: ScanContext) -> dict[str, ToolkitScopeBound]: - reference = context.diff_reference - facts = getattr(reference, "facts", None) if reference is not None else None - if facts is None: - return {} - bounds: dict[str, ToolkitScopeBound] = {} - for fact in facts.policies: - bound = bound_from_policy_fact(fact) - if bound is not None: - bounds[policy_key_for(bound)] = bound - return bounds - - -def _head_bounds(context: ScanContext) -> dict[str, ToolkitScopeBound]: - return {policy_key_for(bound): bound for bound in context.toolkit_bounds} - - -def _pair_bounds( - base: dict[str, ToolkitScopeBound], head: dict[str, ToolkitScopeBound] -) -> list[tuple[ToolkitScopeBound, ToolkitScopeBound]]: - """Pair base↔head bounds for comparison, keeping instances distinct. - - First by exact ``provider:binding`` key, so multiple toolkit instances of a - provider do not collapse. Then a leftover fallback: for a provider with - exactly one unmatched bound on each side, pair them — this re-pairs a - *renamed* single toolkit so a rename-plus-broaden cannot slip through. - Ambiguous leftovers (more than one unmatched per side) are left unpaired - (fail safe). - """ - pairs: list[tuple[ToolkitScopeBound, ToolkitScopeBound]] = [] - matched_head: set[str] = set() - base_left: list[ToolkitScopeBound] = [] - for key in sorted(base): - if key in head: - pairs.append((base[key], head[key])) - matched_head.add(key) - else: - base_left.append(base[key]) - head_left = [head[key] for key in sorted(head) if key not in matched_head] - - base_by_provider: dict[str, list[ToolkitScopeBound]] = defaultdict(list) - head_by_provider: dict[str, list[ToolkitScopeBound]] = defaultdict(list) - for bound in base_left: - base_by_provider[bound.provider].append(bound) - for bound in head_left: - head_by_provider[bound.provider].append(bound) - for provider in sorted(base_by_provider): - bl = base_by_provider[provider] - hl = head_by_provider.get(provider, []) - if len(bl) == 1 and len(hl) == 1: - pairs.append((bl[0], hl[0])) - return pairs - - def _classify(base: ToolkitScopeBound, head: ToolkitScopeBound) -> dict | None: """Return weakening evidence, or ``None`` for no-change / narrowing. @@ -120,6 +69,13 @@ def _classify(base: ToolkitScopeBound, head: ToolkitScopeBound) -> dict | None: """ if not base.bounded: return None + if head.config_binding in {"config", "unknown"}: + # The head still passes a configuration — it moved into config (or + # into something the static parser cannot read). That is not "full + # surface mounted", so the removed/broadened claims would be false; + # the SHIP-CAP-CONFIG-BINDING-* detections and the unknown-binding + # source warning own these transitions. Fail safe, never guess. + return None base_scopes = sorted(set(base.scopes)) if not head.bounded: return { diff --git a/src/agents_shipgate/checks/verify_config_binding.py b/src/agents_shipgate/checks/verify_config_binding.py new file mode 100644 index 00000000..4de3b4ab --- /dev/null +++ b/src/agents_shipgate/checks/verify_config_binding.py @@ -0,0 +1,222 @@ +"""SHIP-CAP-CONFIG-BINDING-* — config-bound dynamic-toolkit authority. + +Implements the detection half of +docs/engineering/config-bound-capability-detection.md. A dynamic toolkit +factory (``*toolkit.get_tools()``) whose authority-bearing constructor +argument is bound from a *config read* hides its effective tool surface from +static enumeration on BOTH sides of a verify diff — so a PR that changes or +removes the binding produces no capability change in the generic diff at all. +The removal case is the dangerous one: deleting a restrictive config line +*expands* authority while looking like cleanup. + +Two detections, both on factory sites present on both sides of the diff +(paired the same way as ``SHIP-VERIFY-CAPABILITY-SCOPE-BROADENED``): + +- ``config-bound → absent`` ⇒ ``SHIP-CAP-CONFIG-BINDING-REMOVED`` + (severity ``high``): the restriction-bearing config binding was removed + from the factory call; the effective tool surface may have expanded to + the toolkit's defaults. +- ``config-bound → config-bound`` with the referenced config file in the + PR's ``changed_files`` ⇒ ``SHIP-CAP-CONFIG-BINDING-CHANGED`` (severity + ``medium``, a review item): the config that binds this toolkit's + authority changed; static analysis cannot diff the effective tool list. + +Calibration guards (the design doc's false-positive warning): + +- A head binding of ``unknown`` (present but not statically readable) NEVER + fires the removal check — it degrades to the extractor's source warning. +- A head that moves to a *literal* allowlist is the safe direction (the + authority became statically visible); nothing fires. +- Everything asserts "authority may have changed invisibly", never + "authority is X": confidence stays explicit and the findings flow into + the one decision engine (category ``verify``: suppression-immune, + floor-protected, no second verdict). +""" + +from __future__ import annotations + +from agents_shipgate.checks._verify_common import ( + base_toolkit_bounds, + changed_files, + head_toolkit_bounds, + pair_toolkit_bounds, + verification_active, + verify_finding, +) +from agents_shipgate.core.context import ScanContext +from agents_shipgate.core.domain import ToolkitScopeBound +from agents_shipgate.schemas.report import Finding + +CHECK_ID_REMOVED = "SHIP-CAP-CONFIG-BINDING-REMOVED" +CHECK_ID_CHANGED = "SHIP-CAP-CONFIG-BINDING-CHANGED" + + +def run(context: ScanContext) -> list[Finding]: + if not verification_active(context): + return [] + base_bounds = base_toolkit_bounds(context) + if not base_bounds: + # No base reference (or a pre-feature base with no carried bounds): + # degrade safely — the dynamic-toolkit surface still reads as + # low-confidence evidence elsewhere, never a silent pass. + return [] + head_bounds = head_toolkit_bounds(context) + files = changed_files(context) + findings: list[Finding] = [] + for base_bound, head_bound in pair_toolkit_bounds(base_bounds, head_bounds): + if base_bound.config_binding != "config": + continue + if head_bound.config_binding == "absent": + findings.append(_removed_finding(context, base_bound, head_bound)) + elif head_bound.config_binding == "config": + base_path = base_bound.config_path + head_path = head_bound.config_path + if base_path and head_path and base_path != head_path: + # Retarget: the binding moved to a different — possibly + # pre-existing — config file. Neither file's content need + # change for the effective tool surface to change, so this + # must not depend on the config appearing in changed_files + # (review-reproduced blind spot: safe.json → broad.json). + # Conservative: both sides must carry a literal path; + # env/settings bindings without a path never compare. + findings.append( + _retargeted_finding(context, base_bound, head_bound) + ) + else: + hit = _changed_config_file(head_path or base_path, files) + if hit is not None: + findings.append( + _changed_finding(context, base_bound, head_bound, hit) + ) + # head "unknown": never fire the removal check on an unreadable + # binding (design-doc guard); the extractor's source warning routes + # it to review. head "literal": authority became statically visible + # — the safe direction; the normal scope-bound checks own it. + return findings + + +def _changed_config_file(config_path: str | None, files: list[str]) -> str | None: + """The changed file matching the factory's bound config path, if any. + + Conservative: an exact repo-relative match, or a changed file whose + path ends with ``/`` (the AST literal may be relative to a + subdirectory). ``None`` when the binding carried no literal path (env / + settings reads) — never guess a match. + """ + if not config_path: + return None + normalized = config_path.lstrip("./") + for changed in files: + if changed == normalized or changed.endswith(f"/{normalized}"): + return changed + return None + + +def _site(bound: ToolkitScopeBound) -> str: + if bound.source_ref and bound.source_line is not None: + return f"{bound.source_ref}:{bound.source_line}" + return bound.source_ref or "the factory call site" + + +def _removed_finding( + context: ScanContext, + base_bound: ToolkitScopeBound, + head_bound: ToolkitScopeBound, +) -> Finding: + provider = head_bound.provider or base_bound.provider + site = _site(head_bound) + base_source = ( + f"configuration read from {base_bound.config_path}" + if base_bound.config_path + else "a configuration read the base bound statically traced" + ) + return verify_finding( + context, + check_id=CHECK_ID_REMOVED, + title=f"{provider} toolkit config binding removed", + severity="high", + evidence={ + "kind": "config_binding_removed", + "provider": provider, + "constructor": head_bound.constructor or base_bound.constructor, + "binding": head_bound.binding or base_bound.binding or "", + "base_config_path": base_bound.config_path or "", + "source_ref": head_bound.source_ref or "", + }, + recommendation=( + f"The base bound this {provider} toolkit's authority to " + f"{base_source}; the head constructs it at {site} without that " + "binding, so the toolkit may mount its everything-enabled " + "defaults — the effective tool surface can expand invisibly. A " + "human must review: restore the config binding (or an explicit " + "literal allowlist), or declare an explicit local tool inventory " + f"for the toolkit constructed at {site} so the mounted surface " + "is statically enumerable." + ), + ) + + +def _retargeted_finding( + context: ScanContext, + base_bound: ToolkitScopeBound, + head_bound: ToolkitScopeBound, +) -> Finding: + provider = head_bound.provider or base_bound.provider + site = _site(head_bound) + return verify_finding( + context, + check_id=CHECK_ID_CHANGED, + title=f"{provider} toolkit authority config retargeted", + severity="medium", + evidence={ + "kind": "config_binding_retargeted", + "provider": provider, + "constructor": head_bound.constructor or base_bound.constructor, + "binding": head_bound.binding or "", + "config_path": head_bound.config_path or "", + "previous_config_path": base_bound.config_path or "", + "source_ref": head_bound.source_ref or "", + }, + recommendation=( + f"This PR points the {provider} toolkit's authority binding at " + f"{site} to a different config ({base_bound.config_path} → " + f"{head_bound.config_path}). The new file may pre-exist with a " + "broader action list, so no config diff appears in this PR — " + "review the new config's effective capabilities, or declare an " + f"explicit local tool inventory for the toolkit at {site} so " + "changes stay statically comparable." + ), + ) + + +def _changed_finding( + context: ScanContext, + base_bound: ToolkitScopeBound, + head_bound: ToolkitScopeBound, + changed_file: str, +) -> Finding: + provider = head_bound.provider or base_bound.provider + site = _site(head_bound) + return verify_finding( + context, + check_id=CHECK_ID_CHANGED, + title=f"{provider} toolkit authority config changed", + severity="medium", + evidence={ + "kind": "config_binding_changed", + "provider": provider, + "constructor": head_bound.constructor or base_bound.constructor, + "binding": head_bound.binding or "", + "config_path": head_bound.config_path or base_bound.config_path or "", + "changed_file": changed_file, + "source_ref": head_bound.source_ref or "", + }, + recommendation=( + f"This PR changes {changed_file}, the config that binds the " + f"{provider} toolkit's authority at {site}. Static analysis " + "cannot diff the effective tool list — review the config delta " + "for added capabilities, or declare an explicit local tool " + f"inventory for the toolkit constructed at {site} so future " + "changes are statically comparable." + ), + ) diff --git a/src/agents_shipgate/cli/scan/decision.py b/src/agents_shipgate/cli/scan/decision.py index 7917c91f..493048c6 100644 --- a/src/agents_shipgate/cli/scan/decision.py +++ b/src/agents_shipgate/cli/scan/decision.py @@ -77,6 +77,11 @@ def _run_checks_and_decide( action_surface_facts, action_reference.facts if action_reference else None, reference=action_reference, + # Fail-soft: duplicate action ids in an engine-generated base + # reference (pre-collision-fix report/baseline) degrade to a + # source_warning (-> review_required) instead of crashing the scan + # with a config error. Same principle as the build-time sink above. + warnings=action_surface_warnings, ) if diffs.diff_reference_error: action_surface_diff.enabled = False diff --git a/src/agents_shipgate/core/domain.py b/src/agents_shipgate/core/domain.py index c63718cf..c68a6556 100644 --- a/src/agents_shipgate/core/domain.py +++ b/src/agents_shipgate/core/domain.py @@ -121,6 +121,23 @@ class ToolkitScopeBound(BaseModel): binding: str | None = None source_ref: str | None = None source_line: int | None = None + # Config-bound capability detection + # (docs/engineering/config-bound-capability-detection.md): how the + # constructor's authority-bearing argument was bound. + # + # - ``"literal"`` — parsed to an explicit allowlist (``scopes`` above). + # - ``"absent"`` — no authority argument; the framework defaults mount. + # - ``"config"`` — statically traceable to a config read (json/yaml/toml + # load, ``os.environ``, pydantic settings); the effective surface is + # configuration-defined and NOT statically enumerable. + # - ``"unknown"`` — present but not statically readable. Never fires the + # removal check (the design doc's false-positive guard). + # - ``None`` — legacy fact (pre-feature report round-trip); treated + # per ``bounded`` by the decoders. + config_binding: Literal["literal", "config", "absent", "unknown"] | None = None + # Repo-relative path of the config file feeding a ``"config"`` binding + # when the read named a literal path; ``None`` for env/settings reads. + config_path: str | None = None class LoadedToolSource(BaseModel): diff --git a/src/agents_shipgate/core/lenses/action_surface.py b/src/agents_shipgate/core/lenses/action_surface.py index 8195c5af..e3b13733 100644 --- a/src/agents_shipgate/core/lenses/action_surface.py +++ b/src/agents_shipgate/core/lenses/action_surface.py @@ -188,7 +188,21 @@ def compute_action_surface_diff( base: ActionSurfaceFacts | None, *, reference: ActionSurfaceDiffReference | None = None, + warnings: list[str] | None = None, ) -> ActionSurfaceDiff: + """Diff two action-surface snapshots. + + With a ``warnings`` sink, duplicate ``action_id`` values on either side + degrade fail-soft instead of raising :class:`ConfigError`. By the time + facts reach the diff they are engine-built artifacts, not user config: + the manifest-authored identity contract was already enforced when each + side was built (see :func:`build_action_surface_facts`). In practice the + duplicates come from a ``--diff-from`` report or baseline serialized by + a pre-collision-fix engine (the block/goose miner shape) — a gate must + fail-safe with a warning on inputs it inferred itself, not crash the + scan with a config error pointing at a manifest that is fine. Callers + without a sink keep the legacy hard :class:`ConfigError`. + """ if base is None: notes = _action_notes(reference) if not notes: @@ -199,8 +213,16 @@ def compute_action_surface_diff( notes=notes, ) - current_by_id = _actions_by_id(current.actions) - base_by_id = _actions_by_id(base.actions) + if warnings is None: + current_by_id = _actions_by_id(current.actions) + base_by_id = _actions_by_id(base.actions) + else: + current_by_id = _actions_by_id_fail_soft( + current.actions, side="current", warnings=warnings + ) + base_by_id = _actions_by_id_fail_soft( + base.actions, side="base reference", warnings=warnings + ) added = [ _action_added_change(current_by_id[action_id]) for action_id in sorted(current_by_id.keys() - base_by_id.keys()) @@ -573,6 +595,34 @@ def _actions_by_id(actions: list[ActionFact]) -> dict[str, ActionFact]: return {action.action_id: action for action in actions} +def _actions_by_id_fail_soft( + actions: list[ActionFact], + *, + side: str, + warnings: list[str], +) -> dict[str, ActionFact]: + """Index actions by id, degrading duplicates instead of raising. + + Reuses :func:`_disambiguate_duplicate_action_ids` with no explicit + tool names: at diff time both snapshots are engine output whose + manifest-authored identities were validated at build time, so every + surviving collision is inferred (most commonly a base report written + before inferred collisions were disambiguated at build time). The + tool-name suffix strategy matches what the head side's build-time + disambiguator produces, so a degraded base lines up with a fresh head + instead of flagging the whole surface as churned. + """ + for message in _disambiguate_duplicate_action_ids( + actions, explicit_tool_names=set() + ): + warnings.append( + f"action_surface_diff ({side}): {message} The colliding ids were " + "read from an engine-generated snapshot, typically a --diff-from " + "report or baseline produced by an older version." + ) + return {action.action_id: action for action in actions} + + def _validate_unique_action_ids(actions: list[ActionFact]) -> None: by_id: dict[str, list[str]] = {} for action in actions: diff --git a/src/agents_shipgate/core/toolkit_scope.py b/src/agents_shipgate/core/toolkit_scope.py index 028e0dfb..97eb4932 100644 --- a/src/agents_shipgate/core/toolkit_scope.py +++ b/src/agents_shipgate/core/toolkit_scope.py @@ -33,6 +33,14 @@ # (``configuration={"actions": {}}`` — bounded to nothing), so the two # decode to different ``bounded`` states. UNBOUNDED_SUMMARY = "(unbounded — full toolkit surface)" +# Config-bound / unreadable-binding sentinels (docs/engineering/ +# config-bound-capability-detection.md). Same carriage trick: the summary +# is both the rendering and the round-trip encoding, so the base side of a +# verify diff recovers ``config_binding`` (and the bound config file's +# path) from the serialized base report without any schema change. +CONFIG_BOUND_SUMMARY_BARE = "(config-bound)" +_CONFIG_BOUND_SUMMARY_PREFIX = "(config-bound: " +UNKNOWN_BINDING_SUMMARY = "(binding unknown — configuration not statically readable)" _SCOPE_SEP = ", " @@ -51,17 +59,29 @@ def policy_key_for(bound: ToolkitScopeBound) -> str: def _bound_summary(bound: ToolkitScopeBound) -> str: + if bound.config_binding == "config": + if bound.config_path: + return f"{_CONFIG_BOUND_SUMMARY_PREFIX}{bound.config_path})" + return CONFIG_BOUND_SUMMARY_BARE + if bound.config_binding == "unknown": + return UNKNOWN_BINDING_SUMMARY if not bound.bounded: return UNBOUNDED_SUMMARY return _SCOPE_SEP.join(sorted(bound.scopes)) def _bound_value_hash(bound: ToolkitScopeBound) -> str: - payload = json.dumps( - {"bounded": bound.bounded, "scopes": sorted(bound.scopes)}, - sort_keys=True, - separators=(",", ":"), - ) + value: dict[str, object] = { + "bounded": bound.bounded, + "scopes": sorted(bound.scopes), + } + if bound.config_binding in {"config", "unknown"}: + # Only the new binding states extend the hash payload: literal and + # absent bounds keep the legacy payload so their serialized + # ``value_hash`` stays byte-identical across the upgrade. + value["config_binding"] = bound.config_binding + value["config_path"] = bound.config_path + payload = json.dumps(value, sort_keys=True, separators=(",", ":")) return hashlib.sha256(payload.encode("utf-8")).hexdigest()[:16] @@ -95,6 +115,31 @@ def bound_from_policy_fact(fact: ToolSurfacePolicyFact) -> ToolkitScopeBound | N bounded=False, scopes=[], binding=binding_value, + config_binding="absent", + ) + if summary == UNKNOWN_BINDING_SUMMARY: + return ToolkitScopeBound( + provider=provider, + constructor="", + bounded=False, + scopes=[], + binding=binding_value, + config_binding="unknown", + ) + if summary == CONFIG_BOUND_SUMMARY_BARE or summary.startswith( + _CONFIG_BOUND_SUMMARY_PREFIX + ): + config_path = None + if summary.startswith(_CONFIG_BOUND_SUMMARY_PREFIX): + config_path = summary[len(_CONFIG_BOUND_SUMMARY_PREFIX) : -1] or None + return ToolkitScopeBound( + provider=provider, + constructor="", + bounded=False, + scopes=[], + binding=binding_value, + config_binding="config", + config_path=config_path, ) scopes = [token for token in summary.split(_SCOPE_SEP) if token] if summary else [] return ToolkitScopeBound( @@ -103,6 +148,7 @@ def bound_from_policy_fact(fact: ToolSurfacePolicyFact) -> ToolkitScopeBound | N bounded=True, scopes=sorted(scopes), binding=binding_value, + config_binding="literal", ) @@ -123,8 +169,10 @@ def toolkit_bound_facts( __all__ = [ + "CONFIG_BOUND_SUMMARY_BARE", "TOOLKIT_BOUND_POLICY_KIND", "UNBOUNDED_SUMMARY", + "UNKNOWN_BINDING_SUMMARY", "bound_from_policy_fact", "bound_to_policy_fact", "policy_key_for", diff --git a/src/agents_shipgate/inputs/config_trace.py b/src/agents_shipgate/inputs/config_trace.py new file mode 100644 index 00000000..c786027b --- /dev/null +++ b/src/agents_shipgate/inputs/config_trace.py @@ -0,0 +1,329 @@ +"""Conservative static trace: is an expression bound from a config read? + +Supports the config-bound capability detection design +(docs/engineering/config-bound-capability-detection.md). A dynamic toolkit +factory's authority-bearing argument is classified as one of: + +- ``config`` — statically traceable to a *config read*: a ``json`` / + ``yaml`` / ``toml`` / ``tomllib`` load call, an ``os.environ`` / + ``os.getenv`` read, or the instantiation of an in-file pydantic + ``BaseSettings`` subclass. +- ``unknown`` — present but not statically readable (a call we do not + recognize, a name we cannot resolve, an ambiguous reassignment). + +The trace is deliberately conservative: same-file, same-scope assignment +tracing only — the enclosing function's body first, then module-level +statements — with no interprocedural or cross-module dataflow. A name +assigned more than once in the searched scope is ambiguous and resolves to +``unknown``. The design doc is explicit that guessing here ships a +false-positive generator: ``unknown`` must never fire the removal check. +""" + +from __future__ import annotations + +import ast +from dataclasses import dataclass + +from agents_shipgate.inputs.python_static import dotted_name + +# Dotted call names that read configuration. Deliberately small: each entry +# is a well-known "parse config from file/env" entry point, not a generic +# I/O call. ``os.environ.get`` is handled as a dotted attribute call; +# ``os.environ[...]`` subscripts are handled structurally below. +_CONFIG_READ_CALLS: frozenset[str] = frozenset( + { + "json.load", + "json.loads", + "yaml.safe_load", + "yaml.load", + "yaml.full_load", + "toml.load", + "toml.loads", + "tomllib.load", + "tomllib.loads", + "os.getenv", + "os.environ.get", + "configparser.ConfigParser", + } +) + +# Modules whose ``from import `` symbols count as config +# readers when called bare (e.g. ``from json import load`` → ``load(...)``). +_CONFIG_READ_FROM_IMPORTS: dict[str, frozenset[str]] = { + "json": frozenset({"load", "loads"}), + "yaml": frozenset({"safe_load", "load", "full_load"}), + "toml": frozenset({"load", "loads"}), + "tomllib": frozenset({"load", "loads"}), + "os": frozenset({"getenv"}), +} + +# Readers whose first argument is a variable/section name, never a file path. +_NON_PATH_READERS: frozenset[str] = frozenset( + {"os.getenv", "os.environ.get", "configparser.ConfigParser"} +) + +_SETTINGS_BASE_SUFFIXES = ("BaseSettings",) + +_MAX_TRACE_DEPTH = 3 + + +@dataclass(frozen=True) +class ConfigBindingTrace: + """Outcome of tracing one authority-bearing expression.""" + + binding: str # "config" | "unknown" + config_path: str | None = None + reader: str | None = None + + +def trace_config_binding( + expr: ast.expr, + *, + tree: ast.Module, + line: int, +) -> ConfigBindingTrace: + """Classify ``expr`` (at ``line`` in ``tree``) as config-bound or unknown. + + ``line`` locates the enclosing function so the trace searches the same + scope the expression executes in; module-level expressions search the + module body. Never raises; anything unresolvable is ``unknown``. + """ + scopes = _scopes_for_line(tree, line) + readers = _reader_names(tree) + settings_classes = _settings_class_names(tree) + return _trace(expr, scopes, readers, settings_classes, depth=0) + + +def _trace( + expr: ast.expr, + scopes: list[dict[str, ast.expr | None]], + readers: dict[str, str], + settings_classes: set[str], + *, + depth: int, +) -> ConfigBindingTrace: + if depth > _MAX_TRACE_DEPTH: + return ConfigBindingTrace(binding="unknown") + node: ast.expr | None = expr + # Unwrap value-preserving wrappers: ``await x``, ``x["key"]``, + # ``x.attr`` all carry the authority of their root object. + while True: + if isinstance(node, ast.Await): + node = node.value + continue + if isinstance(node, ast.Subscript): + # ``os.environ["NAME"]`` is itself a config read. + if dotted_name(node.value) == "os.environ": + return ConfigBindingTrace(binding="config", reader="os.environ") + node = node.value + continue + if isinstance(node, ast.Attribute): + # Attribute access carries its root's authority: + # ``settings.actions`` → ``settings``; ``Settings().actions`` + # → the ``Settings()`` call. + node = node.value + continue + break + if isinstance(node, ast.Call): + classified = _classify_call(node, readers, settings_classes) + if classified is not None: + return classified + # ``cfg.get("actions")``-style method call on a traced name: the + # authority is the receiver's, so trace it. + if isinstance(node.func, ast.Attribute): + return _trace( + node.func.value, scopes, readers, settings_classes, depth=depth + 1 + ) + return ConfigBindingTrace(binding="unknown") + if isinstance(node, ast.Name): + resolved = _resolve_name(node.id, scopes) + if resolved is None: + return ConfigBindingTrace(binding="unknown") + return _trace(resolved, scopes, readers, settings_classes, depth=depth + 1) + return ConfigBindingTrace(binding="unknown") + + +def _classify_call( + call: ast.Call, + readers: dict[str, str], + settings_classes: set[str], +) -> ConfigBindingTrace | None: + name = dotted_name(call.func) + if name is None: + # ``Path("x").read_text()``-style chains have a Call in the attribute + # root; ``dotted_name`` cannot render them. Not a recognized reader. + return None + canonical = readers.get(name, name) + if canonical in _CONFIG_READ_CALLS: + return ConfigBindingTrace( + binding="config", + # Env readers take a variable name, not a file path. + config_path=( + None + if canonical in _NON_PATH_READERS + else _config_path_from_call(call) + ), + reader=canonical, + ) + if name in settings_classes: + return ConfigBindingTrace(binding="config", reader="pydantic_settings") + return None + + +def _config_path_from_call(call: ast.Call) -> str | None: + """Best-effort literal path of the file feeding a config-read call. + + Recognizes ``load("cfg.toml")``, ``load(open("cfg.json"))``, and + ``loads(Path("cfg.yaml").read_text())``. Env/settings reads and + non-literal paths return ``None`` (the detection then degrades to a + path-less binding — never a guess). + """ + if not call.args: + return None + return _literal_path(call.args[0]) + + +def _literal_path(node: ast.expr) -> str | None: + if isinstance(node, ast.Constant) and isinstance(node.value, str): + return node.value + if isinstance(node, ast.Call): + func = node.func + if isinstance(func, ast.Name) and func.id == "open" and node.args: + return _literal_path(node.args[0]) + if isinstance(func, ast.Attribute) and func.attr in {"read_text", "open"}: + inner = func.value + if ( + isinstance(inner, ast.Call) + and isinstance(inner.func, ast.Name) + and inner.func.id == "Path" + and inner.args + ): + return _literal_path(inner.args[0]) + return None + + +def _reader_names(tree: ast.Module) -> dict[str, str]: + """Map local call names to canonical config-reader dotted names. + + Resolves ``from json import load [as jload]`` and + ``import yaml as y`` (→ ``y.safe_load`` canonicalizes to + ``yaml.safe_load``) so aliased imports still count. Only the known + config modules contribute; an unrelated local ``load`` is not matched. + """ + names: dict[str, str] = {} + for node in ast.walk(tree): + if isinstance(node, ast.ImportFrom): + module = (node.module or "").split(".", 1)[0] + symbols = _CONFIG_READ_FROM_IMPORTS.get(module) + if symbols is None: + continue + for alias in node.names: + if alias.name in symbols: + names[alias.asname or alias.name] = f"{module}.{alias.name}" + elif isinstance(node, ast.Import): + for alias in node.names: + top = alias.name.split(".", 1)[0] + if top in _CONFIG_READ_FROM_IMPORTS and alias.asname: + # ``import yaml as y`` → canonicalize ``y.safe_load``. + for symbol in _CONFIG_READ_FROM_IMPORTS[top]: + names[f"{alias.asname}.{symbol}"] = f"{top}.{symbol}" + return names + + +def _settings_class_names(tree: ast.Module) -> set[str]: + """In-file class names inheriting a pydantic ``BaseSettings`` base.""" + names: set[str] = set() + for node in ast.walk(tree): + if not isinstance(node, ast.ClassDef): + continue + for base in node.bases: + base_name = dotted_name(base) or "" + if base_name.endswith(_SETTINGS_BASE_SUFFIXES): + names.add(node.name) + break + return names + + +def _scopes_for_line( + tree: ast.Module, line: int +) -> list[dict[str, ast.expr | None]]: + """Assignment maps for the scopes visible at ``line``. + + Innermost enclosing function first, then the module body. Each map is + ``name → assigned expr`` with ``None`` marking an ambiguous name + (assigned more than once in that scope). Nested function/class bodies + are excluded from their parent's scope. + """ + scopes: list[dict[str, ast.expr | None]] = [] + function = _enclosing_function(tree, line) + if function is not None: + scopes.append(_scope_assignments(function.body)) + scopes.append(_scope_assignments(tree.body)) + return scopes + + +def _enclosing_function( + tree: ast.Module, line: int +) -> ast.FunctionDef | ast.AsyncFunctionDef | None: + best: ast.FunctionDef | ast.AsyncFunctionDef | None = None + for node in ast.walk(tree): + if not isinstance(node, ast.FunctionDef | ast.AsyncFunctionDef): + continue + end = getattr(node, "end_lineno", None) + if end is None or not (node.lineno <= line <= end): + continue + if best is None or node.lineno > best.lineno: + best = node + return best + + +def _scope_assignments(body: list[ast.stmt]) -> dict[str, ast.expr | None]: + assignments: dict[str, ast.expr | None] = {} + + def record(name: str, value: ast.expr) -> None: + assignments[name] = None if name in assignments else value + + def visit(statements: list[ast.stmt]) -> None: + for stmt in statements: + if isinstance(stmt, ast.FunctionDef | ast.AsyncFunctionDef | ast.ClassDef): + continue # nested scope — not this scope's assignments + if isinstance(stmt, ast.Assign): + for target in stmt.targets: + if isinstance(target, ast.Name): + record(target.id, stmt.value) + elif isinstance(stmt, ast.AnnAssign): + if isinstance(stmt.target, ast.Name) and stmt.value is not None: + record(stmt.target.id, stmt.value) + elif isinstance(stmt, ast.With | ast.AsyncWith): + for item in stmt.items: + if isinstance(item.optional_vars, ast.Name): + record(item.optional_vars.id, item.context_expr) + for child_body in _child_bodies(stmt): + visit(child_body) + + visit(body) + return assignments + + +def _child_bodies(stmt: ast.stmt) -> list[list[ast.stmt]]: + bodies: list[list[ast.stmt]] = [] + for field in ("body", "orelse", "finalbody"): + value = getattr(stmt, field, None) + if isinstance(value, list) and value and isinstance(value[0], ast.stmt): + bodies.append(value) + for handler in getattr(stmt, "handlers", []) or []: + if handler.body: + bodies.append(handler.body) + return bodies + + +def _resolve_name( + name: str, scopes: list[dict[str, ast.expr | None]] +) -> ast.expr | None: + for scope in scopes: + if name in scope: + return scope[name] # None (ambiguous) intentionally stops the trace + return None + + +__all__ = ["ConfigBindingTrace", "trace_config_binding"] diff --git a/src/agents_shipgate/inputs/crewai.py b/src/agents_shipgate/inputs/crewai.py index 19526e75..97a8a0a7 100644 --- a/src/agents_shipgate/inputs/crewai.py +++ b/src/agents_shipgate/inputs/crewai.py @@ -22,6 +22,7 @@ unique_tools, ) from agents_shipgate.inputs.common import stable_tool_id, tool_name_warning +from agents_shipgate.inputs.config_trace import trace_config_binding from agents_shipgate.inputs.protocol import LoadedAdapterResult from agents_shipgate.inputs.python_static import ( dotted_name, @@ -248,7 +249,9 @@ def _record_agent_or_crew(self, call: ast.Call) -> None: record = {"source_ref": self.source_ref, "line": call.lineno, "tools": names or []} self.artifacts.agents.append(record) if tools_expr is not None and names is None: - self._dynamic("agent", call.lineno, dynamic_reason(tools_expr)) + self._dynamic( + "agent", call.lineno, dynamic_reason(tools_expr), expr=tools_expr + ) elif call_kind == "Crew": agents_expr = keyword(call, "agents") agents = _resolve_names(agents_expr) if agents_expr is not None else [] @@ -353,8 +356,25 @@ def _add_tool(self, variable_name: str, tool: Tool, artifact_field: str) -> None {"name": tool.name, "source_ref": self.source_ref, "line": source_line(tool.source_location)} ) - def _dynamic(self, kind: str, line: int, reason: str) -> None: - surface = {"kind": kind, "source_ref": self.source_ref, "line": line, "reason": reason} + def _dynamic( + self, kind: str, line: int, reason: str, *, expr: ast.AST | None = None + ) -> None: + surface: dict[str, Any] = { + "kind": kind, + "source_ref": self.source_ref, + "line": line, + "reason": reason, + } + # Config-bound capability detection: when the dynamic tools + # expression is statically traceable to a config read, record the + # binding on the surface fact. Purely additive — an untraceable + # expression leaves the fact byte-identical to the legacy shape. + if isinstance(expr, ast.expr): + trace = trace_config_binding(expr, tree=self.tree, line=line) + if trace.binding == "config": + surface["config_binding"] = "config" + if trace.config_path: + surface["config_path"] = trace.config_path self.artifacts.dynamic_tool_surfaces.append(surface) self.warnings.append( f"CrewAI {kind} at {self.source_ref}:{line} has dynamic tool surface: {reason}." diff --git a/src/agents_shipgate/inputs/langchain.py b/src/agents_shipgate/inputs/langchain.py index 97fd1058..189fd420 100644 --- a/src/agents_shipgate/inputs/langchain.py +++ b/src/agents_shipgate/inputs/langchain.py @@ -21,6 +21,7 @@ unique_tools, ) from agents_shipgate.inputs.common import tool_name_warning +from agents_shipgate.inputs.config_trace import trace_config_binding from agents_shipgate.inputs.protocol import LoadedAdapterResult from agents_shipgate.inputs.python_static import ( dotted_name, @@ -209,7 +210,7 @@ def _record_binding(self, kind: str, call: ast.Call, tools_expr: ast.AST | None) return names = self._resolve_tool_names(tools_expr) if names is None: - self._dynamic(kind, call.lineno, dynamic_reason(tools_expr)) + self._dynamic(kind, call.lineno, dynamic_reason(tools_expr), expr=tools_expr) return record = {"source_ref": self.source_ref, "line": call.lineno, "tools": names} if kind == "tool_node": @@ -268,8 +269,25 @@ def _add_tool(self, variable_name: str, tool: Tool, artifact_field: str) -> None {"name": tool.name, "source_ref": self.source_ref, "line": source_line(tool.source_location)} ) - def _dynamic(self, kind: str, line: int, reason: str) -> None: - surface = {"kind": kind, "source_ref": self.source_ref, "line": line, "reason": reason} + def _dynamic( + self, kind: str, line: int, reason: str, *, expr: ast.AST | None = None + ) -> None: + surface: dict[str, Any] = { + "kind": kind, + "source_ref": self.source_ref, + "line": line, + "reason": reason, + } + # Config-bound capability detection: when the dynamic tools + # expression is statically traceable to a config read, record the + # binding on the surface fact. Purely additive — an untraceable + # expression leaves the fact byte-identical to the legacy shape. + if isinstance(expr, ast.expr): + trace = trace_config_binding(expr, tree=self.tree, line=line) + if trace.binding == "config": + surface["config_binding"] = "config" + if trace.config_path: + surface["config_path"] = trace.config_path self.artifacts.dynamic_tool_surfaces.append(surface) self.warnings.append( f"LangChain {kind} at {self.source_ref}:{line} has dynamic tool surface: {reason}." diff --git a/src/agents_shipgate/inputs/openai_sdk_static.py b/src/agents_shipgate/inputs/openai_sdk_static.py index 6228e5b3..abbfb034 100644 --- a/src/agents_shipgate/inputs/openai_sdk_static.py +++ b/src/agents_shipgate/inputs/openai_sdk_static.py @@ -12,6 +12,7 @@ ) from agents_shipgate.core.errors import InputParseError from agents_shipgate.inputs.common import resolve_input_path, stable_tool_id +from agents_shipgate.inputs.config_trace import trace_config_binding from agents_shipgate.inputs.protocol import LoadedAdapterResult from agents_shipgate.inputs.python_static import ( display_path, @@ -69,9 +70,35 @@ def load_openai_sdk_static_tools( source_type="openai_agents_sdk", tools=tools, toolkit_bounds=toolkit_bounds, + warnings=_toolkit_binding_warnings(toolkit_bounds), ) +def _toolkit_binding_warnings(bounds: list[ToolkitScopeBound]) -> list[str]: + """Source warnings for toolkit factories whose binding is unreadable. + + An ``unknown`` binding means the constructor's authority-bearing + argument exists but is not statically traceable — previously that case + was silently skipped (a fail-open: the factory disappeared from every + surface). Mirror the ADK dynamic-toolset fix: record a warning so the + scan routes to review instead of silence. ``config``-bound factories + stay warning-free — they carry a comparable marker the verify-tier + config-binding checks own. + """ + return [ + ( + f"{bound.provider} toolkit constructor {bound.constructor} at " + f"{bound.source_ref}:{bound.source_line} passes a configuration " + "that is not statically readable; its effective tool surface " + "cannot be enumerated or compared. Use a literal configuration " + "allowlist, bind it from a config file, or declare an explicit " + "local tool inventory." + ) + for bound in bounds + if bound.config_binding == "unknown" + ] + + def _load_python_file( path: Path, source: ToolSourceConfig, @@ -226,7 +253,7 @@ def _detect_toolkit_bounds(tree: ast.Module, ref: str) -> list[ToolkitScopeBound if call is None: continue bound = _bound_from_toolkit_call( - call, ref, _first_assign_name(targets), name_map, module_aliases + call, tree, ref, _first_assign_name(targets), name_map, module_aliases ) if bound is not None: bounds.append(bound) @@ -235,7 +262,7 @@ def _detect_toolkit_bounds(tree: ast.Module, ref: str) -> list[ToolkitScopeBound for node in ast.walk(tree): if not isinstance(node, ast.Call) or id(node) in captured: continue - bound = _bound_from_toolkit_call(node, ref, None, name_map, module_aliases) + bound = _bound_from_toolkit_call(node, tree, ref, None, name_map, module_aliases) if bound is not None: bounds.append(bound) return bounds @@ -326,6 +353,7 @@ def _first_assign_name(targets: list[ast.expr]) -> str | None: def _bound_from_toolkit_call( call: ast.Call, + tree: ast.Module, ref: str, binding: str | None, name_map: dict[str, tuple[str, str]], @@ -337,24 +365,43 @@ def _bound_from_toolkit_call( provider, constructor = resolved parser = _CONFIG_PARSERS.get(provider) parsed = parser(call) if parser is not None else (False, []) - if parsed is None: - # ``configuration`` was supplied but is not a literal we can read - # (e.g. a variable). Skip rather than guess — the dynamic-toolkit - # path still degrades to insufficient_evidence, never a silent pass. - return None - bounded, scopes = parsed + if isinstance(parsed, tuple): + bounded, scopes = parsed + return ToolkitScopeBound( + provider=provider, + constructor=constructor, + bounded=bounded, + scopes=sorted(scopes), + binding=binding, + source_ref=ref, + source_line=getattr(call, "lineno", None), + config_binding="literal" if bounded else "absent", + ) + # The authority-bearing argument is present but not a literal. Trace it + # conservatively (docs/engineering/config-bound-capability-detection.md): + # a name bound from a config read becomes a ``config``-bound marker the + # verify diff can compare base-vs-head; anything else is ``unknown`` — a + # marker + source warning (never silence, mirroring the ADK dynamic- + # toolset fail-open fix) that no removal check ever fires on. + trace = trace_config_binding( + parsed, tree=tree, line=getattr(call, "lineno", 0) + ) return ToolkitScopeBound( provider=provider, constructor=constructor, - bounded=bounded, - scopes=sorted(scopes), + bounded=False, + scopes=[], binding=binding, source_ref=ref, source_line=getattr(call, "lineno", None), + config_binding=trace.binding, + config_path=trace.config_path, ) -def _parse_stripe_configuration(call: ast.Call) -> tuple[bool, list[str]] | None: +def _parse_stripe_configuration( + call: ast.Call, +) -> tuple[bool, list[str]] | ast.expr: """Read the Stripe ``configuration={"actions": {...}}`` allowlist. Returns ``(bounded, scopes)``: @@ -366,7 +413,8 @@ def _parse_stripe_configuration(call: ast.Call) -> tuple[bool, list[str]] | None ``configuration={"context": …}`` migration would read as a narrowing and escape the gate. - ``configuration`` (or its ``actions``) present but not a dict literal → - ``None`` (ambiguous; the caller skips the bound). + the offending *expression* (the authority-bearing argument), which the + caller traces for a config binding instead of silently skipping. - ``actions`` is a dict literal → ``(True, scopes)`` with each truthy ``resource:verb`` flattened. An explicit empty ``actions={}`` is ``(True, [])`` — explicitly bounded to no tools. @@ -375,12 +423,12 @@ def _parse_stripe_configuration(call: ast.Call) -> tuple[bool, list[str]] | None if config is None: return (False, []) if not isinstance(config, ast.Dict): - return None + return config actions = _dict_get(config, "actions") if actions is None: return (False, []) if not isinstance(actions, ast.Dict): - return None + return actions scopes: list[str] = [] for resource_key, resource_value in zip(actions.keys, actions.values, strict=False): resource = _const_str(resource_key) diff --git a/tests/fixtures/config_bound_factory/base/config/stripe_actions.json b/tests/fixtures/config_bound_factory/base/config/stripe_actions.json new file mode 100644 index 00000000..9721c941 --- /dev/null +++ b/tests/fixtures/config_bound_factory/base/config/stripe_actions.json @@ -0,0 +1,7 @@ +{ + "actions": { + "customers": {"read": true}, + "invoices": {"read": true}, + "billing_portal_sessions": {"create": true} + } +} diff --git a/tests/fixtures/config_bound_factory/base/shipgate.yaml b/tests/fixtures/config_bound_factory/base/shipgate.yaml new file mode 100644 index 00000000..42475a93 --- /dev/null +++ b/tests/fixtures/config_bound_factory/base/shipgate.yaml @@ -0,0 +1,43 @@ +# Config-bound dynamic-toolkit fixture, BASE side: the Stripe toolkit's +# authority allowlist is bound from config/stripe_actions.json (a config +# read), so the effective tool surface is configuration-defined and not +# statically enumerable. See docs/engineering/config-bound-capability-detection.md. +version: "0.1" + +project: + name: config-bound-email-agent + +agent: + name: "Standup Jack Agent" + declared_purpose: + - Answer customer FAQ questions + - Update billing information through the Stripe Customer Portal + - Send customers their missing invoices + prohibited_actions: + - Issue a refund without human approval + - Cancel a subscription without explicit confirmation + - Update or close a payment dispute without review + +environment: + target: production_like + +tool_sources: + - id: openai_sdk_support_agent + type: openai_agents_sdk + path: support_agent.py + +policies: + require_approval_for_tools: [] + require_confirmation_for_tools: [] + require_idempotency_for_tools: [] + +permissions: + scopes: [] + +ci: + mode: advisory + +output: + directory: agents-shipgate-reports + formats: + - json diff --git a/tests/fixtures/config_bound_factory/base/support_agent.py b/tests/fixtures/config_bound_factory/base/support_agent.py new file mode 100644 index 00000000..09b8af36 --- /dev/null +++ b/tests/fixtures/config_bound_factory/base/support_agent.py @@ -0,0 +1,52 @@ +import json + +import env +import requests +from agents import Agent, Runner, RunResult, TResponseInputItem, function_tool +from stripe_agent_toolkit.openai.toolkit import StripeAgentToolkit + +env.ensure("OPENAI_API_KEY") + +# The toolkit's authority allowlist is bound from configuration: the +# effective tool surface is defined by config/stripe_actions.json, not by +# code. Static analysis can see the *binding* but not the resulting tools. +_CONFIGURATION = json.load(open("config/stripe_actions.json")) + +# Global toolkit instance - initialized in init() +stripe_agent_toolkit = None +support_agent = None + + +@function_tool +def search_faq(question: str) -> str: + response = requests.get("https://standupjack.com/faq") + if response.status_code != 200: + return "Not sure" + return f"Given the following context:\n{response.text}\n\nAnswer '{question}' or response with not sure\n" + + +async def init(): + """Initialize the toolkit and agent. Must be called before run().""" + global stripe_agent_toolkit, support_agent + + stripe_agent_toolkit = StripeAgentToolkit( + secret_key=env.ensure("STRIPE_SECRET_KEY"), + configuration=_CONFIGURATION, + ) + + support_agent = Agent( + name="Standup Jack Agent", + instructions=( + "You are a helpful customer support assistant" + "Be casual and concise" + "Use tools to support customers" + "Sign off with Standup Jack Bot" + ), + tools=[search_faq, *stripe_agent_toolkit.get_tools()], + ) + + +async def run(input: list[TResponseInputItem]) -> RunResult: + if support_agent is None: + raise RuntimeError("support_agent not initialized. Call init() first.") + return await Runner.run(support_agent, input) diff --git a/tests/fixtures/config_bound_factory/head/config/stripe_actions.json b/tests/fixtures/config_bound_factory/head/config/stripe_actions.json new file mode 100644 index 00000000..9721c941 --- /dev/null +++ b/tests/fixtures/config_bound_factory/head/config/stripe_actions.json @@ -0,0 +1,7 @@ +{ + "actions": { + "customers": {"read": true}, + "invoices": {"read": true}, + "billing_portal_sessions": {"create": true} + } +} diff --git a/tests/fixtures/config_bound_factory/head/shipgate.yaml b/tests/fixtures/config_bound_factory/head/shipgate.yaml new file mode 100644 index 00000000..fb5cb9bd --- /dev/null +++ b/tests/fixtures/config_bound_factory/head/shipgate.yaml @@ -0,0 +1,43 @@ +# Config-bound dynamic-toolkit fixture, HEAD side: the PR removed the +# config binding from the factory call, so the Stripe toolkit falls back +# to its everything-enabled defaults while the diff looks like cleanup. +# See docs/engineering/config-bound-capability-detection.md. +version: "0.1" + +project: + name: config-bound-email-agent + +agent: + name: "Standup Jack Agent" + declared_purpose: + - Answer customer FAQ questions + - Update billing information through the Stripe Customer Portal + - Send customers their missing invoices + prohibited_actions: + - Issue a refund without human approval + - Cancel a subscription without explicit confirmation + - Update or close a payment dispute without review + +environment: + target: production_like + +tool_sources: + - id: openai_sdk_support_agent + type: openai_agents_sdk + path: support_agent.py + +policies: + require_approval_for_tools: [] + require_confirmation_for_tools: [] + require_idempotency_for_tools: [] + +permissions: + scopes: [] + +ci: + mode: advisory + +output: + directory: agents-shipgate-reports + formats: + - json diff --git a/tests/fixtures/config_bound_factory/head/support_agent.py b/tests/fixtures/config_bound_factory/head/support_agent.py new file mode 100644 index 00000000..28bf2cdd --- /dev/null +++ b/tests/fixtures/config_bound_factory/head/support_agent.py @@ -0,0 +1,46 @@ +import env +import requests +from agents import Agent, Runner, RunResult, TResponseInputItem, function_tool +from stripe_agent_toolkit.openai.toolkit import StripeAgentToolkit + +env.ensure("OPENAI_API_KEY") + +# Global toolkit instance - initialized in init() +stripe_agent_toolkit = None +support_agent = None + + +@function_tool +def search_faq(question: str) -> str: + response = requests.get("https://standupjack.com/faq") + if response.status_code != 200: + return "Not sure" + return f"Given the following context:\n{response.text}\n\nAnswer '{question}' or response with not sure\n" + + +async def init(): + """Initialize the toolkit and agent. Must be called before run().""" + global stripe_agent_toolkit, support_agent + + # The config binding was removed: no configuration argument means the + # toolkit mounts its everything-enabled defaults. + stripe_agent_toolkit = StripeAgentToolkit( + secret_key=env.ensure("STRIPE_SECRET_KEY"), + ) + + support_agent = Agent( + name="Standup Jack Agent", + instructions=( + "You are a helpful customer support assistant" + "Be casual and concise" + "Use tools to support customers" + "Sign off with Standup Jack Bot" + ), + tools=[search_faq, *stripe_agent_toolkit.get_tools()], + ) + + +async def run(input: list[TResponseInputItem]) -> RunResult: + if support_agent is None: + raise RuntimeError("support_agent not initialized. Call init() first.") + return await Runner.run(support_agent, input) diff --git a/tests/test_action_surface_diff.py b/tests/test_action_surface_diff.py index 78029900..efd806de 100644 --- a/tests/test_action_surface_diff.py +++ b/tests/test_action_surface_diff.py @@ -355,6 +355,48 @@ def test_action_surface_disambiguates_openapi_action_id_collisions_fail_soft(): assert "get_session" in warnings[0] and "get_session_detail" in warnings[0] +def test_action_surface_diff_degrades_duplicate_base_action_ids_fail_soft(): + """A ``--diff-from`` report or baseline serialized by a pre-collision-fix + engine can carry duplicate ``action_id`` values in its round-tripped + ``action_surface_facts``. The diff must degrade fail-soft with a warning + (routing to review_required) instead of crashing the scan with a + ConfigError that points at a manifest which is fine — the engine + inferred those ids itself. Without a sink the legacy hard error is + preserved. Real-world repro: block/goose miner rows dying with exit 2.""" + base_id = "agent:action-test/agent:openapi:goose-api:GET /sessions/{session_id}" + base_first = _action(base_id) + base_first.tool_name = "get_session" + base_second = _action(base_id) + base_second.tool_name = "get_session_detail" + base = ActionSurfaceFacts(actions=[base_first, base_second]) + # Fresh head facts: build-time disambiguation already suffixed the second. + head_first = _action(base_id) + head_first.tool_name = "get_session" + head_second = _action(f"{base_id}#get_session_detail") + head_second.tool_name = "get_session_detail" + current = ActionSurfaceFacts(actions=[head_first, head_second]) + + # Legacy no-sink callers keep the hard ConfigError. + with pytest.raises(ConfigError, match="Duplicate action_surface action_id"): + compute_action_surface_diff(current, base) + + warnings: list[str] = [] + diff = compute_action_surface_diff(current, base, warnings=warnings) + + assert diff.enabled + # The base side was disambiguated with the same tool-name suffix strategy + # the head build uses, so an identical surface diffs without churn. + assert diff.added == [] + assert diff.removed == [] + assert {action.action_id for action in base.actions} == { + base_id, + f"{base_id}#get_session_detail", + } + assert len(warnings) == 1 + assert "base reference" in warnings[0] + assert "Duplicate action_surface action_id" in warnings[0] + + def test_action_surface_policy_requires_typed_expected_values(): with pytest.raises(ValueError, match="safeguards.audit_log"): _manifest( diff --git a/tests/test_scan.py b/tests/test_scan.py index e7e058cd..9e2bfec9 100644 --- a/tests/test_scan.py +++ b/tests/test_scan.py @@ -1421,3 +1421,104 @@ def test_openapi_action_id_collision_degrades_instead_of_crashing(tmp_path): "Duplicate action_surface action_id" in warning for warning in report.source_warnings ), f"Expected an action_id collision warning; got: {report.source_warnings}" + + +def test_diff_from_report_with_duplicate_base_action_ids_degrades(tmp_path): + """A ``--diff-from`` base report whose round-tripped + ``action_surface_facts`` carry duplicate ``action_id`` values (written + by a pre-collision-fix engine) must not crash the scan with a hard + Config error pointing at a manifest that is fine. Same fail-soft + principle as the build-time collision fix (#226): degrade to a + source_warning, keep the diff enabled.""" + project = tmp_path / "project" + project.mkdir() + (project / "openapi.yaml").write_text( + """ +openapi: 3.1.0 +info: + title: Goose-like API + version: "1.0" +paths: + /sessions/{session_id}: + get: + operationId: get_session + summary: Get a session. + parameters: + - name: session_id + in: path + required: true + schema: {type: string} + responses: + "200": + description: ok + /sessions/{session_id}/: + get: + operationId: get_session_detail + summary: Get a session detail. + parameters: + - name: session_id + in: path + required: true + schema: {type: string} + responses: + "200": + description: ok +""", + encoding="utf-8", + ) + (project / "shipgate.yaml").write_text( + """ +version: "0.1" +project: + name: goose-collision +agent: + name: goose-agent + declared_purpose: + - test duplicate base action ids +environment: + target: local +tool_sources: + - id: goose-api + type: openapi + path: openapi.yaml +""", + encoding="utf-8", + ) + + _, exit_code = run_scan( + config_path=project / "shipgate.yaml", + output_dir=tmp_path / "base-reports", + formats=["json"], + ci_mode="advisory", + packet_enabled=False, + ) + assert exit_code == 0 + base_report_path = tmp_path / "base-reports" / "report.json" + payload = json.loads(base_report_path.read_text(encoding="utf-8")) + actions = payload["action_surface_facts"]["actions"] + assert len(actions) == 2 + # Simulate the pre-fix serialization: both operations collapsed onto + # the bare collided id. + bare_id = min((action["action_id"] for action in actions), key=len) + for action in actions: + action["action_id"] = bare_id + stale_base_path = tmp_path / "stale-base-report.json" + stale_base_path.write_text(json.dumps(payload), encoding="utf-8") + + # Must not raise ConfigError — the scan completes and returns a report. + report, exit_code = run_scan( + config_path=project / "shipgate.yaml", + output_dir=tmp_path / "head-reports", + formats=["json"], + ci_mode="advisory", + packet_enabled=False, + diff_from_path=stale_base_path, + ) + + assert exit_code == 0 + assert report.action_surface_diff.enabled + assert any( + "base reference" in warning + and "Duplicate action_surface action_id" in warning + for warning in report.source_warnings + ), f"Expected a base-side collision warning; got: {report.source_warnings}" diff --git a/tests/test_verify_capability_scope.py b/tests/test_verify_capability_scope.py index 6eb46104..1629cfbc 100644 --- a/tests/test_verify_capability_scope.py +++ b/tests/test_verify_capability_scope.py @@ -90,15 +90,20 @@ def test_extractor_drops_only_false_actions(): assert bound.scopes == ["customers:read"] -def test_extractor_skips_non_literal_configuration(): - # A configuration passed as a variable is ambiguous; emit no bound rather - # than guess (the dynamic-toolkit path still degrades to low confidence). +def test_extractor_marks_non_literal_configuration_unknown(): + # A configuration passed as a variable the tracer cannot resolve is + # ambiguous; emit an ``unknown`` marker (never silence — the previous + # skip was a fail-open) but never claim it is bounded or unbounded. + # The removal check must never fire on an unknown binding. src = ( "from stripe_agent_toolkit.openai.toolkit import StripeAgentToolkit\n" "cfg = load_cfg()\n" "tk = StripeAgentToolkit(configuration=cfg)\n" ) - assert _bounds_from_src(src) == [] + [bound] = _bounds_from_src(src) + assert bound.bounded is False + assert bound.config_binding == "unknown" + assert bound.config_path is None def test_extractor_ignores_unknown_constructor(): diff --git a/tests/test_verify_config_binding.py b/tests/test_verify_config_binding.py new file mode 100644 index 00000000..9c74f4c6 --- /dev/null +++ b/tests/test_verify_config_binding.py @@ -0,0 +1,547 @@ +"""SHIP-CAP-CONFIG-BINDING-* — config-bound dynamic-toolkit authority. + +Covers the detection half of +docs/engineering/config-bound-capability-detection.md, reproducing the +2026-06 design-partner pilot blind spot: a toolkit factory whose authority +allowlist is *bound from config* hides its effective tool surface from both +sides of a verify diff, so removing the binding (authority expansion) or +editing the bound config produced no signal at all. + +Layers exercised: the conservative config-read tracer, the extractor +(config-bound / unknown markers), the carriage codec (policy-fact round +trip incl. legacy hash stability), the check (base-vs-head classification), +end-to-end base→head verdicts on the ``config_bound_factory`` fixture, and +the fix_task projection carrying the inventory remedy. +""" + +from __future__ import annotations + +import ast +import hashlib +import json +from pathlib import Path + +from agents_shipgate.checks import toolkit_bounds, verify_config_binding +from agents_shipgate.cli.scan import run_scan +from agents_shipgate.cli.verify.fix_task import build_fix_task +from agents_shipgate.config.loader import load_manifest +from agents_shipgate.core.context import ScanContext +from agents_shipgate.core.domain import Agent, ToolkitScopeBound +from agents_shipgate.core.lenses.tool_surface import ToolSurfaceDiffReference +from agents_shipgate.core.toolkit_scope import ( + TOOLKIT_BOUND_POLICY_KIND, + bound_from_policy_fact, + bound_to_policy_fact, + toolkit_bound_facts, +) +from agents_shipgate.inputs.config_trace import trace_config_binding +from agents_shipgate.inputs.openai_sdk_static import _detect_toolkit_bounds +from agents_shipgate.schemas.surfaces import ToolSurfaceFacts +from agents_shipgate.schemas.verification import VerificationContext +from agents_shipgate.schemas.verifier import VerifierCapabilityReview + +FIXTURES = Path(__file__).parent / "fixtures" / "config_bound_factory" +CHECK_REMOVED = "SHIP-CAP-CONFIG-BINDING-REMOVED" +CHECK_CHANGED = "SHIP-CAP-CONFIG-BINDING-CHANGED" + + +# --- tracer ------------------------------------------------------------------ + + +def _trace(src: str, expr_src: str, line: int | None = None): + tree = ast.parse(src) + expr = ast.parse(expr_src, mode="eval").body + return trace_config_binding( + expr, tree=tree, line=line if line is not None else len(src.splitlines()) + ) + + +def test_trace_json_load_open_literal_path(): + trace = _trace('import json\ncfg = json.load(open("config/tools.json"))\n', "cfg") + assert trace.binding == "config" + assert trace.config_path == "config/tools.json" + + +def test_trace_yaml_module_alias(): + trace = _trace('import yaml as y\ncfg = y.safe_load(open("cfg.yaml"))\n', "cfg") + assert trace.binding == "config" + assert trace.config_path == "cfg.yaml" + + +def test_trace_from_import_alias(): + trace = _trace('from json import load as jload\ncfg = jload(open("a.json"))\n', "cfg") + assert trace.binding == "config" + assert trace.config_path == "a.json" + + +def test_trace_toml_literal_path_argument(): + trace = _trace('import toml\ncfg = toml.load("pyproject.toml")\n', "cfg") + assert trace.binding == "config" + assert trace.config_path == "pyproject.toml" + + +def test_trace_pathlib_read_text(): + src = ( + "import json\nfrom pathlib import Path\n" + 'cfg = json.loads(Path("config/actions.json").read_text())\n' + ) + trace = _trace(src, "cfg") + assert trace.binding == "config" + assert trace.config_path == "config/actions.json" + + +def test_trace_os_environ_subscript_and_getenv(): + assert _trace("import os\n", 'os.environ["ACTIONS"]').binding == "config" + getenv = _trace('import os\ncfg = os.getenv("ACTIONS")\n', "cfg") + assert getenv.binding == "config" + assert getenv.config_path is None + + +def test_trace_pydantic_settings_subclass(): + src = ( + "from pydantic_settings import BaseSettings\n" + "class ToolSettings(BaseSettings):\n" + " actions: dict = {}\n" + "settings = ToolSettings()\n" + ) + assert _trace(src, "settings").binding == "config" + # Attribute access on the settings object carries the same binding. + assert _trace(src, "settings.actions").binding == "config" + + +def test_trace_subscript_and_get_on_traced_name(): + src = 'import json\ncfg = json.load(open("cfg.json"))\n' + assert _trace(src, 'cfg["actions"]').binding == "config" + assert _trace(src, 'cfg.get("actions")').binding == "config" + + +def test_trace_unresolvable_name_is_unknown(): + assert _trace("x = 1\n", "somewhere_else").binding == "unknown" + + +def test_trace_non_config_call_is_unknown(): + assert _trace("cfg = build_config()\n", "cfg").binding == "unknown" + + +def test_trace_reassigned_name_is_ambiguous_unknown(): + # Same-scope reassignment is ambiguous; the conservative tracer must not + # pick a winner (the design doc's false-positive guard). + src = ( + "import json\n" + 'cfg = json.load(open("a.json"))\n' + "cfg = build_config()\n" + ) + assert _trace(src, "cfg").binding == "unknown" + + +def test_trace_function_scope_falls_back_to_module_scope(): + src = ( + "import json\n" + '_CONF = json.load(open("config/actions.json"))\n' + "def init():\n" + " tk = make(configuration=_CONF)\n" + ) + trace = _trace(src, "_CONF", line=4) + assert trace.binding == "config" + assert trace.config_path == "config/actions.json" + + +def test_trace_local_shadow_wins_over_module_scope(): + src = ( + "import json\n" + '_CONF = json.load(open("config/actions.json"))\n' + "def init():\n" + " _CONF = build_config()\n" + " tk = make(configuration=_CONF)\n" + ) + assert _trace(src, "_CONF", line=5).binding == "unknown" + + +# --- extractor --------------------------------------------------------------- + + +def _bounds_from_src(src: str) -> list[ToolkitScopeBound]: + return _detect_toolkit_bounds(ast.parse(src), "support_agent.py") + + +def test_extractor_marks_config_bound_factory(): + src = ( + "import json\n" + "from stripe_agent_toolkit.openai.toolkit import StripeAgentToolkit\n" + '_CONF = json.load(open("config/stripe_actions.json"))\n' + "tk = StripeAgentToolkit(configuration=_CONF)\n" + ) + [bound] = _bounds_from_src(src) + assert bound.bounded is False + assert bound.config_binding == "config" + assert bound.config_path == "config/stripe_actions.json" + assert bound.binding == "tk" + + +def test_extractor_traces_non_literal_actions_value(): + # configuration is a dict literal but the authority-bearing `actions` + # value is the traced name. + src = ( + "import yaml\n" + "from stripe_agent_toolkit.openai.toolkit import StripeAgentToolkit\n" + '_ACTIONS = yaml.safe_load(open("actions.yaml"))\n' + "tk = StripeAgentToolkit(configuration={'actions': _ACTIONS})\n" + ) + [bound] = _bounds_from_src(src) + assert bound.config_binding == "config" + assert bound.config_path == "actions.yaml" + + +def test_extractor_literal_and_absent_bindings_are_labeled(): + literal = _bounds_from_src( + "from stripe_agent_toolkit.openai.toolkit import StripeAgentToolkit\n" + "tk = StripeAgentToolkit(configuration={'actions': {'customers': {'read': True}}})\n" + ) + assert literal[0].config_binding == "literal" + absent = _bounds_from_src( + "from stripe_agent_toolkit.openai.toolkit import create_stripe_agent_toolkit\n" + "tk = create_stripe_agent_toolkit(secret_key='x')\n" + ) + assert absent[0].config_binding == "absent" + + +# --- carriage codec ------------------------------------------------------------ + + +def _bound( + *, + config_binding=None, + config_path=None, + bounded=False, + scopes=(), + binding="stripe_agent_toolkit", +): + return ToolkitScopeBound( + provider="stripe", + constructor="StripeAgentToolkit", + bounded=bounded, + scopes=sorted(scopes), + binding=binding, + source_ref="support_agent.py", + source_line=32, + config_binding=config_binding, + config_path=config_path, + ) + + +def test_policy_fact_round_trip_config_bound_with_path(): + fact = bound_to_policy_fact( + _bound(config_binding="config", config_path="config/stripe_actions.json") + ) + assert fact.kind == TOOLKIT_BOUND_POLICY_KIND + assert "config-bound" in (fact.summary or "") + decoded = bound_from_policy_fact(fact) + assert decoded.config_binding == "config" + assert decoded.config_path == "config/stripe_actions.json" + assert decoded.bounded is False + + +def test_policy_fact_round_trip_config_bound_without_path(): + decoded = bound_from_policy_fact( + bound_to_policy_fact(_bound(config_binding="config")) + ) + assert decoded.config_binding == "config" + assert decoded.config_path is None + + +def test_policy_fact_round_trip_unknown_binding(): + decoded = bound_from_policy_fact( + bound_to_policy_fact(_bound(config_binding="unknown")) + ) + assert decoded.config_binding == "unknown" + assert decoded.bounded is False + + +def test_legacy_value_hash_unchanged_for_literal_and_absent(): + # The serialized value_hash of literal/absent bounds must stay + # byte-identical to the pre-feature formula, or every existing base + # report's toolkit facts would churn on upgrade. + def legacy_hash(bounded: bool, scopes: list[str]) -> str: + payload = json.dumps( + {"bounded": bounded, "scopes": sorted(scopes)}, + sort_keys=True, + separators=(",", ":"), + ) + return hashlib.sha256(payload.encode("utf-8")).hexdigest()[:16] + + literal = _bound( + config_binding="literal", bounded=True, scopes=["customers:read"] + ) + assert bound_to_policy_fact(literal).value_hash == legacy_hash( + True, ["customers:read"] + ) + absent = _bound(config_binding="absent") + assert bound_to_policy_fact(absent).value_hash == legacy_hash(False, []) + + +# --- check: base-vs-head classification --------------------------------------- + + +def _cfg(side: str) -> Path: + return FIXTURES / side / "shipgate.yaml" + + +def _ctx( + *, + head_bounds=(), + base_bounds=None, + verification=True, + changed=("support_agent.py",), +) -> ScanContext: + vc = VerificationContext(changed_files=list(changed)) if verification else None + diff_reference = None + if base_bounds is not None: + facts = ToolSurfaceFacts(policies=toolkit_bound_facts(list(base_bounds))) + diff_reference = ToolSurfaceDiffReference(kind="report", facts=facts) + return ScanContext( + manifest=load_manifest(_cfg("head")), + agent=Agent(id="agent:test/test", name="test"), + tools=[], + config_path=_cfg("head"), + verification=vc, + diff_reference=diff_reference, + toolkit_bounds=list(head_bounds), + ) + + +_CONFIG_BASE = dict(config_binding="config", config_path="config/stripe_actions.json") + + +def test_binding_removed_emits_high_finding(): + ctx = _ctx( + base_bounds=[_bound(**_CONFIG_BASE)], + head_bounds=[_bound(config_binding="absent")], + ) + [finding] = verify_config_binding.run(ctx) + assert finding.check_id == CHECK_REMOVED + assert finding.severity == "high" + assert finding.category == "verify" + assert finding.blocks_release is False + assert finding.evidence["kind"] == "config_binding_removed" + assert finding.evidence["base_config_path"] == "config/stripe_actions.json" + # The remedy carries the inventory remedy with the factory's site. + assert "support_agent.py:32" in finding.recommendation + assert "tool inventory" in finding.recommendation + + +def test_unknown_head_binding_never_fires_removal(): + # The design doc's calibration guard: an unreadable binding must not be + # read as a removal (it degrades to the extractor's source warning). + ctx = _ctx( + base_bounds=[_bound(**_CONFIG_BASE)], + head_bounds=[_bound(config_binding="unknown")], + ) + assert verify_config_binding.run(ctx) == [] + + +def test_head_moving_to_literal_allowlist_emits_nothing(): + ctx = _ctx( + base_bounds=[_bound(**_CONFIG_BASE)], + head_bounds=[ + _bound(config_binding="literal", bounded=True, scopes=["customers:read"]) + ], + ) + assert verify_config_binding.run(ctx) == [] + + +def test_config_file_changed_emits_review_item(): + ctx = _ctx( + base_bounds=[_bound(**_CONFIG_BASE)], + head_bounds=[_bound(**_CONFIG_BASE)], + changed=("config/stripe_actions.json",), + ) + [finding] = verify_config_binding.run(ctx) + assert finding.check_id == CHECK_CHANGED + assert finding.severity == "medium" + assert finding.evidence["kind"] == "config_binding_changed" + assert finding.evidence["changed_file"] == "config/stripe_actions.json" + + +def test_config_file_untouched_emits_nothing(): + ctx = _ctx( + base_bounds=[_bound(**_CONFIG_BASE)], + head_bounds=[_bound(**_CONFIG_BASE)], + changed=("README.md",), + ) + assert verify_config_binding.run(ctx) == [] + + +def test_config_binding_retarget_fires_without_config_in_changed_files(): + # Review-reproduced blind spot: the PR edits only the .py file, repointing + # the binding from safe.json to a PRE-EXISTING broad.json. Neither config + # file appears in changed_files, yet the effective tool surface may have + # expanded — the retarget itself must be the review item. + base = dict(_CONFIG_BASE) + head = dict(_CONFIG_BASE) + base["config_path"] = "config/safe.json" + head["config_path"] = "config/broad.json" + ctx = _ctx( + base_bounds=[_bound(**base)], + head_bounds=[_bound(**head)], + changed=("support_agent.py",), + ) + [finding] = verify_config_binding.run(ctx) + assert finding.check_id == CHECK_CHANGED + assert finding.severity == "medium" + assert finding.evidence["kind"] == "config_binding_retargeted" + assert finding.evidence["previous_config_path"] == "config/safe.json" + assert finding.evidence["config_path"] == "config/broad.json" + assert "config/safe.json → config/broad.json" in finding.recommendation + + +def test_pathless_config_binding_never_guesses_a_match(): + # Env/settings-bound factories carry no literal path; a changed file must + # never be matched by guesswork. + ctx = _ctx( + base_bounds=[_bound(config_binding="config")], + head_bounds=[_bound(config_binding="config")], + changed=("config/stripe_actions.json",), + ) + assert verify_config_binding.run(ctx) == [] + + +def test_literal_base_binding_is_not_this_checks_concern(): + # literal → absent is SHIP-VERIFY-CAPABILITY-SCOPE-BROADENED's territory. + ctx = _ctx( + base_bounds=[ + _bound(config_binding="literal", bounded=True, scopes=["customers:read"]) + ], + head_bounds=[_bound(config_binding="absent")], + ) + assert verify_config_binding.run(ctx) == [] + + +def test_no_verification_context_emits_nothing(): + ctx = _ctx( + base_bounds=[_bound(**_CONFIG_BASE)], + head_bounds=[_bound(config_binding="absent")], + verification=False, + ) + assert verify_config_binding.run(ctx) == [] + + +def test_no_base_reference_emits_nothing(): + ctx = _ctx(base_bounds=None, head_bounds=[_bound(config_binding="absent")]) + assert verify_config_binding.run(ctx) == [] + + +def test_toolkit_unbounded_check_skips_config_bound_factories(): + # "Mounted without a scope bound" would be a false claim for a factory + # whose configuration lives in config (or is unreadable). + ctx = _ctx( + head_bounds=[ + _bound(**_CONFIG_BASE), + _bound(config_binding="unknown", binding="other_tk"), + ] + ) + assert toolkit_bounds.run(ctx) == [] + ctx_absent = _ctx(head_bounds=[_bound(config_binding="absent")]) + [finding] = toolkit_bounds.run(ctx_absent) + assert finding.check_id == "SHIP-SCOPE-TOOLKIT-UNBOUNDED" + + +# --- end-to-end: the pilot scenario ------------------------------------------- + + +def test_base_report_carries_config_bound_policy_fact(tmp_path): + base_report, _ = run_scan( + config_path=_cfg("base"), + output_dir=tmp_path / "base", + formats=["json"], + ci_mode="advisory", + packet_enabled=False, + ) + toolkit_facts = [ + p + for p in base_report.tool_surface_facts.policies + if p.kind == TOOLKIT_BOUND_POLICY_KIND + ] + assert [p.key for p in toolkit_facts] == ["stripe:stripe_agent_toolkit"] + assert toolkit_facts[0].summary == "(config-bound: config/stripe_actions.json)" + + +def test_config_binding_removal_routes_to_review(tmp_path): + base_report, _ = run_scan( + config_path=_cfg("base"), + output_dir=tmp_path / "base", + formats=["json"], + ci_mode="advisory", + packet_enabled=False, + ) + # Base in isolation cannot enumerate the toolkit tools -> insufficient. + assert base_report.release_decision.decision == "insufficient_evidence" + + head_report, _ = run_scan( + config_path=_cfg("head"), + output_dir=tmp_path / "head", + formats=["json"], + ci_mode="advisory", + diff_from_path=tmp_path / "base" / "report.json", + verification_context=VerificationContext(changed_files=["support_agent.py"]), + packet_enabled=False, + ) + # The statically-visible binding removal lifts the verdict out of silent + # insufficient_evidence parity. + assert head_report.release_decision.decision in {"blocked", "review_required"} + gating = { + item.check_id + for item in [ + *head_report.release_decision.blockers, + *head_report.release_decision.review_items, + ] + } + assert CHECK_REMOVED in gating + [finding] = [f for f in head_report.findings if f.check_id == CHECK_REMOVED] + assert finding.severity == "high" + assert finding.evidence["provider"] == "stripe" + assert finding.evidence["base_config_path"] == "config/stripe_actions.json" + + # Acceptance: fix_task for the new finding carries the inventory remedy + # with the factory's source path and line. + fix_task = build_fix_task( + head_report, + merge_verdict="human_review_required", + capability_review=VerifierCapabilityReview(), + base_ref="origin/main", + head_ref="HEAD", + ) + assert fix_task is not None + assert fix_task.actor == "human" + combined = "\n".join( + [*fix_task.instructions, *[r.reason or "" for r in fix_task.allowed_repairs]] + ) + assert "tool inventory" in combined + assert "support_agent.py:" in combined + + +def test_config_delta_routes_to_review_item(tmp_path): + # Same tree on both sides; the PR touches only the bound config file. + base_report, _ = run_scan( + config_path=_cfg("base"), + output_dir=tmp_path / "base", + formats=["json"], + ci_mode="advisory", + packet_enabled=False, + ) + head_report, _ = run_scan( + config_path=_cfg("base"), + output_dir=tmp_path / "head", + formats=["json"], + ci_mode="advisory", + diff_from_path=tmp_path / "base" / "report.json", + verification_context=VerificationContext( + changed_files=["config/stripe_actions.json"] + ), + packet_enabled=False, + ) + [finding] = [f for f in head_report.findings if f.check_id == CHECK_CHANGED] + assert finding.evidence["changed_file"] == "config/stripe_actions.json" + review_ids = { + item.check_id for item in head_report.release_decision.review_items + } + assert CHECK_CHANGED in review_ids