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..3bef059b 100755 --- a/scripts/constant_test_loop.py +++ b/scripts/constant_test_loop.py @@ -28,22 +28,141 @@ 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. 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: + 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 +295,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 +375,33 @@ 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") + } + governed_tool, selection_error = select_governed_tool(registry, pinned_tool) + if 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, + credits_per_call(registry[governed_tool]) * PERMIT_CREDIT_HEADROOM, + ) # Issue scoped permit print("[constant-test] issuing scoped permit", file=sys.stderr) @@ -281,7 +421,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 +437,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 @@ -372,48 +517,48 @@ def run_constant_test(api_url: str, agent_key: str, wallet_id: str, key_id: str) 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) - # Find a tool that's not the governed one - other_tool = next((t for t in tool_names if t != governed_tool), None) - 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={}, - ) - 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, - ) + 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, ) @@ -437,6 +582,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 +620,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..f1c245ff 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,273 @@ 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 + + +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 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