From b83bf108b044eceb06f7b6388578333d7efd879b Mon Sep 17 00:00:00 2001 From: Claude Date: Fri, 14 Aug 2026 06:37:30 +0000 Subject: [PATCH 1/6] feat(demo): one-command live loop proof with partner handoff bundle The live self-provision demo flow proved authenticate -> authorize -> invoke -> meter -> receipt -> replay but skipped three stages of the core loop: discovery, audit-chain verification, and the out-of-scope denial. Close those gaps and make the whole loop repeatable in one command: - scripts/live_loop_proof.py drives all eight stages over real HTTP against a running quickstart server as a self-provisioned non-admin caller, asserts every invariant (single debit, same-receipt replay, valid audit chain, signed zero-charge denial), verifies both the success and denial receipts offline with the SDK CLI, and writes a partner handoff bundle (portable receipts, public key set, VERIFY.md instructions, machine-readable transcript) to data/live-loop-proof/. - docs/quickstart.md gains the missing audit step: verify your wallet's tamper-evident audit chain with your own key via POST /v1/audit/verify-chain. - tests/test_quickstart_path.py guards both additions: the documented audit-chain step and the proof script end to end, including offline verification of the bundle it emits. - Makefile: `make live-loop-proof`; DEMO_SCRIPT.md points the live flow at it and at the partner-owned verification step the validation milestone asks for. Co-Authored-By: Claude Fable 5 Claude-Session: https://claude.ai/code/session_01MpjPj59M4n6boNmbs7CpGR --- .gitignore | 1 + DEMO_SCRIPT.md | 16 ++ Makefile | 11 +- docs/quickstart.md | 16 ++ scripts/live_loop_proof.py | 483 ++++++++++++++++++++++++++++++++++ tests/test_quickstart_path.py | 69 ++++- 6 files changed, 592 insertions(+), 4 deletions(-) create mode 100644 scripts/live_loop_proof.py diff --git a/.gitignore b/.gitignore index 56c50931..444f5ee3 100644 --- a/.gitignore +++ b/.gitignore @@ -46,6 +46,7 @@ data/dogfood_partner_notes.jsonl data/dogfood_trust_plane.db* data/demo_trust_plane.db* data/quickstart/ +data/live-loop-proof/ # OS .DS_Store diff --git a/DEMO_SCRIPT.md b/DEMO_SCRIPT.md index 629ec214..264f52d7 100644 --- a/DEMO_SCRIPT.md +++ b/DEMO_SCRIPT.md @@ -79,6 +79,22 @@ For machine-readable verification of the same war-room flow: make agent-ops-war-room-check ``` +For the full loop over **real HTTP** as a self-provisioned non-admin caller +— discover → authenticate → authorize → invoke → meter → receipt → replay → +audit-chain verification → out-of-scope denial, with both receipts verified +offline — plus a partner handoff bundle (portable receipts, public key set, +and independent verification instructions) written to +`data/live-loop-proof/`: + +```bash +make quickstart # terminal 1: boots the server +make live-loop-proof # terminal 2: drives the loop, writes the bundle +``` + +Handing that bundle's directory to a partner engineer, and having them run +the verifier themselves, is the partner-owned verification step the +customer-validation milestone asks for. + The proof artifact shape is captured in [`docs/demo-trust-plane-output.md`](docs/demo-trust-plane-output.md). Use the live demo flow below when walking a partner through the product story. diff --git a/Makefile b/Makefile index 47bd6064..a66b6726 100644 --- a/Makefile +++ b/Makefile @@ -1,4 +1,4 @@ -.PHONY: quickstart quickstart-check test test-all test-proof coverage prove-trust-plane prove-trust-plane-postgres prove-crash-recovery demo-trust-plane demo-trust-plane-check dogfood-trust-plane dogfood-trust-plane-check red-team-trust-plane red-team-trust-plane-check agent-ops-war-room agent-ops-war-room-check trust-coverage-gate trust-release-gate trust-conformance-live adversarial-battery-live railway-preflight railway-preflight-live +.PHONY: quickstart quickstart-check live-loop-proof test test-all test-proof coverage prove-trust-plane prove-trust-plane-postgres prove-crash-recovery demo-trust-plane demo-trust-plane-check dogfood-trust-plane dogfood-trust-plane-check red-team-trust-plane red-team-trust-plane-check agent-ops-war-room agent-ops-war-room-check trust-coverage-gate trust-release-gate trust-conformance-live adversarial-battery-live railway-preflight railway-preflight-live # The 15-minute golden path: boot a real local trust plane on loopback with # self-serve key minting and one invokable governed tool, then follow @@ -13,6 +13,15 @@ quickstart: quickstart-check: uv run --with-requirements requirements.txt pytest tests/test_quickstart_path.py -v +# One-command live proof against an already-running quickstart server +# (terminal 1: `make quickstart`). Drives discover -> authenticate -> +# authorize -> invoke -> meter -> receipt -> replay -> audit -> govern over +# real HTTP as a self-provisioned non-admin caller, verifies both the +# success and denial receipts offline, and writes a partner handoff bundle +# to data/live-loop-proof/. +live-loop-proof: + uv run --with-requirements requirements.txt python scripts/live_loop_proof.py $(LIVE_LOOP_PROOF_ARGS) + # Fast inner loop: trust-plane (product) tests only. Proof-surface workloads # are skipped here — run them with `make test-all` (what CI runs) or `make test-proof`. # `--with-requirements` makes these self-contained: uv installs the runtime + diff --git a/docs/quickstart.md b/docs/quickstart.md index 227838bf..96cdc3ab 100644 --- a/docs/quickstart.md +++ b/docs/quickstart.md @@ -224,6 +224,22 @@ Confirm the ledger agrees — one funding entry, one debit: curl -s "$API_URL/v1/billing/ledger/$WALLET_ID" -H "X-API-Key: $AGENT_API_KEY" | python3 -m json.tool ``` +Then verify the tamper-evident audit chain behind those entries. Every +governed action appends a hash-linked audit event, and your own key is +enough to verify your wallet's chain — tamper evidence you cannot use +yourself protects no one: + +```bash +curl -s -X POST "$API_URL/v1/audit/verify-chain" \ + -H "X-API-Key: $AGENT_API_KEY" \ + -H "Content-Type: application/json" \ + -d '{}' | python3 -m json.tool +``` + +`"valid": true` with a non-zero `checked_events` means the recorded history +hash-links end to end; an edited or deleted event would surface as +`"valid": false` with the broken link named in `broken_event_id`. + ## 7. Overspend on purpose (~2 minutes) Spend the rest of the permit — notes two and three succeed: diff --git a/scripts/live_loop_proof.py b/scripts/live_loop_proof.py new file mode 100644 index 00000000..9b91fb83 --- /dev/null +++ b/scripts/live_loop_proof.py @@ -0,0 +1,483 @@ +#!/usr/bin/env python3 +"""Drive the complete core loop over live HTTP and emit a partner handoff bundle. + +The quickstart walkthrough (docs/quickstart.md) proves each stage by hand; +this script proves all of them in one command, as the non-admin caller a +partner's agent would be, against an already-running quickstart-posture +server (``make quickstart``): + + discover -> authenticate -> authorize -> invoke -> meter -> receipt + -> replay -> audit -> govern + +Every stage is asserted, not just printed: the run exits non-zero the moment +any invariant breaks (wrong charge, replay minting a second receipt, audit +chain invalid, out-of-scope call not denied, offline verification failing). + +On success it writes a handoff bundle a partner engineer can verify with no +access to this server and no credential: + + receipt-bundle.json portable signed success receipt + denial-bundle.json portable signed out-of-scope denial receipt + trust-keys.json the issuer's published public key set + transcript.json machine-readable record of every stage + VERIFY.md instructions for independent offline verification + +Run:: + + make quickstart # terminal 1: boots the server + make live-loop-proof # terminal 2: drives the loop, writes the bundle + +Self-provision only exists on local dev postures — production-like +deployments refuse it at boot — so this script is loopback-only unless +``--allow-remote`` is passed for a staging host you control. +""" + +from __future__ import annotations + +import argparse +import json +import subprocess +import sys +import uuid +from datetime import datetime, timedelta, timezone +from pathlib import Path +from typing import Any +from urllib.parse import urlparse + +import httpx + +ROOT = Path(__file__).resolve().parents[1] +SDK_SRC = ROOT / "b2a_sdk" / "src" +DEFAULT_API_URL = "http://127.0.0.1:8000" +DEFAULT_OUTPUT_DIR = ROOT / "data" / "live-loop-proof" +GOVERNED_TOOL = "partner.notes.write" +UNGRANTED_TOOL = "some.other.tool" +PERMIT_MAX_CREDITS = 10 +LOOPBACK_HOSTS = {"127.0.0.1", "localhost", "::1"} + +VERIFY_MD = """\ +# Independent verification of this receipt bundle + +You are holding signed artifacts from an Agent Middleware API trust plane. +Nothing in this directory requires an account, an API key, or network access +to the issuing server. + +## What is here + +- `receipt-bundle.json` — a portable signed receipt for one governed, + metered tool invocation. +- `denial-bundle.json` — a portable signed receipt for a call the permit + did **not** allow, denied with zero credits charged. Denials are receipted + and signed exactly like successes. +- `trust-keys.json` — the issuer's published Ed25519 public key set + (`/.well-known/trust-keys.json`). Public keys only. +- `transcript.json` — the stage-by-stage record of the run that produced + these artifacts. + +## Verify offline (no network) + +From a clone of the repository: + + PYTHONPATH=b2a_sdk/src python -m b2a_sdk.verify_cli \\ + --bundle receipt-bundle.json --keys trust-keys.json + +Or with the SDK installed (`pip install ./b2a_sdk` from the repo, which +provides the `b2a-verify-receipt` entry point): + + b2a-verify-receipt --bundle receipt-bundle.json --keys trust-keys.json + +Expected: `VERIFIED`, exit code 0. Repeat with `denial-bundle.json`. + +Exit codes are meant to be branched on: 0 verified, 1 well-formed but does +not verify (treat as tampered), 2 undetermined (unknown key, malformed +input). Do not treat 2 as forgery. + +## Prove tampering is detected + +Edit one digit inside the `signing_input` field of `receipt-bundle.json` +(for example, change `"credits_charged":"2"` to `"0"`) and re-run the +verifier. It must exit 1. + +## Verify against the live issuer instead + +If you can reach the issuing server, you can fetch its key set yourself +rather than trusting the copy in this directory: + + b2a-verify-receipt --bundle receipt-bundle.json --issuer +""" + + +class StageFailure(RuntimeError): + pass + + +def require(condition: bool, message: str) -> None: + if not condition: + raise StageFailure(message) + + +class ProofRun: + def __init__(self, client: httpx.Client, output_dir: Path) -> None: + self.client = client + self.output_dir = output_dir + self.run_id = uuid.uuid4().hex[:12] + self.stages: list[dict[str, Any]] = [] + + def record(self, stage: str, **detail: Any) -> None: + self.stages.append({"stage": stage, "ok": True, **detail}) + summary = ", ".join(f"{k}={v}" for k, v in detail.items()) + print(f"[{len(self.stages)}] {stage:<12} {summary}") + + # -- stages ------------------------------------------------------------ + + def discover(self) -> dict[str, Any]: + agent_manifest = self.client.get("/.well-known/agent.json") + require(agent_manifest.status_code == 200, "agent.json not served") + tools = self.client.get("/mcp/tools.json") + require(tools.status_code == 200, "tools.json not served") + tool_names = [tool["name"] for tool in tools.json().get("tools", [])] + require( + GOVERNED_TOOL in tool_names, + f"{GOVERNED_TOOL} not discoverable; is the server running " + "the quickstart posture (ENABLE_DOGFOOD_TOOL=true)?", + ) + self.record("discover", tools=tool_names) + return tools.json() + + def authenticate(self) -> dict[str, str]: + minted = self.client.post( + "/v1/dev-keys/self-provision", + json={"agent_id": f"live-loop-proof-{self.run_id}"}, + ) + require( + minted.status_code in (200, 201), + f"self-provision refused ({minted.status_code}): {minted.text}", + ) + body = minted.json() + identity = { + "wallet_id": body["wallet_id"], + "key_id": body["key_id"], + } + self.client.headers["X-API-Key"] = body["api_key"] + self.record("authenticate", **identity) + return identity + + def authorize(self, identity: dict[str, str], *, allowed_tool: str, + label: str) -> str: + expires_at = ( + datetime.now(timezone.utc) + timedelta(minutes=30) + ).strftime("%Y-%m-%dT%H:%M:%SZ") + permit = self.client.post( + "/v1/permits", + headers={"Idempotency-Key": f"proof-{self.run_id}-{label}"}, + json={ + "issuer_wallet_id": identity["wallet_id"], + "subject_wallet_id": identity["wallet_id"], + "subject_key_id": identity["key_id"], + "allowed_tools": [allowed_tool], + "scopes": [f"tool:{allowed_tool}:invoke", "billing:charge"], + "max_credits": PERMIT_MAX_CREDITS, + "expires_at": expires_at, + }, + ) + require( + permit.status_code == 201, + f"permit creation failed ({permit.status_code}): {permit.text}", + ) + permit_id = permit.json()["permit_id"] + if label == "grant": + self.record( + "authorize", + permit_id=permit_id, + allowed_tools=[allowed_tool], + max_credits=PERMIT_MAX_CREDITS, + ) + return permit_id + + def _invoke_body(self, wallet_id: str, permit_id: str, + idempotency_key: str, call_id: str) -> dict[str, Any]: + return { + "jsonrpc": "2.0", + "id": call_id, + "method": "tools/call", + "params": { + "name": GOVERNED_TOOL, + "arguments": {"text": f"live loop proof {self.run_id}"}, + "mcpContext": { + "wallet_id": wallet_id, + "permit_id": permit_id, + "idempotency_key": idempotency_key, + }, + }, + } + + def invoke(self, wallet_id: str, permit_id: str) -> dict[str, Any]: + response = self.client.post( + "/mcp/messages", + json=self._invoke_body( + wallet_id, permit_id, f"proof-{self.run_id}-note-1", "call-1" + ), + ).json() + require( + "result" in response, + f"governed invoke failed: {json.dumps(response.get('error'))}", + ) + receipt = response["result"]["receipt"] + require(receipt["outcome"] == "success", "receipt outcome != success") + require(bool(receipt["signature"]), "receipt is unsigned") + require( + receipt["ledger_entry_id"] is not None, + "success receipt has no ledger entry", + ) + self.record( + "invoke", + receipt_id=receipt["receipt_id"], + outcome=receipt["outcome"], + ) + return receipt + + def meter(self, wallet_id: str, receipt: dict[str, Any]) -> int: + ledger = self.client.get(f"/v1/billing/ledger/{wallet_id}") + require(ledger.status_code == 200, "ledger read failed") + entries = ledger.json()["entries"] + matching = [ + e for e in entries if e["entry_id"] == receipt["ledger_entry_id"] + ] + require( + len(matching) == 1, + "receipt's ledger_entry_id not found in the wallet ledger", + ) + debits = [e for e in entries if GOVERNED_TOOL in e["description"]] + self.record( + "meter", + ledger_entry_id=matching[0]["entry_id"], + amount=matching[0]["amount"], + balance_after=matching[0]["balance_after"], + ) + return len(debits) + + def receipt_bundle(self, receipt_id: str, filename: str) -> Path: + portable = self.client.get(f"/v1/receipts/{receipt_id}/portable") + require( + portable.status_code == 200, + f"portable receipt export failed ({portable.status_code})", + ) + path = self.output_dir / filename + path.write_text(portable.text, encoding="utf-8") + return path + + def trust_keys(self) -> Path: + # Deliberately unauthenticated: the key set must be fetchable by a + # verifier who holds no credential at all. + keys = httpx.get( + f"{self.client.base_url}/.well-known/trust-keys.json", timeout=30 + ) + require(keys.status_code == 200, "trust-keys.json not served") + path = self.output_dir / "trust-keys.json" + path.write_text(keys.text, encoding="utf-8") + return path + + def replay(self, wallet_id: str, permit_id: str, + receipt: dict[str, Any], debits_before: int) -> None: + replayed = self.client.post( + "/mcp/messages", + json=self._invoke_body( + wallet_id, permit_id, f"proof-{self.run_id}-note-1", "call-1" + ), + ).json() + require("result" in replayed, "replay was rejected instead of replayed") + require( + replayed["result"]["receipt"]["receipt_id"] == receipt["receipt_id"], + "replay returned a different receipt", + ) + ledger = self.client.get(f"/v1/billing/ledger/{wallet_id}").json() + debits_after = len( + [e for e in ledger["entries"] if GOVERNED_TOOL in e["description"]] + ) + require(debits_after == debits_before, "replay produced a second debit") + self.record( + "replay", + receipt_id=receipt["receipt_id"], + second_debit="none", + ) + + def audit(self) -> None: + # A wallet-scoped key verifies its own chain; no operator needed. + verdict = self.client.post("/v1/audit/verify-chain", json={}) + require( + verdict.status_code == 200, + f"audit chain verification failed ({verdict.status_code}): " + f"{verdict.text}", + ) + chain = verdict.json() + require( + chain["valid"] is True, + f"audit chain INVALID: reason={chain.get('reason')} " + f"broken_event_id={chain.get('broken_event_id')}", + ) + require(chain["checked_events"] > 0, "audit chain verified zero events") + self.record("audit", valid=True, checked_events=chain["checked_events"]) + + def govern(self, identity: dict[str, str]) -> dict[str, Any]: + narrow_permit = self.authorize( + identity, allowed_tool=UNGRANTED_TOOL, label="deny" + ) + denied = self.client.post( + "/mcp/messages", + json=self._invoke_body( + identity["wallet_id"], + narrow_permit, + f"proof-{self.run_id}-deny-1", + "deny-1", + ), + ).json() + require( + "error" in denied, + "out-of-scope invoke was NOT denied — governance failure", + ) + require( + denied["error"]["message"] == "permit_tool_not_allowed", + f"expected permit_tool_not_allowed, got {denied['error']['message']}", + ) + denial = denied["error"]["data"]["receipt"] + require(denial["outcome"] == "denied", "denial receipt outcome != denied") + require(denial["credits_charged"] == "0", "denial was charged") + require(bool(denial["signature"]), "denial receipt is unsigned") + self.record( + "govern", + denied="permit_tool_not_allowed", + denial_receipt=denial["receipt_id"], + credits_charged=denial["credits_charged"], + ) + return denial + + def verify_offline(self, bundle_path: Path, keys_path: Path, + label: str) -> str: + result = subprocess.run( + [ + sys.executable, + "-m", + "b2a_sdk.verify_cli", + "--bundle", + str(bundle_path), + "--keys", + str(keys_path), + ], + env={"PYTHONPATH": str(SDK_SRC)}, + capture_output=True, + text=True, + timeout=60, + ) + require( + result.returncode == 0, + f"offline verification of {bundle_path.name} failed " + f"(exit {result.returncode}): {result.stdout}{result.stderr}", + ) + self.record(f"verify:{label}", bundle=bundle_path.name, exit_code=0) + return result.stdout + + +def run(api_url: str, output_dir: Path) -> None: + output_dir.mkdir(parents=True, exist_ok=True) + with httpx.Client(base_url=api_url, timeout=30) as client: + proof = ProofRun(client, output_dir) + print(f"live loop proof {proof.run_id} against {api_url}\n") + + proof.discover() + identity = proof.authenticate() + permit_id = proof.authorize( + identity, allowed_tool=GOVERNED_TOOL, label="grant" + ) + receipt = proof.invoke(identity["wallet_id"], permit_id) + debits = proof.meter(identity["wallet_id"], receipt) + + receipt_path = proof.receipt_bundle( + receipt["receipt_id"], "receipt-bundle.json" + ) + keys_path = proof.trust_keys() + proof.record( + "receipt", + bundle=receipt_path.name, + signature_key_id=receipt.get("signature_key_id"), + ) + + proof.replay(identity["wallet_id"], permit_id, receipt, debits) + proof.audit() + denial = proof.govern(identity) + denial_path = proof.receipt_bundle( + denial["receipt_id"], "denial-bundle.json" + ) + + success_out = proof.verify_offline(receipt_path, keys_path, "success") + proof.verify_offline(denial_path, keys_path, "denial") + + (output_dir / "VERIFY.md").write_text(VERIFY_MD, encoding="utf-8") + (output_dir / "transcript.json").write_text( + json.dumps( + { + "run_id": proof.run_id, + "api_url": api_url, + "completed_at": datetime.now(timezone.utc).isoformat(), + "stages": proof.stages, + }, + indent=2, + ) + + "\n", + encoding="utf-8", + ) + + print(f"\n{success_out.rstrip()}") + print(f"\nAll stages held. Handoff bundle: {output_dir}") + print("Hand the directory to a partner engineer; VERIFY.md tells them") + print("how to verify both receipts offline with no credential.") + + +def main(argv: list[str] | None = None) -> int: + parser = argparse.ArgumentParser(description=__doc__.splitlines()[0]) + parser.add_argument( + "--api-url", + default=DEFAULT_API_URL, + help=f"Running quickstart-posture server (default: {DEFAULT_API_URL}).", + ) + parser.add_argument( + "--output-dir", + type=Path, + default=DEFAULT_OUTPUT_DIR, + help=f"Where to write the handoff bundle (default: {DEFAULT_OUTPUT_DIR}).", + ) + parser.add_argument( + "--allow-remote", + action="store_true", + help=( + "Permit a non-loopback --api-url. Only for a staging host you " + "control; the run self-provisions keys and writes test data." + ), + ) + args = parser.parse_args(argv) + + host = urlparse(args.api_url).hostname or "" + if host not in LOOPBACK_HOSTS and not args.allow_remote: + print( + f"refusing non-loopback target {args.api_url!r} without " + "--allow-remote", + file=sys.stderr, + ) + return 2 + + try: + run(args.api_url.rstrip("/"), args.output_dir) + except StageFailure as failure: + print(f"\nSTAGE FAILED: {failure}", file=sys.stderr) + return 1 + except httpx.HTTPError as network_error: + print( + f"\nCould not reach {args.api_url}: {network_error}\n" + "Is the server running? Start it with: make quickstart", + file=sys.stderr, + ) + return 2 + return 0 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/tests/test_quickstart_path.py b/tests/test_quickstart_path.py index 23c233b3..83fffc72 100644 --- a/tests/test_quickstart_path.py +++ b/tests/test_quickstart_path.py @@ -5,9 +5,9 @@ then drives every step of the quickstart page over HTTP: self-provision a key, self-issue a permit, invoke the governed tool, replay it, reuse the idempotency key with a changed payload, overspend the permit on the call -the doc predicts, verify the receipt offline with the SDK CLI, forge it the -way the doc forges it, and get denied by a permit that does not allow the -tool. +the doc predicts, verify the wallet's audit chain with its own key, verify +the receipt offline with the SDK CLI, forge it the way the doc forges it, +and get denied by a permit that does not allow the tool. If docs/quickstart.md and the code disagree, this test is what breaks. Assertions deliberately pin the *documented* observables — error strings, @@ -216,6 +216,12 @@ def test_documented_quickstart_path(quickstart_server, tmp_path): debits = [e for e in ledger["entries"] if GOVERNED_TOOL in e["description"]] assert len(debits) == 3 + # ... and the tamper-evident audit chain behind those entries verifies + # with the wallet's own key — no operator credential required. + chain = client.post("/v1/audit/verify-chain", headers=auth, json={}).json() + assert chain["valid"] is True + assert chain["checked_events"] > 0 + # Step 8 — offline verification, exactly as documented: fetch the # portable bundle and the unauthenticated key set, run the SDK CLI. bundle_path = tmp_path / "receipt-bundle.json" @@ -284,3 +290,60 @@ def test_documented_quickstart_path(quickstart_server, tmp_path): assert "VERIFIED" in denial_verified.stdout client.close() + + +def test_live_loop_proof_script(quickstart_server, tmp_path): + """The one-command live proof drives every stage and writes the bundle. + + scripts/live_loop_proof.py is the scripted form of the walkthrough above; + this keeps the two from drifting apart. + """ + output_dir = tmp_path / "handoff" + result = subprocess.run( + [ + sys.executable, + str(REPO_ROOT / "scripts" / "live_loop_proof.py"), + "--api-url", + quickstart_server, + "--output-dir", + str(output_dir), + ], + cwd=REPO_ROOT, + env={**os.environ, "PYTHONPATH": str(REPO_ROOT)}, + capture_output=True, + text=True, + timeout=180, + ) + assert result.returncode == 0, result.stdout + result.stderr + + for artifact in ( + "receipt-bundle.json", + "denial-bundle.json", + "trust-keys.json", + "transcript.json", + "VERIFY.md", + ): + assert (output_dir / artifact).exists(), f"missing {artifact}" + + transcript = json.loads( + (output_dir / "transcript.json").read_text(encoding="utf-8") + ) + assert [stage["stage"] for stage in transcript["stages"]] == [ + "discover", + "authenticate", + "authorize", + "invoke", + "meter", + "receipt", + "replay", + "audit", + "govern", + "verify:success", + "verify:denial", + ] + + # The bundle must be independently verifiable exactly as VERIFY.md + # instructs a partner engineer to do it. + for bundle in ("receipt-bundle.json", "denial-bundle.json"): + verified = _verify_cli(output_dir / bundle, output_dir / "trust-keys.json") + assert verified.returncode == 0, verified.stdout + verified.stderr From 50c01e8975b2759804a340bf8aee42ee37e33b77 Mon Sep 17 00:00:00 2001 From: Claude Date: Fri, 14 Aug 2026 06:58:00 +0000 Subject: [PATCH 2/6] fix(sdk): offline receipt verification must not require httpx The partner handoff bundle's VERIFY.md tells a verifier to run `python -m b2a_sdk.verify_cli --keys ...` with only `cryptography` installed -- the whole point of the offline path is no network and no network library. But importing the b2a_sdk package eagerly imported the httpx-backed HTTP client, so that command died at import time with ModuleNotFoundError: httpx even though the verifier never touches httpx. The earlier live run only passed because it went through a uv env that happened to have httpx present. Root cause: b2a_sdk/__init__.py imported .client (-> httpx) at package load. The offline verifier (receipt_verifier) needs only the stdlib plus cryptography, per its own docstring. Fix: load the httpx-dependent surface (client, edge_client, decorators) lazily via PEP 562 __getattr__, keeping errors/models/receipt_verifier eager. `import b2a_sdk`, `from b2a_sdk import verify_bundle`, and `python -m b2a_sdk.verify_cli --keys ...` now run with no networking library installed; B2AClient and the decorators still import on first access (raising the normal httpx ImportError only if httpx is truly needed and absent). Public API and __all__ are unchanged. - b2a_sdk/tests/test_offline_verify_no_httpx.py: regression guard that imports the package and runs the verifier CLI in a subprocess with httpx forced unavailable; the real-signature case skips when cryptography is absent (matching the bare-wheel CI job) and the lazy client import is re-checked when httpx is present. - scripts/live_loop_proof.py VERIFY.md: name the single real dependency (cryptography) and note the --issuer path additionally needs httpx. - docs/quickstart.md: tighten the audit-chain claim to "every governed tool invocation appends a hash-linked audit event" (permit issuance and self-provision do not), matching what verify-chain actually counts. - verify_cli.py: correct the stale "httpx is a base dependency" comment. Verified: dev env (cryptography present) 57 SDK tests pass; CI-faithful bare wheel (httpx present, cryptography absent) 55 pass + 2 skip; the real handoff-bundle receipts verify offline with httpx blocked. Co-Authored-By: Claude Fable 5 Claude-Session: https://claude.ai/code/session_01MpjPj59M4n6boNmbs7CpGR --- b2a_sdk/src/b2a_sdk/__init__.py | 51 ++++++- b2a_sdk/src/b2a_sdk/verify_cli.py | 2 +- b2a_sdk/tests/test_offline_verify_no_httpx.py | 140 ++++++++++++++++++ docs/quickstart.md | 7 +- scripts/live_loop_proof.py | 11 +- 5 files changed, 200 insertions(+), 11 deletions(-) create mode 100644 b2a_sdk/tests/test_offline_verify_no_httpx.py diff --git a/b2a_sdk/src/b2a_sdk/__init__.py b/b2a_sdk/src/b2a_sdk/__init__.py index 05a6f615..d4c34f81 100644 --- a/b2a_sdk/src/b2a_sdk/__init__.py +++ b/b2a_sdk/src/b2a_sdk/__init__.py @@ -1,8 +1,15 @@ """Python client for governed Agent Middleware tool calls.""" -from .client import AgentMiddlewareClient, B2AClient -from .decorators import billable, combined, monitored -from .edge_client import B2AEdgeClient +from __future__ import annotations + +from typing import TYPE_CHECKING + +# Dependency-minimal surface, imported eagerly: errors, models, and the offline +# receipt verifier need only the standard library plus ``cryptography`` (loaded +# lazily inside the verifier itself). Keeping these free of the HTTP stack is +# what lets ``python -m b2a_sdk.verify_cli --keys ...`` and +# ``from b2a_sdk import verify_bundle`` run with no networking library +# installed — the whole point of offline receipt verification. from .errors import ( AgentMiddlewareError, APIError, @@ -32,6 +39,44 @@ ) __version__ = "0.4.0" + +# The HTTP client surface (client, edge_client, and the decorators that wrap it) +# pulls in httpx. Load those names lazily via PEP 562 so importing the package — +# or running the offline verifier — does not require httpx to be installed. +# Accessing one of these names imports its backing module on first use, which +# still raises the normal ImportError for httpx only when httpx is genuinely +# needed and absent. +_LAZY_ATTRS = { + "AgentMiddlewareClient": "client", + "B2AClient": "client", + "B2AEdgeClient": "edge_client", + "billable": "decorators", + "combined": "decorators", + "monitored": "decorators", +} + +if TYPE_CHECKING: # let type checkers and IDEs resolve the lazy names statically + from .client import AgentMiddlewareClient, B2AClient + from .decorators import billable, combined, monitored + from .edge_client import B2AEdgeClient + + +def __getattr__(name: str) -> object: + module_name = _LAZY_ATTRS.get(name) + if module_name is None: + raise AttributeError(f"module {__name__!r} has no attribute {name!r}") + import importlib + + module = importlib.import_module(f".{module_name}", __name__) + value = getattr(module, name) + globals()[name] = value # cache so later access is a plain attribute lookup + return value + + +def __dir__() -> list[str]: + return sorted(__all__) + + __all__ = [ "APIError", "AgentMiddlewareClient", diff --git a/b2a_sdk/src/b2a_sdk/verify_cli.py b/b2a_sdk/src/b2a_sdk/verify_cli.py index bb38b146..26cb46a3 100644 --- a/b2a_sdk/src/b2a_sdk/verify_cli.py +++ b/b2a_sdk/src/b2a_sdk/verify_cli.py @@ -54,7 +54,7 @@ def _read_json(path: str) -> Any: def _fetch_key_document(issuer: str, timeout: float) -> Any: try: import httpx - except ImportError as exc: # pragma: no cover - httpx is a base dependency + except ImportError as exc: # httpx is only needed for the --issuer fetch raise VerificationError( "Fetching a key set requires httpx; pass --keys to verify offline." ) from exc diff --git a/b2a_sdk/tests/test_offline_verify_no_httpx.py b/b2a_sdk/tests/test_offline_verify_no_httpx.py new file mode 100644 index 00000000..be10f518 --- /dev/null +++ b/b2a_sdk/tests/test_offline_verify_no_httpx.py @@ -0,0 +1,140 @@ +"""The offline receipt verifier must work with only ``cryptography`` installed. + +The ``--keys`` verification path is the trust plane's promise that anyone can +check a signed receipt with no network access and no account. That promise is +only real if the code path also carries no *network dependency*: a partner who +``pip install cryptography`` and runs the verifier must not hit a +``ModuleNotFoundError`` for httpx. + +Regression guard: importing ``b2a_sdk`` used to eagerly import the httpx-backed +HTTP client, so ``python -m b2a_sdk.verify_cli --keys ...`` died at import time +when httpx was absent — even though the verifier itself never touches httpx. +These tests run the offline path in a subprocess with httpx forced unavailable. + +Collection must not require ``cryptography`` (the SDK wheel job installs the +bare wheel), so the signing helper imports it lazily and the real-verification +test skips when it is absent, matching ``test_receipt_verifier.py``. +""" + +from __future__ import annotations + +import base64 +import json +import os +import subprocess +import sys +from pathlib import Path + +import pytest + +SDK_SRC = Path(__file__).resolve().parents[1] / "src" + +# A preamble that makes ``import httpx`` fail even when httpx is installed, so +# the test proves independence from httpx on any machine (the wheel job has +# httpx present as a base dependency). +_BLOCK_HTTPX = "import sys; sys.modules['httpx'] = None\n" + + +def _run(code: str, *, block_httpx: bool) -> subprocess.CompletedProcess: + # Merge, don't replace, the environment: a bare ``env={...}`` would drop the + # interpreter's own paths and break site-packages resolution. + return subprocess.run( + [sys.executable, "-c", (_BLOCK_HTTPX if block_httpx else "") + code], + env={**os.environ, "PYTHONPATH": str(SDK_SRC)}, + capture_output=True, + text=True, + timeout=60, + ) + + +def _write_signed_bundle(tmp_path: Path) -> tuple[Path, Path]: + """Produce a genuinely-signed bundle and its public key document. + + Signs with ``cryptography`` only — the same dependency the verifier uses — + so the fixture needs nothing the offline path itself would not have. + """ + from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey + + private_key = Ed25519PrivateKey.generate() + public_raw = private_key.public_key().public_bytes_raw() + kid = "offline-test-ed25519" + + signing_input = json.dumps( + {"kid": kid, "outcome": "success", "receipt_id": "rcpt-offline-test"}, + separators=(",", ":"), + sort_keys=True, + ) + signature = private_key.sign(signing_input.encode("utf-8")) + + bundle = { + "schema_version": "1.0", + "receipt_id": "rcpt-offline-test", + "alg": "Ed25519", + "kid": kid, + "canonicalization": "awi-canonical-json/1", + "signing_input": signing_input, + "signature": base64.b64encode(signature).decode(), + } + key_doc = { + "schema_version": "1.0", + "alg": "Ed25519", + "keys": [ + { + "kid": kid, + "alg": "Ed25519", + "status": "active", + "public_key_b64": base64.b64encode(public_raw).decode(), + } + ], + } + + bundle_path = tmp_path / "receipt-bundle.json" + keys_path = tmp_path / "trust-keys.json" + bundle_path.write_text(json.dumps(bundle), encoding="utf-8") + keys_path.write_text(json.dumps(key_doc), encoding="utf-8") + return bundle_path, keys_path + + +def test_package_and_verifier_import_without_httpx(): + # Import only — needs neither httpx nor cryptography. This is the exact + # step that used to fail: importing b2a_sdk pulled in the httpx client. + result = _run( + "import b2a_sdk\n" + "from b2a_sdk import verify_bundle, key_set_from_document\n" + "import b2a_sdk.verify_cli\n" + "print('ok')\n", + block_httpx=True, + ) + assert result.returncode == 0, result.stdout + result.stderr + assert "ok" in result.stdout + + +def test_verify_cli_keys_path_runs_without_httpx(tmp_path): + pytest.importorskip("cryptography") + bundle_path, keys_path = _write_signed_bundle(tmp_path) + # Drive the CLI's main() exactly as `python -m b2a_sdk.verify_cli` would, + # which imports the b2a_sdk package first — the step that used to fail. + result = _run( + "from b2a_sdk.verify_cli import main\n" + f"raise SystemExit(main(['--bundle', {str(bundle_path)!r}, " + f"'--keys', {str(keys_path)!r}]))\n", + block_httpx=True, + ) + assert result.returncode == 0, result.stdout + result.stderr + assert "VERIFIED" in result.stdout + + +def test_lazy_http_client_still_importable_when_httpx_present(): + # The lazy PEP 562 surface must not have broken the real client import. + pytest.importorskip("httpx") + result = _run( + "import b2a_sdk\n" + "assert b2a_sdk.B2AClient is not None\n" + "assert b2a_sdk.billable is not None\n" + "from b2a_sdk import AgentMiddlewareClient\n" + "assert AgentMiddlewareClient is not None\n" + "print('client-ok')\n", + block_httpx=False, + ) + assert result.returncode == 0, result.stdout + result.stderr + assert "client-ok" in result.stdout diff --git a/docs/quickstart.md b/docs/quickstart.md index 96cdc3ab..4d00ddd4 100644 --- a/docs/quickstart.md +++ b/docs/quickstart.md @@ -225,9 +225,10 @@ curl -s "$API_URL/v1/billing/ledger/$WALLET_ID" -H "X-API-Key: $AGENT_API_KEY" | ``` Then verify the tamper-evident audit chain behind those entries. Every -governed action appends a hash-linked audit event, and your own key is -enough to verify your wallet's chain — tamper evidence you cannot use -yourself protects no one: +governed tool invocation — the successful call you just made, and the +denials you are about to trigger — appends a hash-linked audit event, and +your own key is enough to verify your wallet's chain (tamper evidence you +cannot use yourself protects no one): ```bash curl -s -X POST "$API_URL/v1/audit/verify-chain" \ diff --git a/scripts/live_loop_proof.py b/scripts/live_loop_proof.py index 9b91fb83..690b9fe5 100644 --- a/scripts/live_loop_proof.py +++ b/scripts/live_loop_proof.py @@ -76,13 +76,15 @@ ## Verify offline (no network) -From a clone of the repository: +The only dependency is `cryptography` for the Ed25519 check — no networking +library, no account. From a clone of the repository: + pip install cryptography PYTHONPATH=b2a_sdk/src python -m b2a_sdk.verify_cli \\ --bundle receipt-bundle.json --keys trust-keys.json -Or with the SDK installed (`pip install ./b2a_sdk` from the repo, which -provides the `b2a-verify-receipt` entry point): +Or with the SDK installed (`pip install "./b2a_sdk[verify]"` from the repo, +which provides the `b2a-verify-receipt` entry point and the verifier extra): b2a-verify-receipt --bundle receipt-bundle.json --keys trust-keys.json @@ -101,7 +103,8 @@ ## Verify against the live issuer instead If you can reach the issuing server, you can fetch its key set yourself -rather than trusting the copy in this directory: +rather than trusting the copy in this directory (this path fetches over the +network, so it also needs `httpx`: `pip install "./b2a_sdk[verify]" httpx`): b2a-verify-receipt --bundle receipt-bundle.json --issuer """ From a22cf32c6d12e00d85bb569baf5d05d9d56a8791 Mon Sep 17 00:00:00 2001 From: Claude Date: Fri, 14 Aug 2026 07:03:31 +0000 Subject: [PATCH 3/6] test(demo): make the live-loop proof assert what it claims to prove MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Adversarial review of the proof script found stages that could green against a broken server. Close the gaps so the proof fails when the invariant it demonstrates is actually violated: - meter integrity: invoke() now asserts the success receipt charged the known tool price (2 credits), and meter() asserts the ledger debit equals -2 and agrees with the receipt's credits_charged. Previously a server that metered 0 (revenue leak) or a wrong amount still passed the meter stage — the exact billing-integrity property the stage exists to prove went unchecked. - receipt semantics: receipt_bundle() parses each exported bundle and asserts its signed outcome and credits match the stage (success/2, denied/0). Offline signature verification proves the bytes are authentic, not that they carry the right outcome; a validly-signed denial exported as a success would otherwise ship in the handoff bundle unnoticed. The CI guard asserts the same on the emitted files. - secret hygiene: the transcript in the handoff bundle now records a sanitized api_url with any credential (userinfo) and query/fragment stripped, so a secret embedded in --api-url for a staging host cannot leak into an artifact meant for an external partner. - verify_offline() inherits the environment ({**os.environ, ...}) instead of replacing it, matching the repo's other subprocess verifier calls. Verified: tests/test_quickstart_path.py (both the documented walkthrough and the end-to-end script guard) pass; the URL sanitizer strips the userinfo/query exploit case; ruff clean. Co-Authored-By: Claude Fable 5 Claude-Session: https://claude.ai/code/session_01MpjPj59M4n6boNmbs7CpGR --- scripts/live_loop_proof.py | 81 ++++++++++++++++++++++++++++++++--- tests/test_quickstart_path.py | 20 +++++++++ 2 files changed, 95 insertions(+), 6 deletions(-) diff --git a/scripts/live_loop_proof.py b/scripts/live_loop_proof.py index 690b9fe5..c8184e38 100644 --- a/scripts/live_loop_proof.py +++ b/scripts/live_loop_proof.py @@ -36,13 +36,15 @@ import argparse import json +import os import subprocess import sys import uuid from datetime import datetime, timedelta, timezone +from decimal import Decimal from pathlib import Path from typing import Any -from urllib.parse import urlparse +from urllib.parse import urlparse, urlunparse import httpx @@ -53,8 +55,27 @@ GOVERNED_TOOL = "partner.notes.write" UNGRANTED_TOOL = "some.other.tool" PERMIT_MAX_CREDITS = 10 +# The quickstart dogfood tool bills a fixed 2 credits/call +# (DOGFOOD_CREDITS_PER_UNIT in app/services/dogfood_tool.py). Pinning it here +# lets the meter stage prove the charge is not just present but *correct* — a +# server that regressed metering to 0 (revenue leak) must fail, not pass. +EXPECTED_CREDITS_PER_CALL = Decimal("2") LOOPBACK_HOSTS = {"127.0.0.1", "localhost", "::1"} + +def _sanitize_url_for_record(url: str) -> str: + """Drop any embedded credential (userinfo) and query/fragment from a URL. + + The transcript ships inside the partner handoff bundle, so a secret carried + in ``--api-url`` (basic-auth userinfo or a token query param) must never be + persisted there. Keep only scheme, host, port, and path. + """ + parts = urlparse(url) + netloc = parts.hostname or "" + if parts.port is not None: + netloc = f"{netloc}:{parts.port}" + return urlunparse((parts.scheme, netloc, parts.path, "", "", "")) + VERIFY_MD = """\ # Independent verification of this receipt bundle @@ -232,10 +253,21 @@ def invoke(self, wallet_id: str, permit_id: str) -> dict[str, Any]: receipt["ledger_entry_id"] is not None, "success receipt has no ledger entry", ) + # A metered call must actually charge the known price. Without this a + # server that regressed to charging 0 (revenue leak) or a wrong amount + # would still green the loop -- defeating the billing-integrity claim + # the meter stage exists to prove. + charged = Decimal(str(receipt["credits_charged"])) + require( + charged == EXPECTED_CREDITS_PER_CALL, + f"success charged {charged} credits, expected " + f"{EXPECTED_CREDITS_PER_CALL}", + ) self.record( "invoke", receipt_id=receipt["receipt_id"], outcome=receipt["outcome"], + credits_charged=str(charged), ) return receipt @@ -250,6 +282,20 @@ def meter(self, wallet_id: str, receipt: dict[str, Any]) -> int: len(matching) == 1, "receipt's ledger_entry_id not found in the wallet ledger", ) + # The ledger debit must agree with the receipt's charge, in the right + # direction and magnitude -- the receipt and the money must tell the + # same story, or metering integrity is broken. + debit_amount = Decimal(str(matching[0]["amount"])) + require( + debit_amount == -EXPECTED_CREDITS_PER_CALL, + f"ledger debit {debit_amount} != expected " + f"{-EXPECTED_CREDITS_PER_CALL}", + ) + require( + -debit_amount == Decimal(str(receipt["credits_charged"])), + "ledger debit magnitude disagrees with the receipt's " + "credits_charged", + ) debits = [e for e in entries if GOVERNED_TOOL in e["description"]] self.record( "meter", @@ -259,7 +305,9 @@ def meter(self, wallet_id: str, receipt: dict[str, Any]) -> int: ) return len(debits) - def receipt_bundle(self, receipt_id: str, filename: str) -> Path: + def receipt_bundle(self, receipt_id: str, filename: str, *, + expected_outcome: str, + expected_credits: Decimal) -> Path: portable = self.client.get(f"/v1/receipts/{receipt_id}/portable") require( portable.status_code == 200, @@ -267,6 +315,21 @@ def receipt_bundle(self, receipt_id: str, filename: str) -> Path: ) path = self.output_dir / filename path.write_text(portable.text, encoding="utf-8") + # Verify the *exported* bundle says what this stage claims. Offline + # signature verification proves the bytes are authentic, not that they + # carry the right outcome; a validly-signed denial exported as a + # success would otherwise ship in the handoff bundle unnoticed. + signed = json.loads(json.loads(path.read_text(encoding="utf-8"))["signing_input"]) + require( + signed.get("outcome") == expected_outcome, + f"{filename}: exported outcome {signed.get('outcome')!r} != " + f"{expected_outcome!r}", + ) + require( + Decimal(str(signed.get("credits_charged"))) == expected_credits, + f"{filename}: exported credits_charged " + f"{signed.get('credits_charged')!r} != {expected_credits}", + ) return path def trust_keys(self) -> Path: @@ -366,7 +429,9 @@ def verify_offline(self, bundle_path: Path, keys_path: Path, "--keys", str(keys_path), ], - env={"PYTHONPATH": str(SDK_SRC)}, + # Merge, don't replace, the environment: a bare env={...} would + # drop the interpreter's own paths and can break the subprocess. + env={**os.environ, "PYTHONPATH": str(SDK_SRC)}, capture_output=True, text=True, timeout=60, @@ -395,7 +460,9 @@ def run(api_url: str, output_dir: Path) -> None: debits = proof.meter(identity["wallet_id"], receipt) receipt_path = proof.receipt_bundle( - receipt["receipt_id"], "receipt-bundle.json" + receipt["receipt_id"], "receipt-bundle.json", + expected_outcome="success", + expected_credits=EXPECTED_CREDITS_PER_CALL, ) keys_path = proof.trust_keys() proof.record( @@ -408,7 +475,9 @@ def run(api_url: str, output_dir: Path) -> None: proof.audit() denial = proof.govern(identity) denial_path = proof.receipt_bundle( - denial["receipt_id"], "denial-bundle.json" + denial["receipt_id"], "denial-bundle.json", + expected_outcome="denied", + expected_credits=Decimal("0"), ) success_out = proof.verify_offline(receipt_path, keys_path, "success") @@ -419,7 +488,7 @@ def run(api_url: str, output_dir: Path) -> None: json.dumps( { "run_id": proof.run_id, - "api_url": api_url, + "api_url": _sanitize_url_for_record(api_url), "completed_at": datetime.now(timezone.utc).isoformat(), "stages": proof.stages, }, diff --git a/tests/test_quickstart_path.py b/tests/test_quickstart_path.py index 83fffc72..734263f8 100644 --- a/tests/test_quickstart_path.py +++ b/tests/test_quickstart_path.py @@ -347,3 +347,23 @@ def test_live_loop_proof_script(quickstart_server, tmp_path): for bundle in ("receipt-bundle.json", "denial-bundle.json"): verified = _verify_cli(output_dir / bundle, output_dir / "trust-keys.json") assert verified.returncode == 0, verified.stdout + verified.stderr + + # Signature validity alone does not prove a bundle carries the right + # outcome. Assert the *semantics* of each exported bundle so a validly + # signed but wrong-outcome receipt (e.g. a denial exported as a success) + # cannot ship in the handoff unnoticed. + def _signed(bundle_name: str) -> dict: + outer = json.loads((output_dir / bundle_name).read_text(encoding="utf-8")) + return json.loads(outer["signing_input"]) + + success_signed = _signed("receipt-bundle.json") + assert success_signed["outcome"] == "success" + assert success_signed["credits_charged"] == "2" + + denial_signed = _signed("denial-bundle.json") + assert denial_signed["outcome"] == "denied" + assert denial_signed["credits_charged"] == "0" + + # The transcript must not carry a credential embedded in --api-url. + transcript_api_url = transcript["api_url"] + assert "@" not in transcript_api_url From 48202e127a9c366b0d55da1e769ff8c1e69ec6c2 Mon Sep 17 00:00:00 2001 From: Claude Date: Fri, 14 Aug 2026 07:18:41 +0000 Subject: [PATCH 4/6] docs(readme): document the one-command live loop proof and handoff bundle MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Add a golden-path subsection for `make live-loop-proof`: it drives the full core loop against a running quickstart server as a self-provisioned non-admin caller, asserts each invariant, and writes a partner handoff bundle (portable success + denial receipts, public key set, transcript, and VERIFY.md) that a partner engineer can verify offline with no credential. Note the new audit-chain step in the quickstart walkthrough, and document that offline receipt verification now needs only `cryptography` — importing the SDK no longer pulls in httpx. Co-Authored-By: Claude Fable 5 Claude-Session: https://claude.ai/code/session_01MpjPj59M4n6boNmbs7CpGR --- README.md | 39 ++++++++++++++++++++++++++++++++++++++- 1 file changed, 38 insertions(+), 1 deletion(-) diff --git a/README.md b/README.md index b13f1966..b4a4c988 100644 --- a/README.md +++ b/README.md @@ -255,11 +255,39 @@ make quickstart Then follow [docs/quickstart.md](docs/quickstart.md): mint your own wallet-scoped key (no operator, no pre-shared secret), issue yourself a permit, invoke a real governed tool, deliberately try to double-charge and -overspend, and finish holding a signed receipt you verified offline — plus +overspend, verify your wallet's tamper-evident audit chain with your own +key, and finish holding a signed receipt you verified offline — plus a forged one the verifier rejected. Every step of that page runs in CI against a freshly booted server (`make quickstart-check`), so the documented path cannot silently rot. +### One command, whole loop, partner handoff bundle + +To drive the entire loop end to end against a running quickstart server — +and produce artifacts you can hand to someone else to verify — run: + +```bash +make quickstart # terminal 1: boots the server +make live-loop-proof # terminal 2: drives the loop, writes the bundle +``` + +`live-loop-proof` walks the full core loop as a self-provisioned non-admin +caller — `discover → authenticate → authorize → invoke → meter → receipt → +replay → audit → govern` — and asserts each invariant rather than only +printing it: the call charges the known price once, the replay returns the +same receipt with no second debit, the audit chain verifies, and the +out-of-scope call is denied with a signed, zero-charge receipt. It exits +non-zero the moment any stage's invariant breaks. + +On success it writes a handoff bundle to `data/live-loop-proof/`: the +portable success and denial receipts, the issuer's public key set, a +machine-readable transcript, and a `VERIFY.md` a partner engineer can +follow to verify both receipts offline — no account, no credential, and no +network access to the issuing server. Handing that directory to a partner +who verifies a receipt themselves is the independent-verification step the +customer-validation milestone asks for. The command runs end to end in CI +(`tests/test_quickstart_path.py`). + ## Run the API locally The following keeps strict trust mode enabled while using local SQLite files. @@ -653,6 +681,15 @@ python -m pip install -e './b2a_sdk[dev]' `B2AClient` remains as deprecated compatibility during the 0.4.x transition. No TypeScript package is published. Do not advertise PyPI or npm installation. +Offline receipt verification is deliberately dependency-minimal: the +`b2a-verify-receipt` CLI (and `verify_bundle`) verify a signed receipt +against a published key set with only `cryptography` installed — no +networking library and no account. Importing the package no longer pulls in +the HTTP client, so `pip install "./b2a_sdk[verify]"` (or just +`cryptography` with `PYTHONPATH=b2a_sdk/src`) is enough to check a receipt. +The networked `--issuer` fetch is the only path that additionally needs +`httpx`. + ## Documentation - [docs/30-day-customer-validation.md](docs/30-day-customer-validation.md) — active company milestone: customer interviews, partner-owned pilot, and From 5ec3b8384c1a83e3cc064f7cecd6e6e994ba38f1 Mon Sep 17 00:00:00 2001 From: Claude Date: Fri, 14 Aug 2026 07:26:00 +0000 Subject: [PATCH 5/6] fix(demo): fail closed on unsafe --allow-remote proof targets Address two review findings on the opt-in remote path of the live-loop proof, both flagged on a security-critical file: - Cleartext credentials: the run mints an API key and sends it in X-API-Key on every request. A non-loopback target now must use https:// or the run refuses to start, so the minted key can never cross the network unencrypted. Loopback http (localhost) is still fine. - Honest scope: --allow-remote previously advertised "a staging host you control," but self-provision is refused on production-like environments (staging/preprod/preview) by design, so that mode could never pass the authenticate stage there. The help text, module docstring, and the self-provision 403 error now say plainly that --allow-remote targets a remote instance with self-provision enabled, not a production-like host. Adds a fast guard test (no server needed) that both rejections fail closed before any credential is minted or sent. Co-Authored-By: Claude Fable 5 Claude-Session: https://claude.ai/code/session_01MpjPj59M4n6boNmbs7CpGR --- scripts/live_loop_proof.py | 54 ++++++++++++++++++++++++++--------- tests/test_quickstart_path.py | 30 +++++++++++++++++++ 2 files changed, 70 insertions(+), 14 deletions(-) diff --git a/scripts/live_loop_proof.py b/scripts/live_loop_proof.py index c8184e38..75d83340 100644 --- a/scripts/live_loop_proof.py +++ b/scripts/live_loop_proof.py @@ -27,9 +27,12 @@ make quickstart # terminal 1: boots the server make live-loop-proof # terminal 2: drives the loop, writes the bundle -Self-provision only exists on local dev postures — production-like -deployments refuse it at boot — so this script is loopback-only unless -``--allow-remote`` is passed for a staging host you control. +Self-provision only exists on dev-like postures — production-like +deployments (staging, preprod, preview, ...) refuse it — so this script is +loopback-only unless ``--allow-remote`` is passed, and then only against a +remote instance that has self-provision enabled. Non-loopback targets must +use ``https://``: the run mints an API key and sends it on every request, +which must never cross the network in the clear. """ from __future__ import annotations @@ -173,9 +176,17 @@ def authenticate(self) -> dict[str, str]: "/v1/dev-keys/self-provision", json={"agent_id": f"live-loop-proof-{self.run_id}"}, ) + hint = ( + " -- a production-like target (staging/preprod/preview) refuses " + "self-provision by design; point --allow-remote at a remote " + "instance with ENABLE_DEV_KEY_SELF_PROVISION=true" + if minted.status_code == 403 + else "" + ) require( minted.status_code in (200, 201), - f"self-provision refused ({minted.status_code}): {minted.text}", + f"self-provision refused ({minted.status_code}): " + f"{minted.text}{hint}", ) body = minted.json() identity = { @@ -521,20 +532,35 @@ def main(argv: list[str] | None = None) -> int: "--allow-remote", action="store_true", help=( - "Permit a non-loopback --api-url. Only for a staging host you " - "control; the run self-provisions keys and writes test data." + "Permit a non-loopback --api-url (which must then use https://). " + "Only for a remote instance with self-provision enabled -- " + "production-like hosts (staging, preprod, preview, ...) refuse " + "self-provision by design, so this mode cannot authenticate there." ), ) args = parser.parse_args(argv) - host = urlparse(args.api_url).hostname or "" - if host not in LOOPBACK_HOSTS and not args.allow_remote: - print( - f"refusing non-loopback target {args.api_url!r} without " - "--allow-remote", - file=sys.stderr, - ) - return 2 + parsed = urlparse(args.api_url) + host = parsed.hostname or "" + if host not in LOOPBACK_HOSTS: + if not args.allow_remote: + print( + f"refusing non-loopback target {args.api_url!r} without " + "--allow-remote", + file=sys.stderr, + ) + return 2 + # The run mints an API key and sends it in X-API-Key on every request; + # a non-loopback hop must be encrypted or that credential (and every + # governed call) crosses the network in the clear. + if parsed.scheme != "https": + print( + f"refusing cleartext {parsed.scheme or 'http'}:// for " + f"non-loopback target {args.api_url!r}; use https:// so the " + "minted key is not sent over the wire unencrypted", + file=sys.stderr, + ) + return 2 try: run(args.api_url.rstrip("/"), args.output_dir) diff --git a/tests/test_quickstart_path.py b/tests/test_quickstart_path.py index 734263f8..ea791f22 100644 --- a/tests/test_quickstart_path.py +++ b/tests/test_quickstart_path.py @@ -292,6 +292,36 @@ def test_documented_quickstart_path(quickstart_server, tmp_path): client.close() +def _run_proof_script(*args: str) -> subprocess.CompletedProcess: + return subprocess.run( + [sys.executable, str(REPO_ROOT / "scripts" / "live_loop_proof.py"), *args], + cwd=REPO_ROOT, + env={**os.environ, "PYTHONPATH": str(REPO_ROOT)}, + capture_output=True, + text=True, + timeout=30, + ) + + +def test_live_loop_proof_refuses_unsafe_remote_targets(): + """The guard must fail closed before minting or sending any credential. + + Both rejections happen before any network call, so no server is needed. + """ + # Non-loopback without opting in. + without_optin = _run_proof_script("--api-url", "https://example.com") + assert without_optin.returncode == 2 + assert "without --allow-remote" in without_optin.stderr + + # Non-loopback over cleartext, even with --allow-remote: the minted key + # would cross the network unencrypted, so it must be refused. + cleartext = _run_proof_script( + "--api-url", "http://example.com", "--allow-remote" + ) + assert cleartext.returncode == 2 + assert "cleartext" in cleartext.stderr + + def test_live_loop_proof_script(quickstart_server, tmp_path): """The one-command live proof drives every stage and writes the bundle. From f95d1d5a431fb2edf85c396f048ab10edc6af8e8 Mon Sep 17 00:00:00 2001 From: Claude Date: Fri, 14 Aug 2026 07:32:34 +0000 Subject: [PATCH 6/6] refactor(demo): address review feedback on the live-loop proof Apply the actionable findings from automated review: - Robust handoff bundle: clear prior artifacts before a run so a partial failure can't leave a transcript describing one run beside receipts from another (the directory is reused and handed to a partner whole). - Trust-key URL: build it with base_url.join() so a trailing-slash base_url (some httpx versions) can't produce "//.well-known/..." that misses the single-slash route. - VERIFY.md: state the working directory explicitly (run from the bundle directory) and use an absolute PYTHONPATH, so the from-source and installed-CLI forms each run from one consistent cwd. - Negative-path tests: a tampered bundle still exits 1 (INVALID) on the httpx-free path; unknown SDK attributes raise AttributeError (not ImportError); the proof refuses a non-loopback and an unreachable target with the documented exit codes and writes no partial bundle; and the URL sanitizer is exercised with a real credential-bearing URL instead of a vacuous loopback assertion. - Consistency: compare the denial charge as Decimal like the success charge; drive transcript recording by an explicit `record` parameter rather than a label string; prepend to inherited PYTHONPATH in subprocess helpers rather than replacing it; hyphenate "end-to-end" in the README. Verified: quickstart-path suite 5 passed; SDK offline tests 5 passed (dev), 3 passed + 2 skipped (bare wheel, cryptography absent); ruff clean. Co-Authored-By: Claude Fable 5 Claude-Session: https://claude.ai/code/session_01MpjPj59M4n6boNmbs7CpGR --- README.md | 2 +- b2a_sdk/tests/test_offline_verify_no_httpx.py | 54 ++++++++++++- scripts/live_loop_proof.py | 75 ++++++++++++++----- tests/test_quickstart_path.py | 63 +++++++++++++--- 4 files changed, 162 insertions(+), 32 deletions(-) diff --git a/README.md b/README.md index b4a4c988..5e51b106 100644 --- a/README.md +++ b/README.md @@ -263,7 +263,7 @@ documented path cannot silently rot. ### One command, whole loop, partner handoff bundle -To drive the entire loop end to end against a running quickstart server — +To drive the entire loop end-to-end against a running quickstart server — and produce artifacts you can hand to someone else to verify — run: ```bash diff --git a/b2a_sdk/tests/test_offline_verify_no_httpx.py b/b2a_sdk/tests/test_offline_verify_no_httpx.py index be10f518..a7fada5f 100644 --- a/b2a_sdk/tests/test_offline_verify_no_httpx.py +++ b/b2a_sdk/tests/test_offline_verify_no_httpx.py @@ -36,11 +36,16 @@ def _run(code: str, *, block_httpx: bool) -> subprocess.CompletedProcess: - # Merge, don't replace, the environment: a bare ``env={...}`` would drop the - # interpreter's own paths and break site-packages resolution. + # Prepend to (not replace) PYTHONPATH: keep any inherited PYTHONPATH intact + # while ensuring the SDK source wins on import. + env = dict(os.environ) + existing = env.get("PYTHONPATH") + env["PYTHONPATH"] = ( + f"{SDK_SRC}{os.pathsep}{existing}" if existing else str(SDK_SRC) + ) return subprocess.run( [sys.executable, "-c", (_BLOCK_HTTPX if block_httpx else "") + code], - env={**os.environ, "PYTHONPATH": str(SDK_SRC)}, + env=env, capture_output=True, text=True, timeout=60, @@ -124,6 +129,49 @@ def test_verify_cli_keys_path_runs_without_httpx(tmp_path): assert "VERIFIED" in result.stdout +def test_tampered_bundle_rejected_without_httpx(tmp_path): + """The dependency-minimal path must still REJECT a forged bundle. + + VERIFY.md tells partners to branch on exit code 1; prove that contract + holds in the same httpx-free configuration, not just the accept path. + """ + pytest.importorskip("cryptography") + bundle_path, keys_path = _write_signed_bundle(tmp_path) + bundle = json.loads(bundle_path.read_text(encoding="utf-8")) + bundle["signing_input"] = bundle["signing_input"].replace( + '"outcome":"success"', '"outcome":"denied"' + ) + assert '"outcome":"denied"' in bundle["signing_input"] # the edit must bite + forged = tmp_path / "forged-bundle.json" + forged.write_text(json.dumps(bundle), encoding="utf-8") + result = _run( + "from b2a_sdk.verify_cli import main\n" + f"raise SystemExit(main(['--bundle', {str(forged)!r}, " + f"'--keys', {str(keys_path)!r}]))\n", + block_httpx=True, + ) + # 1 = well-formed but does not verify; must not be 0 and must not crash (2). + assert result.returncode == 1, result.stdout + result.stderr + assert "INVALID" in result.stderr + + +def test_unknown_attribute_raises_attribute_error(): + """The lazy __getattr__ must raise AttributeError for unknown names, not + ImportError -- wrong type breaks hasattr/getattr-default/pickle probing.""" + result = _run( + "import b2a_sdk\n" + "try:\n" + " b2a_sdk.NotAThing\n" + "except AttributeError:\n" + " print('attribute-error-ok')\n" + "else:\n" + " raise SystemExit('expected AttributeError')\n", + block_httpx=True, + ) + assert result.returncode == 0, result.stdout + result.stderr + assert "attribute-error-ok" in result.stdout + + def test_lazy_http_client_still_importable_when_httpx_present(): # The lazy PEP 562 surface must not have broken the real client import. pytest.importorskip("httpx") diff --git a/scripts/live_loop_proof.py b/scripts/live_loop_proof.py index 75d83340..67cf8a6a 100644 --- a/scripts/live_loop_proof.py +++ b/scripts/live_loop_proof.py @@ -64,6 +64,28 @@ # server that regressed metering to 0 (revenue leak) must fail, not pass. EXPECTED_CREDITS_PER_CALL = Decimal("2") LOOPBACK_HOSTS = {"127.0.0.1", "localhost", "::1"} +# Every file the run writes into the handoff directory. +ARTIFACTS = ( + "receipt-bundle.json", + "denial-bundle.json", + "trust-keys.json", + "transcript.json", + "VERIFY.md", +) + + +def _child_env(prepend: Path) -> dict[str, str]: + """Inherit the environment, prepending ``prepend`` to PYTHONPATH. + + Prepending rather than replacing keeps an inherited PYTHONPATH (CI, tox, a + developer shell) intact while ensuring the intended source wins on import. + """ + env = dict(os.environ) + existing = env.get("PYTHONPATH") + env["PYTHONPATH"] = ( + f"{prepend}{os.pathsep}{existing}" if existing else str(prepend) + ) + return env def _sanitize_url_for_record(url: str) -> str: @@ -101,17 +123,22 @@ def _sanitize_url_for_record(url: str) -> str: ## Verify offline (no network) The only dependency is `cryptography` for the Ed25519 check — no networking -library, no account. From a clone of the repository: - - pip install cryptography - PYTHONPATH=b2a_sdk/src python -m b2a_sdk.verify_cli \\ - --bundle receipt-bundle.json --keys trust-keys.json +library, no account. Run the commands below **from this directory** (the one +holding `receipt-bundle.json`); the bundle and key paths are relative to it. -Or with the SDK installed (`pip install "./b2a_sdk[verify]"` from the repo, -which provides the `b2a-verify-receipt` entry point and the verifier extra): +With the SDK installed from a clone — `pip install "/path/to/agent-middleware-api/b2a_sdk[verify]"`, +which pulls in `cryptography` and provides the `b2a-verify-receipt` command: b2a-verify-receipt --bundle receipt-bundle.json --keys trust-keys.json +Or straight from a clone's source, pointing `PYTHONPATH` at its `b2a_sdk/src` +by **absolute** path so it resolves no matter your working directory: + + pip install cryptography + PYTHONPATH=/path/to/agent-middleware-api/b2a_sdk/src \\ + python -m b2a_sdk.verify_cli \\ + --bundle receipt-bundle.json --keys trust-keys.json + Expected: `VERIFIED`, exit code 0. Repeat with `denial-bundle.json`. Exit codes are meant to be branched on: 0 verified, 1 well-formed but does @@ -198,7 +225,7 @@ def authenticate(self) -> dict[str, str]: return identity def authorize(self, identity: dict[str, str], *, allowed_tool: str, - label: str) -> str: + label: str, record: bool = True) -> str: expires_at = ( datetime.now(timezone.utc) + timedelta(minutes=30) ).strftime("%Y-%m-%dT%H:%M:%SZ") @@ -220,7 +247,7 @@ def authorize(self, identity: dict[str, str], *, allowed_tool: str, f"permit creation failed ({permit.status_code}): {permit.text}", ) permit_id = permit.json()["permit_id"] - if label == "grant": + if record: self.record( "authorize", permit_id=permit_id, @@ -346,9 +373,11 @@ def receipt_bundle(self, receipt_id: str, filename: str, *, def trust_keys(self) -> Path: # Deliberately unauthenticated: the key set must be fetchable by a # verifier who holds no credential at all. - keys = httpx.get( - f"{self.client.base_url}/.well-known/trust-keys.json", timeout=30 - ) + # base_url.join avoids a double slash: some httpx versions store + # base_url with a trailing slash, and Starlette does not collapse + # "//.well-known/..." to the single-slash route. + keys_url = self.client.base_url.join(".well-known/trust-keys.json") + keys = httpx.get(str(keys_url), timeout=30) require(keys.status_code == 200, "trust-keys.json not served") path = self.output_dir / "trust-keys.json" path.write_text(keys.text, encoding="utf-8") @@ -396,8 +425,10 @@ def audit(self) -> None: self.record("audit", valid=True, checked_events=chain["checked_events"]) def govern(self, identity: dict[str, str]) -> dict[str, Any]: + # record=False: the denial's own "govern" stage is what the transcript + # records, not this setup permit. narrow_permit = self.authorize( - identity, allowed_tool=UNGRANTED_TOOL, label="deny" + identity, allowed_tool=UNGRANTED_TOOL, label="deny", record=False ) denied = self.client.post( "/mcp/messages", @@ -418,7 +449,13 @@ def govern(self, identity: dict[str, str]) -> dict[str, Any]: ) denial = denied["error"]["data"]["receipt"] require(denial["outcome"] == "denied", "denial receipt outcome != denied") - require(denial["credits_charged"] == "0", "denial was charged") + # Compare as Decimal, like the success charge, so a formatting change + # ("0.00000000") cannot make this assertion silently pass or fail. + denial_charged = Decimal(str(denial["credits_charged"])) + require( + denial_charged == Decimal("0"), + f"denial was charged {denial_charged} credits", + ) require(bool(denial["signature"]), "denial receipt is unsigned") self.record( "govern", @@ -440,9 +477,7 @@ def verify_offline(self, bundle_path: Path, keys_path: Path, "--keys", str(keys_path), ], - # Merge, don't replace, the environment: a bare env={...} would - # drop the interpreter's own paths and can break the subprocess. - env={**os.environ, "PYTHONPATH": str(SDK_SRC)}, + env=_child_env(SDK_SRC), capture_output=True, text=True, timeout=60, @@ -458,6 +493,12 @@ def verify_offline(self, bundle_path: Path, keys_path: Path, def run(api_url: str, output_dir: Path) -> None: output_dir.mkdir(parents=True, exist_ok=True) + # Clear a prior run's artifacts first: a partial failure must not leave a + # transcript describing one run beside receipts from another, since the + # default output directory is reused and the directory is handed to a + # partner as a single coherent bundle. + for name in ARTIFACTS: + (output_dir / name).unlink(missing_ok=True) with httpx.Client(base_url=api_url, timeout=30) as client: proof = ProofRun(client, output_dir) print(f"live loop proof {proof.run_id} against {api_url}\n") diff --git a/tests/test_quickstart_path.py b/tests/test_quickstart_path.py index ea791f22..c75775a9 100644 --- a/tests/test_quickstart_path.py +++ b/tests/test_quickstart_path.py @@ -92,11 +92,19 @@ def quickstart_server(tmp_path_factory): process.wait(timeout=10) +def _prepend_pythonpath(path: Path) -> dict: + """Environment with ``path`` prepended to (not replacing) PYTHONPATH.""" + env = dict(os.environ) + existing = env.get("PYTHONPATH") + env["PYTHONPATH"] = f"{path}{os.pathsep}{existing}" if existing else str(path) + return env + + def _verify_cli(bundle_path: Path, keys_path: Path) -> subprocess.CompletedProcess: """Run the offline verifier exactly as the doc does: CLI, branchable exit.""" return subprocess.run( [sys.executable, "-m", "b2a_sdk.verify_cli", "--bundle", str(bundle_path), "--keys", str(keys_path)], - env={**os.environ, "PYTHONPATH": str(REPO_ROOT / "b2a_sdk" / "src")}, + env=_prepend_pythonpath(REPO_ROOT / "b2a_sdk" / "src"), capture_output=True, text=True, timeout=60, @@ -292,35 +300,66 @@ def test_documented_quickstart_path(quickstart_server, tmp_path): client.close() -def _run_proof_script(*args: str) -> subprocess.CompletedProcess: +def _run_proof_script(*args: str, timeout: int = 30) -> subprocess.CompletedProcess: return subprocess.run( [sys.executable, str(REPO_ROOT / "scripts" / "live_loop_proof.py"), *args], cwd=REPO_ROOT, - env={**os.environ, "PYTHONPATH": str(REPO_ROOT)}, + env=_prepend_pythonpath(REPO_ROOT), capture_output=True, text=True, - timeout=30, + timeout=timeout, ) -def test_live_loop_proof_refuses_unsafe_remote_targets(): +def test_live_loop_proof_refuses_unsafe_remote_targets(tmp_path): """The guard must fail closed before minting or sending any credential. Both rejections happen before any network call, so no server is needed. """ + out = tmp_path / "out" # Non-loopback without opting in. - without_optin = _run_proof_script("--api-url", "https://example.com") + without_optin = _run_proof_script( + "--api-url", "https://example.com", "--output-dir", str(out) + ) assert without_optin.returncode == 2 assert "without --allow-remote" in without_optin.stderr # Non-loopback over cleartext, even with --allow-remote: the minted key # would cross the network unencrypted, so it must be refused. cleartext = _run_proof_script( - "--api-url", "http://example.com", "--allow-remote" + "--api-url", "http://example.com", "--allow-remote", "--output-dir", str(out) ) assert cleartext.returncode == 2 assert "cleartext" in cleartext.stderr + # Fail closed means no handoff directory, not a half-written bundle. + assert not out.exists() + + +def test_live_loop_proof_reports_unreachable_server(tmp_path): + """An unused loopback port is a reachability failure (exit 2), distinct + from a broken invariant (exit 1); the codes must stay distinguishable.""" + dead_port = _free_port() + result = _run_proof_script( + "--api-url", f"http://127.0.0.1:{dead_port}", + "--output-dir", str(tmp_path / "out"), + ) + assert result.returncode == 2, result.stdout + result.stderr + assert "make quickstart" in result.stderr + + +def test_sanitize_url_for_record_strips_credentials(): + """The transcript ships to partners; a secret in --api-url must not.""" + sys.path.insert(0, str(REPO_ROOT / "scripts")) + from live_loop_proof import _sanitize_url_for_record + + sanitized = _sanitize_url_for_record( + "http://alice:s3cr3t@127.0.0.1:8000/base?token=t0ken#frag" + ) + assert sanitized == "http://127.0.0.1:8000/base" + for secret in ("alice", "s3cr3t", "t0ken", "frag"): + assert secret not in sanitized + def test_live_loop_proof_script(quickstart_server, tmp_path): """The one-command live proof drives every stage and writes the bundle. @@ -339,7 +378,7 @@ def test_live_loop_proof_script(quickstart_server, tmp_path): str(output_dir), ], cwd=REPO_ROOT, - env={**os.environ, "PYTHONPATH": str(REPO_ROOT)}, + env=_prepend_pythonpath(REPO_ROOT), capture_output=True, text=True, timeout=180, @@ -394,6 +433,8 @@ def _signed(bundle_name: str) -> dict: assert denial_signed["outcome"] == "denied" assert denial_signed["credits_charged"] == "0" - # The transcript must not carry a credential embedded in --api-url. - transcript_api_url = transcript["api_url"] - assert "@" not in transcript_api_url + # The recorded api_url is the sanitized loopback target verbatim (it + # carries no credential to strip). The credential-stripping behavior itself + # is proven by test_sanitize_url_for_record_strips_credentials, so this + # assertion is not the sole guard on the sanitizer. + assert transcript["api_url"] == quickstart_server