From 319b20972dc923b43ba09ca5e9c55b77af237db0 Mon Sep 17 00:00:00 2001 From: Claude Date: Thu, 20 Aug 2026 10:07:28 +0000 Subject: [PATCH 1/2] refactor(scripts): one constant-test loop, not two MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit #324 and #325 landed equivalent loops within an hour of each other, built in parallel against the same brief. main now carries constant_test_loop.py and scoped_smoke_loop.py, which assert overlapping invariants against the same endpoints and document themselves separately. Two monitors that drift apart is worse than either alone, because nobody can say which one going red is the one that matters. constant_test_loop.py survives — it shipped first, has docs, enforces HTTPS off loopback, and self-provisions on loopback. scoped_smoke_loop.py and its tests are removed, with its distinct parts folded in: Tool selection no longer dies when neither partner.echo nor partner.notes.write is registered. --tool (or $CI_SMOKE_TOOL) names one instead, and the failure message lists what the deployment actually exposes. There is deliberately no "just use the first one" fallback: schema-derived arguments satisfy a tool's declared shape but not its semantics — verified locally, awi_passkey_challenge accepts the payload and then refuses the action — and that failure reads as a broken trust plane when it is really a bad choice of tool. Arguments come from the selected tool's inputSchema rather than a hardcoded {"text": ...}, so the loop runs against a registry it was not written for. The replay still reuses the identical body, so it stays a replay. The permit cap is derived from the tool's advertised creditsPerCall with headroom rather than fixed at 10. Tools here are priced 1 to 200 credits, so a fixed cap fails permit_budget_exceeded on a healthy deployment as soon as a pricier tool is selected, which reads as a product fault. Off loopback, --tool-args is required. Derivation fills required fields from types, defaults, and the first enum member, which for a consequential tool could be "delete", sent again every run. Schema validity is not evidence of safety. --tool-args '{}' is a fine answer; it just has to be an answer. The companion tool for the out-of-scope check is selected through select_companion_tool, which requires it to be registered and to differ from the permitted tool, so the check cannot pass by proving tool-not-found. Also drops OUT_OF_SCOPE_TOOL, a constant referenced nowhere — the denial check already picked a real tool from the registry. Verified against a local instance exposing neither preferred tool: the selection error names the available tools, and with --tool semantic-search the full loop reports ALL INVARIANTS HELD at exit 0, charging 15 credits where the old fixed cap of 10 would have refused the call. Co-Authored-By: Claude Opus 5 Claude-Session: https://claude.ai/code/session_01Ri576RW3ocgqpx1xYN9vBm --- Makefile | 3 - docs/PROOF_MATRIX.md | 24 +- docs/constant-test-loop.md | 28 +- scripts/constant_test_loop.py | 250 +++++++++++-- scripts/scoped_smoke_loop.py | 612 ------------------------------- tests/test_constant_test_loop.py | 223 +++++++++++ tests/test_scoped_smoke_loop.py | 328 ----------------- 7 files changed, 489 insertions(+), 979 deletions(-) delete mode 100644 scripts/scoped_smoke_loop.py delete mode 100644 tests/test_scoped_smoke_loop.py diff --git a/Makefile b/Makefile index 20b9c1b7..09098fd0 100644 --- a/Makefile +++ b/Makefile @@ -129,9 +129,6 @@ trust-conformance-live: adversarial-battery-live: uv run --with-requirements requirements.txt python scripts/adversarial_battery.py -scoped-smoke-loop: - uv run --with-requirements requirements.txt python scripts/scoped_smoke_loop.py $(SCOPED_SMOKE_ARGS) - # Railway deploy gate. Run under `railway run` (or with DATABASE_URL + # PUBLIC_URL exported) to check migration parity and live posture together. railway-preflight: diff --git a/docs/PROOF_MATRIX.md b/docs/PROOF_MATRIX.md index b4554bf0..417f3c56 100644 --- a/docs/PROOF_MATRIX.md +++ b/docs/PROOF_MATRIX.md @@ -188,18 +188,22 @@ Both write real rows to the target. Point them at staging. | Command | Proves | Requires | |---|---|---| | `make trust-conformance-live` | Golden path; sequential replay; 15 identical concurrent requests exposing one receipt identity or explicit `idempotency_in_progress`, followed by a completed replay and one charge; a changed payload under a reused key conflicting rather than replaying; budget denial; expired and forged permit rejection; receipt and audit-chain verification; tenant isolation against a directly-supplied foreign wallet and permit id | `AGENT_MIDDLEWARE_API_KEY`; set `AGENT_MIDDLEWARE_API_URL` or it defaults to production | -| `make scoped-smoke-loop` | Authority readable for the caller's own wallet; scoped permit issued and carrying a nonce; governed invoke; the debit matching the receipt's own stated cost; replay returning the same receipt and charging nothing further; a changed payload under a reused key refused; an out-of-scope but genuinely registered tool denied and not charged; the portable receipt carrying a signature, key id, and named canonicalization | `AGENT_MIDDLEWARE_API_KEY` holding a **wallet-scoped** key (the loop refuses a bootstrap admin key); `--tool` required off loopback | +| `python scripts/constant_test_loop.py` | Scoped permit sized from the tool's advertised `creditsPerCall`; governed invoke; signed success receipt with a ledger entry; replay returning the same `receipt_id` with no second debit; an out-of-scope but genuinely registered tool refused with `permit_tool_not_allowed` and charged nothing | `CI_SMOKE_AGENT_KEY` holding a **wallet-scoped** key (self-provisions on loopback when absent); `--tool` and `--tool-args` required off loopback | | `make adversarial-battery-live` | Wallet isolation, invalid-key rejection, forged-receipt rejection, permit key binding, expired permits, revoked keys, replay idempotency; always revokes keys it minted | `API_URL` (no default, by design) and `BOOTSTRAP_KEY` | -`scoped-smoke-loop` is the one built to run continuously: it needs no admin -credential, so the key it runs on cannot mint keys or read another tenant's -audit trail if the runner holding it is compromised. That is also its limit — -tenant isolation, audit-chain verification, and refund reconciliation are -absent from it precisely because a scoped key must not be able to reach them, -and they stay in `trust-conformance-live` for an operator to run deliberately. -It reports SKIP rather than a false PASS for the out-of-scope denial when the -deployment registers only one tool, since invoking an unregistered name proves -"tool not found" rather than "the permit refused it". +`constant_test_loop.py` is the one built to run continuously: it needs no +admin credential, so the key it runs on cannot mint keys or read another +tenant's audit trail if the runner holding it is compromised. That is also its +limit — tenant isolation, audit-chain verification, and refund reconciliation +are absent from it precisely because a scoped key must not reach them, and +they stay in `trust-conformance-live` for an operator to run deliberately. + +It skips rather than false-passes the out-of-scope denial when the deployment +registers only one tool, since invoking an unregistered name proves "tool not +found" rather than "the permit refused it". Off loopback it refuses to start +without `--tool` and `--tool-args`: arguments derived from a schema satisfy a +tool's declared shape but not its semantics, and the derivation picks the +first enum member, which for a consequential tool could be `delete`. The battery reports SKIP — never a false PASS — for MCP-invocation checks when the deployment exposes no invokable `golden-path-echo` tool. It does not diff --git a/docs/constant-test-loop.md b/docs/constant-test-loop.md index 1134d93d..73099cfc 100644 --- a/docs/constant-test-loop.md +++ b/docs/constant-test-loop.md @@ -24,6 +24,32 @@ The script self-provisions an agent key via `/v1/dev-keys/self-provision` when n ## Production Use +Off loopback the loop refuses to start without `--tool` and `--tool-args` +(or `$CI_SMOKE_TOOL` / the equivalent flags). This is deliberate, not a +papercut: + +- **Tool selection.** `partner.echo` and `partner.notes.write` are tried in + order when present, but which tool a monitor invokes on every run is an + operator's decision, not the registry ordering's. +- **Payload.** Arguments derived from a tool's `inputSchema` satisfy its + declared *shape*, not its *semantics*, and the derivation fills required + fields from types, defaults, and the **first enum member** — which for a + consequential tool could be `delete`. Schema validity is not evidence of + safety. Pass `--tool-args '{}'` if the tool genuinely needs no arguments; + the point is that it be a decision. + +The permit is sized from the selected tool's advertised +`annotations.creditsPerCall` rather than a fixed cap, so pointing the loop at +a pricier tool does not fail with `permit_budget_exceeded` on a healthy +deployment. + +```bash +python scripts/constant_test_loop.py \ + --api-url https://api.thisisatest.tech \ + --tool partner.echo \ + --tool-args '{"text": "constant test loop"}' +``` + Set `CI_SMOKE_AGENT_KEY` for a pre-provisioned agent credential. Optionally provide `CI_SMOKE_WALLET_ID` and `CI_SMOKE_KEY_ID` for faster startup (the script will fetch them from the API if not provided): ```bash @@ -36,7 +62,7 @@ export CI_SMOKE_AGENT_KEY="$(python scripts/partner_api_key_bootstrap.py \ --agent-id ci-smoke-agent \ --key-name constant-test-loop \ --budget-credits 5000 \ - --json | jq -r .api_key)" + --key-only)" # or --json | jq -r .api_key export CI_SMOKE_WALLET_ID="$(python scripts/partner_api_key_bootstrap.py \ --api-url https://api.thisisatest.tech \ diff --git a/scripts/constant_test_loop.py b/scripts/constant_test_loop.py index 25eed014..31a9c6e1 100755 --- a/scripts/constant_test_loop.py +++ b/scripts/constant_test_loop.py @@ -28,22 +28,139 @@ from __future__ import annotations import argparse +import json import os import sys from datetime import datetime, timedelta, timezone from decimal import Decimal from pathlib import Path from typing import Any +from urllib.parse import urlparse import httpx ROOT = Path(__file__).resolve().parents[1] DEFAULT_API_URL = "http://127.0.0.1:8000" -# Prefer partner.echo (production governed tool) over partner.notes.write (dogfood) +# Preferred governed tools, in order. Neither being present is no longer fatal: +# the registry differs between a local quickstart, a partner pilot, and +# production, and a loop that only runs where these two names exist is not a +# monitor you can point at an arbitrary deployment. PRODUCTION_GOVERNED_TOOL = "partner.echo" DOGFOOD_GOVERNED_TOOL = "partner.notes.write" -OUT_OF_SCOPE_TOOL = "some.other.tool" # Dummy tool not in permit scope -PERMIT_MAX_CREDITS = 10 + +# Headroom over the selected tool's advertised creditsPerCall. One run spends a +# single call — the replay is free and the denial never charges — so this is +# slack for a price that moved between discovery and invocation. A fixed cap +# fails permit_budget_exceeded on a healthy deployment the moment a pricier +# tool is selected, which reads as a product fault rather than a config one. +PERMIT_CREDIT_HEADROOM = 3 +PERMIT_MIN_CREDITS = 20 + + + +def credits_per_call(tool: dict) -> float: + """The tool's advertised price, or 0 when the manifest states none.""" + try: + return float((tool.get("annotations") or {}).get("creditsPerCall") or 0) + except (TypeError, ValueError): + return 0.0 + + +def arguments_for(schema: dict, marker: str) -> dict: + """Build a minimal argument set satisfying a tool's declared inputSchema. + + Only *required* properties are filled, typed as the schema declares, so the + loop can invoke a tool it was not written against without inventing intent + the caller never expressed. ``marker`` is woven into string values so a + run's test data is traceable back to this loop in whatever the tool writes. + + Note the enum branch takes the first member, which is why an operator- + supplied payload is mandatory off loopback: a tool declaring + ``["delete", "preview"]`` would otherwise be sent ``delete`` every run. + """ + properties = (schema or {}).get("properties") or {} + required = (schema or {}).get("required") or [] + arguments: dict = {} + for name in required: + spec = properties.get(name) or {} + if "default" in spec: + arguments[name] = spec["default"] + elif spec.get("enum"): + arguments[name] = spec["enum"][0] + else: + kind = spec.get("type") + arguments[name] = { + "integer": 1, + "number": 1.0, + "boolean": False, + "array": [], + "object": {}, + }.get(kind, f"smoke-{marker}") + return arguments + + +def select_governed_tool( + registry: dict[str, dict], pinned: str | None +) -> tuple[str, str]: + """Choose the tool the permit will scope to. + + Returns ``(tool, error)``. A pinned name must exist; otherwise the + preferred production and dogfood tools are tried in order, and failing + those any registered tool is acceptable — the loop asserts trust-plane + behaviour, which does not depend on which tool is on the other end. + """ + if pinned: + if pinned not in registry: + return "", ( + f"pinned tool {pinned} is not registered; available: " + f"{sorted(registry)[:6]}" + ) + return pinned, "" + for preferred in (PRODUCTION_GOVERNED_TOOL, DOGFOOD_GOVERNED_TOOL): + if preferred in registry: + return preferred, "" + if not registry: + return "", "deployment exposes no tools" + # Deliberately no "just use the first one" fallback. Schema-derived + # arguments satisfy a tool's declared shape but not its semantics — a tool + # can accept the payload and still refuse the action — and that failure + # reads as a broken trust plane when it is really a bad tool choice. Ask + # rather than guess. + return "", ( + f"neither {PRODUCTION_GOVERNED_TOOL} nor {DOGFOOD_GOVERNED_TOOL} is " + f"registered here. Pass --tool (or set $CI_SMOKE_TOOL) naming one " + f"this credential may invoke, with --tool-args if it needs arguments. " + f"Available: {', '.join(sorted(registry))}" + ) + + +def select_companion_tool( + registry: dict[str, dict], governed_tool: str, pinned: str | None +) -> tuple[str, str]: + """Pick a real registered tool the permit does not cover. + + Drawn from the registry, never from the pinned name: with the governed + tool pinned, selecting the companion from that same value would silently + skip the out-of-scope check. It must genuinely exist, because invoking a + name the registry has never heard of proves "tool not found" rather than + "the permit refused it" — a check passing for the wrong reason. + + Returns ``(tool_or_empty, error)``. Empty with no error means the + deployment has no second tool and the check is skipped. + """ + if pinned: + if pinned not in registry: + return "", ( + "an unregistered name proves tool-not-found, not permit " + f"refusal; available: {sorted(registry)[:6]}" + ) + if pinned == governed_tool: + return "", ( + "the companion must differ from the permitted tool, or the " + "check asserts a denial that should never happen" + ) + return pinned, "" + return next((n for n in registry if n != governed_tool), ""), "" class SmokeTestFailure(RuntimeError): @@ -176,7 +293,16 @@ def _first_jsonrpc_error(response: dict[str, Any]) -> dict[str, Any]: return response["error"] -def run_constant_test(api_url: str, agent_key: str, wallet_id: str, key_id: str) -> None: +def run_constant_test( + api_url: str, + agent_key: str, + wallet_id: str, + key_id: str, + *, + pinned_tool: str | None = None, + pinned_other_tool: str | None = None, + tool_arguments: dict | None = None, +) -> None: """Execute the constant test loop and assert all invariants.""" print(f"[constant-test] target: {api_url}", file=sys.stderr) @@ -247,21 +373,23 @@ def run_constant_test(api_url: str, agent_key: str, wallet_id: str, key_id: str) # Discover tools print("[constant-test] discovering tools", file=sys.stderr) tools = _get_json(client, "/mcp/tools.json", expected_status=200) - tool_names = [tool["name"] for tool in tools.get("tools", [])] - - # Select governed tool: prefer production partner.echo over dogfood partner.notes.write - governed_tool = None - if PRODUCTION_GOVERNED_TOOL in tool_names: - governed_tool = PRODUCTION_GOVERNED_TOOL - print(f"[constant-test] using production tool: {governed_tool}", file=sys.stderr) - elif DOGFOOD_GOVERNED_TOOL in tool_names: - governed_tool = DOGFOOD_GOVERNED_TOOL - print(f"[constant-test] using dogfood tool: {governed_tool} (ENABLE_DOGFOOD_TOOL=true)", file=sys.stderr) - else: - raise SmokeTestFailure( - f"neither {PRODUCTION_GOVERNED_TOOL} nor {DOGFOOD_GOVERNED_TOOL} discoverable. " - f"Available tools: {', '.join(tool_names)}" - ) + registry = { + tool["name"]: tool + for tool in tools.get("tools", []) + if isinstance(tool, dict) and tool.get("name") + } + tool_names = list(registry) + + governed_tool, selection_error = select_governed_tool(registry, pinned_tool) + if selection_error: + raise SmokeTestFailure(selection_error) + print(f"[constant-test] using tool: {governed_tool}", file=sys.stderr) + + # Sized from what this tool actually costs rather than a fixed cap. + max_credits = max( + PERMIT_MIN_CREDITS, + credits_per_call(registry[governed_tool]) * PERMIT_CREDIT_HEADROOM, + ) # Issue scoped permit print("[constant-test] issuing scoped permit", file=sys.stderr) @@ -281,7 +409,7 @@ def run_constant_test(api_url: str, agent_key: str, wallet_id: str, key_id: str) "subject_key_id": key_id, "allowed_tools": [governed_tool], "scopes": [f"tool:{governed_tool}:invoke", "billing:charge"], - "max_credits": PERMIT_MAX_CREDITS, + "max_credits": max_credits, "expires_at": expires_at, }, headers=permit_headers, @@ -297,13 +425,18 @@ def run_constant_test(api_url: str, agent_key: str, wallet_id: str, key_id: str) # Invoke the governed tool (unique idempotency key per run) invoke_idempotency_key = f"constant-test-invoke-{run_id}" print(f"[constant-test] invoking {governed_tool}", file=sys.stderr) + governed_arguments = ( + tool_arguments + if tool_arguments is not None + else arguments_for(registry[governed_tool].get("inputSchema") or {}, run_id) + ) call_body = _build_mcp_call( request_id=f"constant-test-req-{run_id}", tool=governed_tool, wallet_id=wallet_id, permit_id=permit_id, idempotency_key=invoke_idempotency_key, - arguments={"text": "constant test loop"}, + arguments=governed_arguments, ) first_call = _post_json( client, "/mcp/messages", call_body, expected_status=200 @@ -378,8 +511,11 @@ def run_constant_test(api_url: str, agent_key: str, wallet_id: str, key_id: str) # permit_tool_not_allowed without registering a second tool. if len(tool_names) > 1: print("[constant-test] attempting out-of-scope tool", file=sys.stderr) - # Find a tool that's not the governed one - other_tool = next((t for t in tool_names if t != governed_tool), None) + other_tool, companion_error = select_companion_tool( + registry, governed_tool, pinned_other_tool + ) + if companion_error: + raise SmokeTestFailure(companion_error) if other_tool: denial_body = _build_mcp_call( request_id=f"constant-test-deny-{run_id}", @@ -387,7 +523,9 @@ def run_constant_test(api_url: str, agent_key: str, wallet_id: str, key_id: str) wallet_id=wallet_id, permit_id=permit_id, idempotency_key=f"constant-test-deny-{run_id}", - arguments={}, + arguments=arguments_for( + registry[other_tool].get("inputSchema") or {}, run_id + ), ) denial_call = _post_json( client, "/mcp/messages", denial_body, expected_status=200 @@ -437,6 +575,32 @@ def _validate_api_url(url: str) -> str: return url +def _require_deliberate_production_target( + api_url: str, args: argparse.Namespace +) -> None: + """Off loopback, the tool and its payload must be operator-chosen. + + Discovery order and schema-derived arguments are development conveniences. + Against a real deployment they would pick whatever the registry lists + first and fill required fields from types, defaults, and the *first* enum + member — potentially "delete" — then send that again on every run. Schema + validity is not evidence of safety. + """ + if urlparse(api_url).hostname in ("localhost", "127.0.0.1", "::1"): + return + if not args.tool: + raise ConfigurationError( + f"refusing to auto-select a tool against {api_url}. Pass --tool " + "(or set $CI_SMOKE_TOOL) so you choose what this loop invokes." + ) + if args.tool_args is None: + raise ConfigurationError( + f"refusing to send derived arguments to {api_url}. Pass " + "--tool-args with a payload you have approved for repeated " + "invocation, or --tool-args '{}' if the tool needs none." + ) + + def main(argv: list[str] | None = None) -> int: """Entry point.""" parser = argparse.ArgumentParser(description=__doc__.splitlines()[0]) @@ -449,13 +613,49 @@ def main(argv: list[str] | None = None) -> int: "$CI_SMOKE_WALLET_ID, $CI_SMOKE_KEY_ID for auth, or self-provisions." ), ) + parser.add_argument( + "--tool", + default=os.environ.get("CI_SMOKE_TOOL") or None, + help=( + "Pin the tool the permit scopes to. Required off loopback: which " + "tool a monitor invokes on every run is an operator's decision, " + "not the registry ordering's." + ), + ) + parser.add_argument( + "--other-tool", + default=os.environ.get("CI_SMOKE_OTHER_TOOL") or None, + help=( + "Pin the registered tool used for the out-of-scope denial check " + "(default: any other discovered tool)" + ), + ) + parser.add_argument( + "--tool-args", + type=json.loads, + default=None, + help=( + "JSON object of arguments for the governed tool. Required off " + "loopback so the payload is one you approved for repeated " + "invocation; pass '{}' if the tool needs none." + ), + ) args = parser.parse_args(argv) try: # Validate API URL (applies to both --api-url and $API_URL) api_url = _validate_api_url(args.api_url or _get_api_url()) + _require_deliberate_production_target(api_url, args) agent_key, wallet_id, key_id = _get_agent_key() - run_constant_test(api_url, agent_key, wallet_id, key_id) + run_constant_test( + api_url, + agent_key, + wallet_id, + key_id, + pinned_tool=args.tool, + pinned_other_tool=args.other_tool, + tool_arguments=args.tool_args, + ) except ConfigurationError as config_error: print(f"\n[constant-test] configuration error: {config_error}", file=sys.stderr) return 2 diff --git a/scripts/scoped_smoke_loop.py b/scripts/scoped_smoke_loop.py deleted file mode 100644 index 758f1215..00000000 --- a/scripts/scoped_smoke_loop.py +++ /dev/null @@ -1,612 +0,0 @@ -#!/usr/bin/env python3 -"""Continuous trust-plane smoke loop that runs on a wallet-scoped key. - -``trust_plane_conformance.py`` is the fuller suite, but it needs a bootstrap -admin key: it creates sponsor wallets, agent wallets, and API keys, and it -reads the audit plane. That makes it the wrong tool for a monitoring agent -left running against a deployment — a credential that can mint keys and read -every tenant's audit trail is not one you leave in a CI runner's environment. - -This loop asserts the subset of invariants a *wallet-scoped* key can prove -about itself, which is most of what the product actually sells: - - authority -> permit -> invoke -> receipt -> replay-is-not-recharged - -> out-of-scope denial -> portable receipt verifies offline - -Every stage is asserted. The run exits non-zero the moment an invariant -breaks, so it can gate a deploy or page whoever is on call. - -What it deliberately cannot check, because a scoped key must not be able to: -tenant isolation across two wallets, audit-chain verification, and refund -reconciliation. Those stay in the admin-gated conformance suite, run -deliberately by an operator rather than continuously by an agent. - -Usage:: - - export AGENT_MIDDLEWARE_API_KEY=... # a WALLET-SCOPED key - export AGENT_MIDDLEWARE_API_URL=https://... # optional; defaults to prod - python scripts/scoped_smoke_loop.py --once - python scripts/scoped_smoke_loop.py --interval 300 --max-iterations 0 - -The key is read from the environment only: never an argument, so it stays out -of process listings and shell history. It is never printed, and the failure -paths below truncate response bodies that could echo it back. -""" - -from __future__ import annotations - -import argparse -import asyncio -import json -import os -import sys -import uuid -from dataclasses import dataclass, field -from datetime import datetime, timedelta, timezone - -import httpx -from urllib.parse import urlsplit - -DEFAULT_API_URL = "https://api.thisisatest.tech" -API_URL = os.environ.get("AGENT_MIDDLEWARE_API_URL", DEFAULT_API_URL).rstrip("/") -API_KEY = os.environ.get("AGENT_MIDDLEWARE_API_KEY", "") - -# Tool names are per-deployment: the registry differs between a local -# quickstart, a partner pilot, and production. Hardcoding one makes the loop -# fail with "tool not found" on every deployment but the author's, so the -# names are discovered unless the operator pins them with --tool/--other-tool. -# Headroom over the selected tool's advertised price. One iteration spends a -# single call — the replay is free and the denial never charges — so this is -# slack for a price that moved between discovery and invocation, not a budget -# for repeated spending. -PERMIT_CREDIT_HEADROOM = 3 -PERMIT_MIN_CREDITS = 20 - -# The exact refusal a permit gives for a tool outside allowed_tools. The -# out-of-scope check asserts this string rather than "any error", so it -# cannot be satisfied by a tool-not-found or a transport failure. -PERMIT_SCOPE_DENIAL = "permit_tool_not_allowed" -# The refusal for a reused idempotency key carrying a different payload. -IDEMPOTENCY_CONFLICT = "idempotency_key_reused" - - -def _redact(text: str, limit: int = 200) -> str: - """Truncate a response body and scrub anything key-shaped out of it. - - A 4xx body can echo the submitted credential back. This loop is built to - run unattended into logs someone else reads, so nothing that looks like a - key reaches stdout even on the failure paths. - """ - snippet = text[:limit] - if API_KEY and API_KEY in snippet: - snippet = snippet.replace(API_KEY, "") - return snippet - - -@dataclass -class Suite: - passed: int = 0 - failed: int = 0 - failures: list[str] = field(default_factory=list) - - def check(self, name: str, ok: bool, detail: str = "") -> bool: - tail = f" — {detail}" if detail else "" - if ok: - self.passed += 1 - print(f"[PASS] {name}{tail}", flush=True) - else: - self.failed += 1 - self.failures.append(name) - print(f"[FAIL] {name}{tail}", flush=True) - return ok - - -def headers(run_tag: str, idempotency_key: str | None = None) -> dict[str, str]: - h = {"X-API-Key": API_KEY, "Content-Type": "application/json"} - if idempotency_key: - h["Idempotency-Key"] = f"{run_tag}-{idempotency_key}" - return h - - -def invoke_body( - wallet_id: str, permit_id: str, tool: str, arguments: dict -) -> dict: - return { - "jsonrpc": "2.0", - "id": 1, - "method": "tools/call", - "params": { - "name": tool, - "arguments": arguments, - "mcpContext": {"wallet_id": wallet_id, "permit_id": permit_id}, - }, - } - - -def arguments_for(schema: dict, marker: str) -> dict: - """Build a minimal argument set satisfying a tool's declared inputSchema. - - Tools differ per deployment, so a fixed ``{"text": ...}`` payload fails on - every registry but the one it was written against. Filling only the - *required* properties, typed as the schema declares, keeps the loop - portable without pretending to know any particular tool's semantics. - - ``marker`` is woven into string values so a run's test data is traceable - back to this loop in whatever the tool writes. - """ - properties = (schema or {}).get("properties") or {} - required = (schema or {}).get("required") or [] - arguments: dict = {} - for name in required: - spec = properties.get(name) or {} - if "default" in spec: - arguments[name] = spec["default"] - continue - enum = spec.get("enum") - if enum: - arguments[name] = enum[0] - continue - kind = spec.get("type") - if kind == "integer": - arguments[name] = 1 - elif kind == "number": - arguments[name] = 1.0 - elif kind == "boolean": - arguments[name] = False - elif kind == "array": - arguments[name] = [] - elif kind == "object": - arguments[name] = {} - else: - arguments[name] = f"smoke-{marker}" - return arguments - - -def _spent(permit: dict) -> float: - """Credits consumed against a permit, however the field is spelled. - - Reading the permit back is how this loop proves a replay did not charge - twice, so a renamed field must fail loudly rather than silently return 0 - and make the double-charge check vacuous. - """ - for key in ("spent_credits", "credits_spent", "spent"): - if key in permit: - return float(permit[key]) - raise KeyError( - f"permit carries no recognised spend field; got keys: {sorted(permit)}" - ) - - -def select_companion_tool( - registry: dict[str, dict], allowed_tool: str, pinned: str | None -) -> tuple[str | None, str]: - """Pick a real registered tool the permit does not cover. - - Drawn from the *registry*, never from the pinned name: ``--tool`` is - mandatory off loopback, so selecting the companion from the pinned list - would silently skip the out-of-scope check on exactly the deployments - where it matters most. - - It must be a tool that genuinely exists, because invoking a name the - registry has never heard of proves "tool not found" rather than "the - permit refused it" — a pass for the wrong reason. - - Returns ``(tool_or_None, error_message)``. ``None`` with no error means - the deployment has no second tool and the check should be skipped. - """ - if pinned: - if pinned not in registry: - return None, ( - "an unregistered name proves tool-not-found, not permit " - f"refusal; registry: {sorted(registry)[:6]}" - ) - if pinned == allowed_tool: - return None, ( - "the companion must differ from the permitted tool, or the " - "check asserts a denial that should never happen" - ) - return pinned, "" - return next((name for name in registry if name != allowed_tool), None), "" - - -async def discover_tools(client: httpx.AsyncClient) -> dict[str, dict]: - """Return {tool name: manifest entry} for what this deployment exposes.""" - - r = await client.get("/mcp/tools.json") - r.raise_for_status() - tools = r.json().get("tools") or [] - return { - t["name"]: t for t in tools if isinstance(t, dict) and t.get("name") - } - - -def credits_per_call(tool: dict) -> float: - """The tool's advertised price, or 0 when the manifest does not state one.""" - - try: - return float((tool.get("annotations") or {}).get("creditsPerCall") or 0) - except (TypeError, ValueError): - return 0.0 - - -async def run_once( - client: httpx.AsyncClient, suite: Suite, args: argparse.Namespace -) -> None: - run_tag = f"smoke-{uuid.uuid4().hex[:10]}" - - # ---- Who am I? ------------------------------------------------------- - # Doubles as the guard against running on the wrong credential class: - # this route answers 403 wallet_key_required for a bootstrap admin key, - # so an operator who exports the admin key by mistake gets told so - # instead of quietly writing admin-authored test data every 5 minutes. - r = await client.get("/v1/me/authority", headers=headers(run_tag)) - if r.status_code in (401, 403): - # Distinguish the two failures that both surface here, because they - # send whoever is paged in opposite directions: a rejected key is a - # credential problem, while wallet_key_required means the key is - # valid but is the admin key this loop must never run on. - try: - error = (r.json().get("detail") or {}).get("error", "") - except ValueError: - error = "" - if error == "wallet_key_required": - suite.check( - "key is wallet-scoped, not bootstrap admin", - False, - "the configured key is a bootstrap admin key; this loop " - "must run on a wallet-scoped key", - ) - else: - suite.check( - "API key is accepted", - False, - f"HTTP {r.status_code} {error or 'rejected'} — check " - "AGENT_MIDDLEWARE_API_KEY for this deployment", - ) - return - if not suite.check( - "authority readable for own wallet", - r.status_code == 200, - f"HTTP {r.status_code}: {_redact(r.text)}", - ): - return - authority = r.json() - wallet_id = authority.get("wallet_id") - if not suite.check("authority names the caller's wallet", bool(wallet_id)): - return - - # ---- Which tools does this deployment actually expose? ---------------- - registry = await discover_tools(client) - if not suite.check("deployment exposes at least one tool", bool(registry)): - return - allowed_tool = args.tool or next(iter(registry)) - if not suite.check( - f"tool {allowed_tool} is registered", - allowed_tool in registry, - f"registry: {sorted(registry)[:6]}", - ): - return - other_tool, companion_error = select_companion_tool( - registry, allowed_tool, args.other_tool - ) - if companion_error: - suite.check(f"out-of-scope companion {args.other_tool}", False, companion_error) - - # ---- Issue a permit for the caller's own wallet ----------------------- - # Sized from what this tool actually costs. A fixed cap fails - # permit_budget_exceeded on a perfectly healthy deployment the moment - # someone pins a tool priced above it, which reads as a product fault. - price = credits_per_call(registry[allowed_tool]) - max_credits = args.max_credits or max( - PERMIT_MIN_CREDITS, price * PERMIT_CREDIT_HEADROOM - ) - expires_at = (datetime.now(timezone.utc) + timedelta(minutes=15)).isoformat() - r = await client.post( - "/v1/permits", - headers=headers(run_tag, "permit"), - json={ - "issuer_wallet_id": wallet_id, - "subject_wallet_id": wallet_id, - "allowed_tools": [allowed_tool], - "max_credits": str(max_credits), - "expires_at": expires_at, - }, - ) - if not suite.check( - "issue scoped permit", - r.status_code == 201, - f"HTTP {r.status_code}: {_redact(r.text)}", - ): - return - permit = r.json() - permit_id = permit["permit_id"] - suite.check("permit carries a nonce", bool(permit.get("nonce"))) - - # ---- Golden path: one governed invocation ---------------------------- - # Keep this body byte-identical for the replay below. The idempotency - # layer hashes the payload, so resending a *different* body under the - # same key is a conflict rather than a replay — asserted separately. - golden_args = args.tool_args or arguments_for( - registry[allowed_tool].get("inputSchema") or {}, run_tag - ) - golden = invoke_body(wallet_id, permit_id, allowed_tool, golden_args) - r = await client.post( - "/mcp/messages", headers=headers(run_tag, "invoke"), json=golden - ) - body = r.json() - receipt = body.get("result", {}).get("receipt", {}) - receipt_id = receipt.get("receipt_id") - if not suite.check( - "governed invoke succeeds", - "error" not in body and bool(receipt_id), - _redact(json.dumps(body)), - ): - return - - r = await client.get(f"/v1/permits/{permit_id}", headers=headers(run_tag)) - if not suite.check("permit readable after invoke", r.status_code == 200): - return - spent_after_first = _spent(r.json()) - # Assert against the receipt's own figure rather than a hardcoded price: - # a pricing change should not fail this loop, but a debit that disagrees - # with the receipt it is supposed to evidence absolutely should. - charged = receipt.get("cost_credits", receipt.get("credits_charged")) - suite.check( - "the debit matches the receipt's stated cost", - charged is not None and float(charged) == spent_after_first, - f"permit spent={spent_after_first}, receipt says {charged}", - ) - - # ---- The claim the product rests on: replay does not recharge --------- - r = await client.post( - "/mcp/messages", headers=headers(run_tag, "invoke"), json=golden - ) - replay = r.json() - replay_id = replay.get("result", {}).get("receipt", {}).get("receipt_id") - suite.check( - "replay returns the same receipt", - replay_id == receipt_id, - f"first={receipt_id}, replay={replay_id}", - ) - - r = await client.get(f"/v1/permits/{permit_id}", headers=headers(run_tag)) - spent_after_replay = _spent(r.json()) - suite.check( - "replay did not charge a second time", - spent_after_replay == spent_after_first, - f"before={spent_after_first}, after={spent_after_replay}", - ) - - # ---- A different payload under the same key is a conflict, not a replay - r = await client.post( - "/mcp/messages", - headers=headers(run_tag, "invoke"), - json=invoke_body( - wallet_id, permit_id, allowed_tool, {**golden_args, "_smoke": "differs"} - ), - ) - conflict = r.json() - # The exact reason, not any error: a tool rejecting the injected argument, - # or an unrelated policy failure, would otherwise report PASS while the - # idempotency behaviour this asserts went unverified. - conflict_message = str((conflict.get("error") or {}).get("message", "")) - suite.check( - "different payload under a used key is refused as a conflict", - conflict_message == IDEMPOTENCY_CONFLICT, - conflict_message - or f"HTTP {r.status_code}: {_redact(json.dumps(conflict))}", - ) - - # ---- Out-of-scope tool is denied, and the denial costs nothing -------- - if other_tool is None: - print( - "[SKIP] out-of-scope denial — no registered tool outside the " - "permit to try against", - flush=True, - ) - else: - r = await client.post( - "/mcp/messages", - headers=headers(run_tag, "outofscope"), - json=invoke_body( - wallet_id, - permit_id, - other_tool, - arguments_for( - (registry.get(other_tool) or {}).get("inputSchema") or {}, - run_tag, - ), - ), - ) - denied = r.json() - # Assert the *specific* refusal, not merely "something went wrong". - # Any-error-counts would pass on a transport hiccup or a schema - # complaint, and "receipt has no ok field" passes on success, since a - # successful receipt carries no such field at all. - error_message = str((denied.get("error") or {}).get("message", "")) - suite.check( - f"out-of-scope tool {other_tool} is refused by the permit", - error_message == PERMIT_SCOPE_DENIAL, - error_message or _redact(json.dumps(denied)), - ) - - r = await client.get(f"/v1/permits/{permit_id}", headers=headers(run_tag)) - spent_after_denial = _spent(r.json()) - suite.check( - "a denied call is not charged", - spent_after_denial == spent_after_replay, - f"before={spent_after_replay}, after={spent_after_denial}", - ) - - # ---- The receipt is portable and verifies with no credential ---------- - r = await client.get( - f"/v1/receipts/{receipt_id}/portable", headers=headers(run_tag) - ) - if not suite.check( - "portable receipt bundle retrievable", - r.status_code == 200, - f"HTTP {r.status_code}", - ): - return - bundle = r.json() - suite.check( - "portable bundle carries a signature and key id", - bool(bundle.get("signature")) and bool(bundle.get("kid")), - f"keys: {sorted(bundle)[:8]}", - ) - # The canonicalization is the byte-level contract an independent verifier - # implements; without it a signature cannot be checked offline at all. - # `issuer` is deliberately not asserted: it is empty on a local instance - # with no PUBLIC_URL, and failing there would be a config complaint - # dressed up as a trust-plane failure. - suite.check( - "portable bundle names its canonicalization", - bool(bundle.get("canonicalization")), - f"canonicalization={bundle.get('canonicalization')!r}, " - f"issuer={bundle.get('issuer')!r}", - ) - - -async def main_async(args: argparse.Namespace) -> int: - iteration = 0 - worst = 0 - async with httpx.AsyncClient(base_url=API_URL, timeout=60.0) as client: - while True: - iteration += 1 - started = datetime.now(timezone.utc) - suite = Suite() - print( - f"\n=== iteration {iteration} — {started.isoformat()} " - f"— {API_URL} ===", - flush=True, - ) - try: - await run_once(client, suite, args) - except (httpx.HTTPError, KeyError, ValueError) as exc: - # A transport error or a renamed field is a failed iteration, - # not a crashed monitor: a loop that dies on the first blip - # stops watching exactly when something is going wrong. - suite.check("iteration completed without error", False, str(exc)[:200]) - - print( - f"--- {suite.passed} passed, {suite.failed} failed" - + (f" ({', '.join(suite.failures)})" if suite.failures else ""), - flush=True, - ) - if suite.failed: - worst = 1 - if args.exit_on_failure: - return 1 - - if args.once or ( - args.max_iterations and iteration >= args.max_iterations - ): - return worst - await asyncio.sleep(args.interval) - - -def main() -> int: - parser = argparse.ArgumentParser( - description=( - "Continuous trust-plane smoke loop for a wallet-scoped API key." - ) - ) - parser.add_argument( - "--once", action="store_true", help="Run a single iteration and exit" - ) - parser.add_argument( - "--interval", - type=float, - default=300.0, - help="Seconds between iterations (default: 300)", - ) - parser.add_argument( - "--max-iterations", - type=int, - default=0, - help="Stop after N iterations; 0 runs until interrupted", - ) - parser.add_argument( - "--tool", - default=os.environ.get("SMOKE_TOOL") or None, - help="Pin the tool the permit scopes to (default: first discovered)", - ) - parser.add_argument( - "--other-tool", - default=os.environ.get("SMOKE_OTHER_TOOL") or None, - help=( - "Pin the real, registered tool used for the out-of-scope denial " - "check (default: second discovered)" - ), - ) - parser.add_argument( - "--tool-args", - type=json.loads, - default=None, - help=( - "JSON object of arguments for the pinned tool. Set this when " - "running against production so the loop sends a payload you have " - "chosen, rather than one derived from the tool's schema." - ), - ) - parser.add_argument( - "--max-credits", - type=float, - default=None, - help=( - "Permit cap in credits (default: 3x the tool's advertised " - "creditsPerCall, floor 20)" - ), - ) - parser.add_argument( - "--exit-on-failure", - action="store_true", - help="Exit non-zero on the first failing iteration instead of continuing", - ) - args = parser.parse_args() - - loopback = urlsplit(API_URL).hostname in {"localhost", "127.0.0.1", "::1"} - if not loopback and not args.tool: - sys.stderr.write( - f"refusing to auto-discover a tool against {API_URL}.\n" - "Off loopback, pass --tool so you choose what this loop invokes " - "on every iteration rather than whatever the registry happens to " - "list first.\n" - ) - return 2 - - if not loopback and args.tool_args is None: - # Schema validity is not evidence of safety. Derived arguments fill - # required fields from types, defaults, and the *first* enum member — - # which for a consequential tool could be "delete" or "commit", sent - # again every interval. Off loopback the payload is a decision an - # operator makes explicitly, even if that decision is "{}". - sys.stderr.write( - f"refusing to send derived arguments to {API_URL}.\n" - "Off loopback, pass --tool-args with a payload you have approved " - "for repeated invocation; derived arguments guess from the " - "schema and may name a destructive enum value. Pass " - "--tool-args '{}' if the tool genuinely needs no arguments.\n" - ) - return 2 - - if not API_KEY: - return int( - bool( - sys.stderr.write( - "AGENT_MIDDLEWARE_API_KEY is not set.\n" - "Export a wallet-scoped key for the target deployment " - "before running; never pass it as an argument.\n" - ) - ) - ) - - try: - return asyncio.run(main_async(args)) - except KeyboardInterrupt: - print("\ninterrupted", file=sys.stderr) - return 130 - - -if __name__ == "__main__": - raise SystemExit(main()) diff --git a/tests/test_constant_test_loop.py b/tests/test_constant_test_loop.py index c0b5e1c5..f49beeec 100644 --- a/tests/test_constant_test_loop.py +++ b/tests/test_constant_test_loop.py @@ -21,6 +21,7 @@ import httpx import pytest +import runpy REPO_ROOT = Path(__file__).resolve().parent.parent BOOTSTRAP_SCRIPT = REPO_ROOT / "scripts" / "partner_api_key_bootstrap.py" @@ -315,3 +316,225 @@ def test_constant_test_loop_never_logs_api_key(test_server): # The agent key must never appear anywhere in the output assert agent_key not in result.stdout, "agent key in stdout" assert agent_key not in result.stderr, "agent key in stderr" + + +# --- Tool selection, argument derivation, and the production guards ------- +# +# Ported from the scoped-smoke-loop work (#325) when the two loops were +# consolidated. These need no server, so they run in the fast path. + + +def _load_loop(): + return runpy.run_path(str(CONSTANT_TEST_SCRIPT)) + + +def test_companion_tool_comes_from_the_registry_not_the_governed_name(): + """Regression: pinning the governed tool must not skip the denial check. + + Selecting the companion from the pinned value rather than the registry + made the out-of-scope check skip silently on exactly the deployments + where a pin is mandatory. + """ + select = _load_loop()["select_companion_tool"] + registry = {"alpha": {}, "beta": {}, "gamma": {}} + + companion, error = select(registry, "alpha", None) + assert companion in {"beta", "gamma"} + assert not error + + companion, error = select(registry, "gamma", None) + assert companion in {"alpha", "beta"} + assert not error + + +def test_companion_tool_must_be_registered_and_distinct(): + """Both ways the denial check could pass without proving permit scoping.""" + select = _load_loop()["select_companion_tool"] + registry = {"alpha": {}, "beta": {}} + + # An unregistered name yields tool-not-found, not a permit refusal. + companion, error = select(registry, "alpha", "does-not-exist") + assert companion == "" + assert "tool-not-found" in error + + # A companion equal to the permitted tool asserts a denial that should + # never happen. + companion, error = select(registry, "alpha", "alpha") + assert companion == "" + assert "differ" in error + + assert select(registry, "alpha", "beta") == ("beta", "") + + +def test_single_tool_deployment_skips_rather_than_false_passes(): + select = _load_loop()["select_companion_tool"] + assert select({"only": {}}, "only", None) == ("", "") + + +def test_governed_tool_selection_asks_rather_than_guessing(): + """No "just use the first one" fallback. + + Schema-derived arguments satisfy a tool's declared shape but not its + semantics, so an arbitrary pick can be refused by the tool itself — a + failure that reads as a broken trust plane rather than a bad choice. + """ + select = _load_loop()["select_governed_tool"] + + assert select({"partner.echo": {}, "other": {}}, None) == ("partner.echo", "") + assert select({"partner.notes.write": {}}, None) == ("partner.notes.write", "") + + tool, error = select({"alpha": {}, "beta": {}}, None) + assert tool == "" + assert "--tool" in error and "alpha" in error + + tool, error = select({"alpha": {}}, "nope") + assert tool == "" + assert "not registered" in error + + assert select({"alpha": {}}, "alpha") == ("alpha", "") + + +def test_arguments_are_derived_from_each_tool_schema(): + """A fixed payload shape only ever fits the registry it was written for.""" + arguments_for = _load_loop()["arguments_for"] + + built = arguments_for( + { + "required": ["session_id", "count", "ratio", "flag", "items", "meta"], + "properties": { + "session_id": {"type": "string"}, + "count": {"type": "integer"}, + "ratio": {"type": "number"}, + "flag": {"type": "boolean"}, + "items": {"type": "array"}, + "meta": {"type": "object"}, + "ignored": {"type": "string"}, + }, + }, + "run1", + ) + # Only required properties: sending optional ones invents intent the + # caller never expressed. + assert set(built) == {"session_id", "count", "ratio", "flag", "items", "meta"} + assert built["session_id"] == "smoke-run1" + assert built["count"] == 1 + assert built["ratio"] == 1.0 + assert built["flag"] is False + assert built["items"] == [] + assert built["meta"] == {} + + # A schema that states an acceptable value beats a guess. + built = arguments_for( + { + "required": ["mode", "region"], + "properties": { + "mode": {"type": "string", "enum": ["preview", "commit"]}, + "region": {"type": "string", "default": "us-east-1"}, + }, + }, + "run2", + ) + assert built["mode"] == "preview" + assert built["region"] == "us-east-1" + + assert arguments_for({}, "run3") == {} + assert arguments_for(None, "run3") == {} + + +def test_advertised_price_is_read_from_the_manifest(): + """A fixed permit cap fails permit_budget_exceeded on a healthy deployment. + + Tool prices in this repo span 1-200 credits, so a cap chosen for one tool + refuses the golden invocation the moment another is selected. + """ + credits_per_call = _load_loop()["credits_per_call"] + + assert credits_per_call({"annotations": {"creditsPerCall": 25.0}}) == 25.0 + assert credits_per_call({"annotations": {"creditsPerCall": "15"}}) == 15.0 + # No stated price must not raise; the floor applies instead. + assert credits_per_call({}) == 0.0 + assert credits_per_call({"annotations": {}}) == 0.0 + assert credits_per_call({"annotations": {"creditsPerCall": "free"}}) == 0.0 + + +def test_off_loopback_runs_require_a_pinned_tool_and_approved_payload(): + """Nobody should discover their way into invoking production every run. + + Derived arguments fill required fields from types, defaults, and the first + enum member -- which for a consequential tool could be "delete". + """ + env = { + k: v for k, v in os.environ.items() if not k.startswith("CI_SMOKE") + } + env["PATH"] = os.environ.get("PATH", "/usr/bin:/bin") + env["CI_SMOKE_AGENT_KEY"] = "b2a_placeholder" + + result = subprocess.run( + [ + sys.executable, + str(CONSTANT_TEST_SCRIPT), + "--api-url", + "https://api.thisisatest.tech", + ], + env=env, + capture_output=True, + text=True, + timeout=60, + ) + assert result.returncode == 2, result.stdout + result.stderr + assert "refusing to auto-select a tool" in result.stderr + # The guard fires before any network call, and leaks no credential. + assert "b2a_placeholder" not in result.stdout + result.stderr + + result = subprocess.run( + [ + sys.executable, + str(CONSTANT_TEST_SCRIPT), + "--api-url", + "https://api.thisisatest.tech", + "--tool", + "partner.echo", + ], + env=env, + capture_output=True, + text=True, + timeout=60, + ) + assert result.returncode == 2, result.stdout + result.stderr + assert "refusing to send derived arguments" in result.stderr + assert "b2a_placeholder" not in result.stdout + result.stderr + + +def test_bootstrap_key_only_pipes_without_jq(): + """--key-only and --json both write stdout, so they cannot both apply.""" + result = subprocess.run( + [ + sys.executable, + str(BOOTSTRAP_SCRIPT), + "--api-url", + "https://api.example.org", + "--key-only", + "--json", + ], + capture_output=True, + text=True, + env={"PATH": "/usr/bin:/bin", "BOOTSTRAP_KEY": "placeholder"}, + ) + assert result.returncode != 0 + assert "mutually exclusive" in result.stderr + + # --key-only must not have opened a path around the fail-closed check. + result = subprocess.run( + [ + sys.executable, + str(BOOTSTRAP_SCRIPT), + "--api-url", + "https://api.example.org", + "--key-only", + ], + capture_output=True, + text=True, + env={"PATH": "/usr/bin:/bin"}, + ) + assert result.returncode != 0 + assert "BOOTSTRAP_KEY" in result.stderr diff --git a/tests/test_scoped_smoke_loop.py b/tests/test_scoped_smoke_loop.py deleted file mode 100644 index c70a55e0..00000000 --- a/tests/test_scoped_smoke_loop.py +++ /dev/null @@ -1,328 +0,0 @@ -"""Contracts for the wallet-scoped smoke loop and the mint that feeds it. - -The loop is meant to run unattended against a live deployment on a credential -that is *not* the admin key. These tests pin the properties that make that -safe — argument derivation, credential redaction, the off-loopback guard — -without needing a server, so they run in CI alongside everything else. -""" - -from __future__ import annotations - -import runpy -import subprocess -import sys -from pathlib import Path - -import pytest - -ROOT = Path(__file__).resolve().parents[1] -LOOP = ROOT / "scripts" / "scoped_smoke_loop.py" -BOOTSTRAP = ROOT / "scripts" / "partner_api_key_bootstrap.py" - - -def _load_loop(monkeypatch, *, api_key: str = "b2a_test_key", api_url: str = ""): - """Import the loop with a chosen environment. - - Module-level constants read the environment at import time, so the env has - to be set before ``run_path`` rather than after. - """ - monkeypatch.setenv("AGENT_MIDDLEWARE_API_KEY", api_key) - if api_url: - monkeypatch.setenv("AGENT_MIDDLEWARE_API_URL", api_url) - else: - monkeypatch.delenv("AGENT_MIDDLEWARE_API_URL", raising=False) - return runpy.run_path(str(LOOP)) - - -def test_arguments_are_derived_from_each_tool_schema(monkeypatch) -> None: - """A fixed payload shape only ever fits the registry it was written for.""" - - arguments_for = _load_loop(monkeypatch)["arguments_for"] - - schema = { - "type": "object", - "required": ["session_id", "count", "ratio", "flag", "items", "meta"], - "properties": { - "session_id": {"type": "string"}, - "count": {"type": "integer"}, - "ratio": {"type": "number"}, - "flag": {"type": "boolean"}, - "items": {"type": "array"}, - "meta": {"type": "object"}, - "ignored": {"type": "string"}, - }, - } - built = arguments_for(schema, "run1") - - # Only required properties: sending optional ones invents intent the - # caller never expressed. - assert set(built) == {"session_id", "count", "ratio", "flag", "items", "meta"} - assert built["session_id"] == "smoke-run1" - assert built["count"] == 1 - assert built["ratio"] == 1.0 - assert built["flag"] is False - assert built["items"] == [] - assert built["meta"] == {} - - -def test_argument_derivation_prefers_declared_defaults_and_enums( - monkeypatch, -) -> None: - """A schema that states an acceptable value is more trustworthy than a guess.""" - - arguments_for = _load_loop(monkeypatch)["arguments_for"] - - built = arguments_for( - { - "required": ["mode", "region"], - "properties": { - "mode": {"type": "string", "enum": ["preview", "commit"]}, - "region": {"type": "string", "default": "us-east-1"}, - }, - }, - "run2", - ) - assert built["mode"] == "preview" - assert built["region"] == "us-east-1" - - -def test_argument_derivation_tolerates_a_schemaless_tool(monkeypatch) -> None: - arguments_for = _load_loop(monkeypatch)["arguments_for"] - - assert arguments_for({}, "run3") == {} - assert arguments_for(None, "run3") == {} - - -def test_response_bodies_never_echo_the_credential(monkeypatch) -> None: - """A 4xx body can quote the key back; this loop logs into shared places.""" - - module = _load_loop(monkeypatch, api_key="b2a_supersecret_value") - redact = module["_redact"] - - body = '{"error":"invalid_api_key","supplied":"b2a_supersecret_value"}' - scrubbed = redact(body, limit=500) - assert "b2a_supersecret_value" not in scrubbed - assert "" in scrubbed - - -def test_spend_field_rename_fails_loudly(monkeypatch) -> None: - """Reading spend back is how the no-double-charge check has any teeth. - - If the field were renamed and this silently returned 0, the replay check - would compare 0 to 0 and pass while the invariant went unverified. - """ - - _spent = _load_loop(monkeypatch)["_spent"] - - assert _spent({"spent_credits": "15.0"}) == 15.0 - assert _spent({"credits_spent": 2}) == 2.0 - with pytest.raises(KeyError): - _spent({"permit_id": "permit-1", "max_credits": "20"}) - - -def test_off_loopback_runs_require_a_pinned_tool() -> None: - """Nobody should discover their way into invoking production every 5 minutes.""" - - result = subprocess.run( - [sys.executable, str(LOOP), "--once"], - capture_output=True, - text=True, - env={ - "PATH": "/usr/bin:/bin", - "AGENT_MIDDLEWARE_API_URL": "https://api.thisisatest.tech", - "AGENT_MIDDLEWARE_API_KEY": "b2a_placeholder", - }, - ) - assert result.returncode == 2, result.stdout + result.stderr - assert "refusing to auto-discover" in result.stderr - # The guard must fire before any network call, and must not leak the key. - assert "b2a_placeholder" not in result.stdout + result.stderr - - -def test_loopback_runs_may_discover(monkeypatch, tmp_path) -> None: - """The pin is a production guard, not a burden on local development.""" - - result = subprocess.run( - [sys.executable, str(LOOP), "--once"], - capture_output=True, - text=True, - env={ - "PATH": "/usr/bin:/bin", - "AGENT_MIDDLEWARE_API_URL": "http://127.0.0.1:1", - "AGENT_MIDDLEWARE_API_KEY": "b2a_placeholder", - }, - ) - # Port 1 refuses the connection, so this fails — but as a failed check - # inside a running loop, not as the off-loopback refusal. - assert result.returncode == 1, result.stdout + result.stderr - assert "refusing to auto-discover" not in result.stderr - - -def test_loop_requires_a_key_and_never_takes_one_as_an_argument() -> None: - result = subprocess.run( - [sys.executable, str(LOOP), "--once", "--tool", "x"], - capture_output=True, - text=True, - env={"PATH": "/usr/bin:/bin", "AGENT_MIDDLEWARE_API_URL": "http://127.0.0.1:1"}, - ) - assert result.returncode != 0 - assert "AGENT_MIDDLEWARE_API_KEY is not set" in result.stderr - - help_text = subprocess.run( - [sys.executable, str(LOOP), "--help"], capture_output=True, text=True - ).stdout - assert "--api-key" not in help_text, ( - "a key passed as an argument is visible in process listings" - ) - - -def test_bootstrap_key_only_and_json_are_mutually_exclusive() -> None: - """Two output modes writing to stdout at once would corrupt either pipe.""" - - result = subprocess.run( - [ - sys.executable, - str(BOOTSTRAP), - "--api-url", - "https://api.example.org", - "--key-only", - "--json", - ], - capture_output=True, - text=True, - env={"PATH": "/usr/bin:/bin", "BOOTSTRAP_KEY": "placeholder"}, - ) - assert result.returncode != 0 - assert "mutually exclusive" in result.stderr - - -def test_bootstrap_still_refuses_to_mint_without_a_bootstrap_secret() -> None: - """--key-only must not have opened a path that skips the fail-closed check.""" - - result = subprocess.run( - [ - sys.executable, - str(BOOTSTRAP), - "--api-url", - "https://api.example.org", - "--key-only", - ], - capture_output=True, - text=True, - env={"PATH": "/usr/bin:/bin"}, - ) - assert result.returncode != 0 - assert "BOOTSTRAP_KEY" in result.stderr - - -def test_companion_tool_comes_from_the_registry_not_the_pinned_name( - monkeypatch, -) -> None: - """Regression: pinning --tool must not silently skip the denial check. - - ``--tool`` is mandatory off loopback. Selecting the companion from the - pinned list rather than the registry made the out-of-scope check skip on - exactly the deployments where it matters most, while the skip message - claimed only one tool existed. - """ - - select = _load_loop(monkeypatch)["select_companion_tool"] - registry = {"alpha": {}, "beta": {}, "gamma": {}} - - companion, error = select(registry, "alpha", None) - assert companion in {"beta", "gamma"} - assert not error - - # Pinning the permitted tool must not shrink the choice to nothing. - companion, error = select(registry, "gamma", None) - assert companion in {"alpha", "beta"} - assert not error - - -def test_companion_tool_must_be_registered_and_distinct(monkeypatch) -> None: - """Both ways this check could pass without proving permit scoping.""" - - select = _load_loop(monkeypatch)["select_companion_tool"] - registry = {"alpha": {}, "beta": {}} - - # An unregistered name yields tool-not-found, not a permit refusal. - companion, error = select(registry, "alpha", "does-not-exist") - assert companion is None - assert "tool-not-found" in error - - # A companion equal to the permitted tool would assert a denial that - # should never occur. - companion, error = select(registry, "alpha", "alpha") - assert companion is None - assert "differ" in error - - companion, error = select(registry, "alpha", "beta") - assert (companion, error) == ("beta", "") - - -def test_single_tool_deployment_skips_rather_than_false_passes( - monkeypatch, -) -> None: - select = _load_loop(monkeypatch)["select_companion_tool"] - assert select({"only": {}}, "only", None) == (None, "") - - -def test_advertised_price_is_read_from_the_manifest(monkeypatch) -> None: - """A fixed permit cap fails permit_budget_exceeded on a healthy deployment. - - Tool prices in this repo span 1–200 credits, so a cap chosen for one tool - refuses the golden invocation the moment another is pinned. - """ - - credits_per_call = _load_loop(monkeypatch)["credits_per_call"] - - assert credits_per_call({"annotations": {"creditsPerCall": 25.0}}) == 25.0 - assert credits_per_call({"annotations": {"creditsPerCall": "15"}}) == 15.0 - # A manifest entry with no stated price must not raise; the floor applies. - assert credits_per_call({}) == 0.0 - assert credits_per_call({"annotations": {}}) == 0.0 - assert credits_per_call({"annotations": {"creditsPerCall": "free"}}) == 0.0 - - -def test_off_loopback_runs_require_an_approved_payload() -> None: - """Schema validity is not evidence of safety. - - Derived arguments fill required fields from types, defaults, and the first - enum member — which for a consequential tool could be "delete", sent again - every interval. - """ - - result = subprocess.run( - [sys.executable, str(LOOP), "--once", "--tool", "some-tool"], - capture_output=True, - text=True, - env={ - "PATH": "/usr/bin:/bin", - "AGENT_MIDDLEWARE_API_URL": "https://api.thisisatest.tech", - "AGENT_MIDDLEWARE_API_KEY": "b2a_placeholder", - }, - ) - assert result.returncode == 2, result.stdout + result.stderr - assert "refusing to send derived arguments" in result.stderr - assert "b2a_placeholder" not in result.stdout + result.stderr - - # An explicit empty payload is a decision, and is accepted as one. - result = subprocess.run( - [ - sys.executable, - str(LOOP), - "--once", - "--tool", - "some-tool", - "--tool-args", - "{}", - ], - capture_output=True, - text=True, - env={ - "PATH": "/usr/bin:/bin", - "AGENT_MIDDLEWARE_API_URL": "http://127.0.0.1:1", - "AGENT_MIDDLEWARE_API_KEY": "b2a_placeholder", - }, - ) - assert "refusing to send derived arguments" not in result.stderr From 9b9ae02e2d9376f25cac0db4241145e40ba025d7 Mon Sep 17 00:00:00 2001 From: Claude Date: Thu, 20 Aug 2026 10:13:00 +0000 Subject: [PATCH 2/2] fix(scripts): a bad tool pin is a config error, not an invariant failure MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Tool-selection problems raised SmokeTestFailure, which exits 1 — the same signal a genuinely broken trust plane gives. A typo in $CI_SMOKE_TOOL, or a registry that happens to carry neither preferred tool, would page whoever is on call as though the product had failed. They are ConfigurationError now, exiting 2, matching how the off-loopback flag guards already behave. Writing the test for that surfaced a second gap: --other-tool was resolved inside the denial block, which only ran when the deployment exposed more than one tool. On a single-tool deployment an invalid pin was silently ignored, so an operator who misspelled it got a clean exit 0 and a SKIP line rather than being told. Selection now happens beside the governed tool, so the pin is validated regardless of how many tools exist, and the denial block simply asks whether a companion was found. That also removes the separate tool-count condition, which was expressing the same thing less directly. Also corrects the select_governed_tool docstring, which still described the "any registered tool" fallback removed earlier in this branch. A docstring promising a fallback the code deliberately refuses to implement is worse than no docstring on a decision this one turns on. Co-Authored-By: Claude Opus 5 Claude-Session: https://claude.ai/code/session_01Ri576RW3ocgqpx1xYN9vBm --- scripts/constant_test_loop.py | 105 ++++++++++++++++--------------- tests/test_constant_test_loop.py | 48 ++++++++++++++ 2 files changed, 104 insertions(+), 49 deletions(-) diff --git a/scripts/constant_test_loop.py b/scripts/constant_test_loop.py index 31a9c6e1..3bef059b 100755 --- a/scripts/constant_test_loop.py +++ b/scripts/constant_test_loop.py @@ -105,9 +105,11 @@ def select_governed_tool( """Choose the tool the permit will scope to. Returns ``(tool, error)``. A pinned name must exist; otherwise the - preferred production and dogfood tools are tried in order, and failing - those any registered tool is acceptable — the loop asserts trust-plane - behaviour, which does not depend on which tool is on the other end. + preferred production and dogfood tools are tried in order. Failing those, + an explicit ``--tool`` pin is required: there is deliberately no fallback + to an arbitrary registered tool, because a guessed tool can accept a + schema-valid payload and still refuse the action, which reads as a broken + trust plane rather than a bad choice of tool. """ if pinned: if pinned not in registry: @@ -378,13 +380,23 @@ def run_constant_test( for tool in tools.get("tools", []) if isinstance(tool, dict) and tool.get("name") } - tool_names = list(registry) - governed_tool, selection_error = select_governed_tool(registry, pinned_tool) if selection_error: - raise SmokeTestFailure(selection_error) + # ConfigurationError (exit 2), not SmokeTestFailure (exit 1). An + # unusable tool pin is the operator's problem; exiting 1 would + # page someone with the same signal a broken trust plane gives. + raise ConfigurationError(selection_error) print(f"[constant-test] using tool: {governed_tool}", file=sys.stderr) + # Resolved here rather than inside the denial block below: a typo in + # --other-tool must be reported even on a single-tool deployment, + # where that block never runs and the bad pin would be swallowed. + other_tool, companion_error = select_companion_tool( + registry, governed_tool, pinned_other_tool + ) + if companion_error: + raise ConfigurationError(companion_error) + # Sized from what this tool actually costs rather than a fixed cap. max_credits = max( PERMIT_MIN_CREDITS, @@ -505,53 +517,48 @@ def run_constant_test( file=sys.stderr, ) - # Out-of-scope tool denial check (optional) - # Only run if there are multiple tools available to test against. - # A single-tool environment exercises the core loop but can't test - # permit_tool_not_allowed without registering a second tool. - if len(tool_names) > 1: + # Out-of-scope tool denial check (optional). + # A single-tool deployment exercises the core loop but cannot test + # permit_tool_not_allowed: invoking a name the registry has never + # heard of proves "tool not found", not "the permit refused it". + if other_tool: print("[constant-test] attempting out-of-scope tool", file=sys.stderr) - other_tool, companion_error = select_companion_tool( - registry, governed_tool, pinned_other_tool + denial_body = _build_mcp_call( + request_id=f"constant-test-deny-{run_id}", + tool=other_tool, + wallet_id=wallet_id, + permit_id=permit_id, + idempotency_key=f"constant-test-deny-{run_id}", + arguments=arguments_for( + registry[other_tool].get("inputSchema") or {}, run_id + ), + ) + denial_call = _post_json( + client, "/mcp/messages", denial_body, expected_status=200 + ) + denial_error = _first_jsonrpc_error(denial_call) + require( + denial_error["message"] == "permit_tool_not_allowed", + f"expected permit_tool_not_allowed, got {denial_error['message']}", + ) + denial_receipt = denial_error["data"]["receipt"] + require( + denial_receipt["outcome"] == "denied", + "denial receipt outcome != denied", + ) + denial_charged = Decimal(str(denial_receipt["credits_charged"])) + require( + denial_charged == Decimal("0"), + f"denial charged {denial_charged} credits (expected 0)", + ) + print( + "[constant-test] denial OK: permit_tool_not_allowed, 0 credits charged", + file=sys.stderr, ) - if companion_error: - raise SmokeTestFailure(companion_error) - if other_tool: - denial_body = _build_mcp_call( - request_id=f"constant-test-deny-{run_id}", - tool=other_tool, - wallet_id=wallet_id, - permit_id=permit_id, - idempotency_key=f"constant-test-deny-{run_id}", - arguments=arguments_for( - registry[other_tool].get("inputSchema") or {}, run_id - ), - ) - denial_call = _post_json( - client, "/mcp/messages", denial_body, expected_status=200 - ) - denial_error = _first_jsonrpc_error(denial_call) - require( - denial_error["message"] == "permit_tool_not_allowed", - f"expected permit_tool_not_allowed, got {denial_error['message']}", - ) - denial_receipt = denial_error["data"]["receipt"] - require( - denial_receipt["outcome"] == "denied", - "denial receipt outcome != denied", - ) - denial_charged = Decimal(str(denial_receipt["credits_charged"])) - require( - denial_charged == Decimal("0"), - f"denial charged {denial_charged} credits (expected 0)", - ) - print( - "[constant-test] denial OK: permit_tool_not_allowed, 0 credits charged", - file=sys.stderr, - ) else: print( - "[constant-test] SKIPPED out-of-scope tool check (single tool only)", + "[constant-test] SKIPPED out-of-scope tool check " + "(no registered tool outside the permit)", file=sys.stderr, ) diff --git a/tests/test_constant_test_loop.py b/tests/test_constant_test_loop.py index f49beeec..f1c245ff 100644 --- a/tests/test_constant_test_loop.py +++ b/tests/test_constant_test_loop.py @@ -538,3 +538,51 @@ def test_bootstrap_key_only_pipes_without_jq(): ) assert result.returncode != 0 assert "BOOTSTRAP_KEY" in result.stderr + + +def test_unusable_tool_pin_exits_as_configuration_not_invariant_failure( + test_server, +): + """A bad pin must not page with the same signal as a broken trust plane. + + Exit 1 means an invariant failed; exit 2 means the loop was configured + wrong. Conflating them makes a typo in $CI_SMOKE_TOOL indistinguishable + from the product actually breaking. + """ + base_url, _ = test_server + env = {k: v for k, v in os.environ.items() if not k.startswith("CI_SMOKE")} + + result = subprocess.run( + [ + sys.executable, + str(CONSTANT_TEST_SCRIPT), + "--api-url", + base_url, + "--tool", + "no-such-tool", + ], + env=env, + capture_output=True, + text=True, + timeout=60, + ) + assert result.returncode == 2, result.stdout + result.stderr + assert "configuration error" in result.stderr + assert "FAILED" not in result.stderr + + result = subprocess.run( + [ + sys.executable, + str(CONSTANT_TEST_SCRIPT), + "--api-url", + base_url, + "--other-tool", + "no-such-tool", + ], + env=env, + capture_output=True, + text=True, + timeout=60, + ) + assert result.returncode == 2, result.stdout + result.stderr + assert "configuration error" in result.stderr