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/README.md b/README.md index b13f1966..5e51b106 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 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..a7fada5f --- /dev/null +++ b/b2a_sdk/tests/test_offline_verify_no_httpx.py @@ -0,0 +1,188 @@ +"""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: + # 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=env, + 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_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") + 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 227838bf..4d00ddd4 100644 --- a/docs/quickstart.md +++ b/docs/quickstart.md @@ -224,6 +224,23 @@ 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 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" \ + -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..67cf8a6a --- /dev/null +++ b/scripts/live_loop_proof.py @@ -0,0 +1,622 @@ +#!/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 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 + +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, urlunparse + +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 +# 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"} +# 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: + """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 + +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) + +The only dependency is `cryptography` for the Ed25519 check — no networking +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. + +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 +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 (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 +""" + + +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}"}, + ) + 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}): " + f"{minted.text}{hint}", + ) + 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, record: bool = True) -> 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 record: + 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", + ) + # 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 + + 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", + ) + # 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", + 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, *, + expected_outcome: str, + expected_credits: Decimal) -> 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") + # 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: + # Deliberately unauthenticated: the key set must be fetchable by a + # verifier who holds no credential at all. + # 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") + 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]: + # 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", record=False + ) + 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") + # 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", + 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=_child_env(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) + # 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") + + 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", + expected_outcome="success", + expected_credits=EXPECTED_CREDITS_PER_CALL, + ) + 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", + expected_outcome="denied", + expected_credits=Decimal("0"), + ) + + 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": _sanitize_url_for_record(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 (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) + + 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) + 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..c75775a9 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, @@ -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, @@ -216,6 +224,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 +298,143 @@ def test_documented_quickstart_path(quickstart_server, tmp_path): assert "VERIFIED" in denial_verified.stdout client.close() + + +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=_prepend_pythonpath(REPO_ROOT), + capture_output=True, + text=True, + timeout=timeout, + ) + + +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", "--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", "--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. + + 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=_prepend_pythonpath(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 + + # 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 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