From 5fd89655eaa5619063c903a3ff682a03d2fa894d Mon Sep 17 00:00:00 2001 From: Cursor Agent Date: Thu, 20 Aug 2026 09:26:38 +0000 Subject: [PATCH 1/3] Add machine-readable bootstrap output and constant test loop MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Two related outcomes: 1. Machine-readable mint output: partner_api_key_bootstrap.py now supports --json mode that prints one JSON object to stdout (api_key, key_id, wallet_id, key_prefix, etc.) with all human/status text going to stderr. This enables pipeable workflows like: script --json | jq -r .api_key | gh secret set CI_SMOKE_AGENT_KEY. Bootstrap key is never printed to stdout or stderr. 2. Constant test loop: scripts/constant_test_loop.py executes the production- ready cycle against the live contract: permit → invoke partner.notes.write (or partner.echo) → signed receipt → replay same idempotency key (same receipt_id, no second debit) → out-of-scope tool denied (optional, requires multiple tools). The loop reads the agent credential from $CI_SMOKE_AGENT_KEY (never hardcoded, never logged, never printed). Falls back to self-provisioning when no key provided. Default API URL overridable via $API_URL. Tests cover: - --json mode produces pipeable JSON to stdout - jq -r .api_key works on stdout - Bootstrap key never leaked to stdout or stderr - Constant test loop runs against local instance - Self-provisioning when no key provided - Agent key never logged during execution Success criteria met: - --json exists and jq -r .api_key works - Tests cover --json and constant loop locally - Loop never prints or logs the API key - Documentation explains local vs production usage - No extra product scope (no proof-surface work, no dogfood as production evidence) Co-authored-by: Christopher --- docs/constant-test-loop.md | 96 +++++++ scripts/constant_test_loop.py | 389 +++++++++++++++++++++++++++ scripts/partner_api_key_bootstrap.py | 18 +- tests/test_constant_test_loop.py | 318 ++++++++++++++++++++++ 4 files changed, 815 insertions(+), 6 deletions(-) create mode 100644 docs/constant-test-loop.md create mode 100755 scripts/constant_test_loop.py create mode 100644 tests/test_constant_test_loop.py diff --git a/docs/constant-test-loop.md b/docs/constant-test-loop.md new file mode 100644 index 00000000..620771d6 --- /dev/null +++ b/docs/constant-test-loop.md @@ -0,0 +1,96 @@ +# Constant Test Loop + +Production-ready smoke test for the governed trust plane. Exercises the complete loop: + +1. **Permit** → issue scoped permit for governed tool +2. **Invoke** → call the tool with the permit → signed receipt +3. **Meter** → verify ledger debit matches receipt charge +4. **Replay** → same idempotency key → same receipt_id, no second debit +5. **Deny** (optional) → out-of-scope tool is denied with 0 charge + +## Local Testing + +Against a running quickstart server: + +```bash +# Terminal 1: start quickstart +make quickstart + +# Terminal 2: run the constant test (auto-provisions its own key) +python scripts/constant_test_loop.py +``` + +The script self-provisions an agent key via `/v1/dev-keys/self-provision` when no credentials are provided. + +## Production Use + +Set `CI_SMOKE_AGENT_KEY` (and optionally `CI_SMOKE_WALLET_ID` and `CI_SMOKE_KEY_ID`) for a pre-provisioned agent credential: + +```bash +# Provision an agent key once (using bootstrap key) +export BOOTSTRAP_KEY="amw_live_..." +python scripts/partner_api_key_bootstrap.py \ + --api-url https://api.thisisatest.tech \ + --agent-id ci-smoke-agent \ + --key-name constant-test-loop \ + --budget-credits 5000 \ + --json | tee /tmp/agent-key.json + +# Extract and set as CI secret +export CI_SMOKE_AGENT_KEY="$(jq -r .api_key /tmp/agent-key.json)" +export CI_SMOKE_WALLET_ID="$(jq -r .agent_wallet_id /tmp/agent-key.json)" +export CI_SMOKE_KEY_ID="$(jq -r .key_id /tmp/agent-key.json)" + +# Run the constant test +API_URL=https://api.thisisatest.tech python scripts/constant_test_loop.py +``` + +## Machine-Readable Bootstrap Output + +`partner_api_key_bootstrap.py --json` prints JSON to stdout so the minted key can be piped directly: + +```bash +# Pipe to jq to extract the agent API key +BOOTSTRAP_KEY="..." python scripts/partner_api_key_bootstrap.py \ + --api-url https://api.thisisatest.tech \ + --agent-id ci-smoke-agent \ + --key-name test-key \ + --budget-credits 5000 \ + --json | jq -r .api_key + +# Or pipe directly to gh secret set +BOOTSTRAP_KEY="..." python scripts/partner_api_key_bootstrap.py \ + --api-url https://api.thisisatest.tech \ + --agent-id ci-smoke-agent \ + --key-name constant-test-loop \ + --budget-credits 5000 \ + --json | jq -r .api_key | \ + gh secret set CI_SMOKE_AGENT_KEY --repo PetrefiedThunder/agent-middleware-api +``` + +### Security Properties + +- Human/status text goes to stderr (never stdout in `--json` mode) +- Bootstrap/admin key is never printed to stdout or stderr +- Agent key is never logged or printed during constant test execution +- All secrets read from environment variables, never hardcoded + +## Exit Codes + +- `0` — all invariants held +- `1` — invariant failure (test failed) +- `2` — configuration error or network failure + +## Tests + +```bash +pytest tests/test_constant_test_loop.py -v +``` + +Covers: +- `--json` mode produces pipeable stdout +- Bootstrap key never leaks to stdout or stderr +- `jq -r .api_key` pipeline works +- Constant test loop runs against local instance +- Self-provisioning when no key provided +- Agent key never logged during execution diff --git a/scripts/constant_test_loop.py b/scripts/constant_test_loop.py new file mode 100755 index 00000000..858313ad --- /dev/null +++ b/scripts/constant_test_loop.py @@ -0,0 +1,389 @@ +#!/usr/bin/env python3 +"""Production-ready constant smoke test: permit → invoke → receipt → replay → deny. + +Exercises the complete governed loop against a live trust plane using the +dogfood tool (``partner.notes.write`` when ``ENABLE_DOGFOOD_TOOL=true``, or +``partner.echo`` when an upstream MCP tool is configured): + +1. Permit → invoke the governed tool with scoped permit → signed receipt +2. Replay same idempotency key → same receipt_id, no second debit +3. Out-of-scope tool is denied + +The loop reads the agent credential only from $CI_SMOKE_AGENT_KEY (never +hardcoded, never logged, never printed). Exits 0 on success, 1 on any +invariant failure, 2 on configuration error. + +Run locally against quickstart:: + + export CI_SMOKE_AGENT_KEY="$( + uv run python scripts/quickstart.py --reset & + sleep 5 + curl -s http://127.0.0.1:8000/v1/dev-keys/self-provision \\ + -H 'Content-Type: application/json' \\ + -d '{"agent_id":"constant-test-local"}' | jq -r .api_key + )" + uv run python scripts/constant_test_loop.py + +Run against production (once CI_SMOKE_AGENT_KEY is set in CI):: + + API_URL=https://api.thisisatest.tech \\ + uv run python scripts/constant_test_loop.py +""" + +from __future__ import annotations + +import argparse +import os +import sys +from datetime import datetime, timedelta, timezone +from decimal import Decimal +from pathlib import Path +from typing import Any + +import httpx + +ROOT = Path(__file__).resolve().parents[1] +DEFAULT_API_URL = "http://127.0.0.1:8000" +GOVERNED_TOOL = "partner.notes.write" # Dogfood tool (enabled by ENABLE_DOGFOOD_TOOL) +OUT_OF_SCOPE_TOOL = "some.other.tool" # Dummy tool not in permit scope +PERMIT_MAX_CREDITS = 10 + + +class SmokeTestFailure(RuntimeError): + """Raised when an invariant fails during the smoke test.""" + + +def require(condition: bool, message: str) -> None: + """Fail immediately if condition is false.""" + if not condition: + raise SmokeTestFailure(message) + + +def _get_agent_key() -> tuple[str, str, str]: + """Read agent credential from $CI_SMOKE_AGENT_KEY or self-provision one. + + Returns (api_key, wallet_id, key_id). Never logs or prints the key. + """ + key = os.environ.get("CI_SMOKE_AGENT_KEY", "").strip() + wallet_id = os.environ.get("CI_SMOKE_WALLET_ID", "").strip() + key_id = os.environ.get("CI_SMOKE_KEY_ID", "").strip() + + if key and wallet_id and key_id: + # All three provided - use them + return (key, wallet_id, key_id) + elif key: + # Key provided but missing wallet_id or key_id + raise SystemExit( + "error: CI_SMOKE_AGENT_KEY set but CI_SMOKE_WALLET_ID or " + "CI_SMOKE_KEY_ID missing. Provide all three or none." + ) + else: + # Nothing provided - will self-provision in run_constant_test + return ("", "", "") + + +def _get_api_url() -> str: + """Read API URL from $API_URL or default to local quickstart.""" + return os.environ.get("API_URL", DEFAULT_API_URL).rstrip("/") + + +def _post_json( + client: httpx.Client, + path: str, + body: dict[str, Any], + *, + expected_status: int, +) -> dict[str, Any]: + """POST JSON and require expected_status.""" + resp = client.post(path, json=body) + require( + resp.status_code == expected_status, + f"POST {path} → {resp.status_code}: {resp.text[:500]}", + ) + data = resp.json() + require(isinstance(data, dict), f"POST {path} returned non-object JSON") + return data + + +def _get_json( + client: httpx.Client, + path: str, + *, + expected_status: int, +) -> dict[str, Any]: + """GET JSON and require expected_status.""" + resp = client.get(path) + require( + resp.status_code == expected_status, + f"GET {path} → {resp.status_code}: {resp.text[:500]}", + ) + data = resp.json() + require(isinstance(data, dict), f"GET {path} returned non-object JSON") + return data + + +def _build_mcp_call( + *, + request_id: str, + tool: str, + wallet_id: str, + permit_id: str, + idempotency_key: str, + arguments: dict[str, Any] | None = None, +) -> dict[str, Any]: + """Build an MCP tools/call request with mcpContext.""" + return { + "jsonrpc": "2.0", + "id": request_id, + "method": "tools/call", + "params": { + "name": tool, + "arguments": arguments or {}, + "mcpContext": { + "wallet_id": wallet_id, + "permit_id": permit_id, + "idempotency_key": idempotency_key, + }, + }, + } + + +def _first_jsonrpc_result(response: dict[str, Any]) -> dict[str, Any]: + """Extract result from JSON-RPC response, failing if error is present.""" + require("result" in response, f"expected JSON-RPC result, got: {response}") + return response["result"] + + +def _first_jsonrpc_error(response: dict[str, Any]) -> dict[str, Any]: + """Extract error from JSON-RPC response, failing if result is present.""" + require("error" in response, f"expected JSON-RPC error, got: {response}") + return response["error"] + + +def run_constant_test(api_url: str, agent_key: str, wallet_id: str, key_id: str) -> None: + """Execute the constant test loop and assert all invariants.""" + print(f"[constant-test] target: {api_url}", file=sys.stderr) + + # If credentials not provided, self-provision + if not agent_key: + print("[constant-test] self-provisioning agent key", file=sys.stderr) + with httpx.Client(base_url=api_url, timeout=30.0) as provision_client: + provision_resp = _post_json( + provision_client, + "/v1/dev-keys/self-provision", + {"agent_id": f"constant-test-{datetime.now(timezone.utc).strftime('%Y%m%d-%H%M%S')}"}, + expected_status=201, + ) + agent_key = provision_resp["api_key"] + wallet_id = provision_resp["wallet_id"] + key_id = provision_resp["key_id"] + print( + f"[constant-test] self-provisioned: wallet_id={wallet_id}", + file=sys.stderr, + ) + + # Never log or print the agent_key. + headers = { + "X-API-Key": agent_key, + "Content-Type": "application/json", + } + + with httpx.Client(base_url=api_url, headers=headers, timeout=30.0) as client: + # Discover tools + print("[constant-test] discovering tools", file=sys.stderr) + tools = _get_json(client, "/mcp/tools.json", expected_status=200) + tool_names = [tool["name"] for tool in tools.get("tools", [])] + require( + GOVERNED_TOOL in tool_names, + f"{GOVERNED_TOOL} not discoverable; is ENABLE_DOGFOOD_TOOL=true?", + ) + + # Issue scoped permit + print("[constant-test] issuing scoped permit", file=sys.stderr) + expires_at = ( + datetime.now(timezone.utc) + timedelta(minutes=30) + ).strftime("%Y-%m-%dT%H:%M:%SZ") + # Add Idempotency-Key header for permit creation + permit_headers = { + **headers, + "Idempotency-Key": f"constant-test-permit-{datetime.now(timezone.utc).isoformat()}", + } + resp = client.post( + "/v1/permits", + json={ + "issuer_wallet_id": wallet_id, + "subject_wallet_id": wallet_id, + "subject_key_id": key_id, + "allowed_tools": [GOVERNED_TOOL], + "scopes": [f"tool:{GOVERNED_TOOL}:invoke", "billing:charge"], + "max_credits": PERMIT_MAX_CREDITS, + "expires_at": expires_at, + }, + headers=permit_headers, + ) + require( + resp.status_code == 201, + f"POST /v1/permits → {resp.status_code}: {resp.text[:500]}", + ) + permit = resp.json() + permit_id = permit["permit_id"] + print(f"[constant-test] permit_id={permit_id}", file=sys.stderr) + + # Invoke the governed tool + print(f"[constant-test] invoking {GOVERNED_TOOL}", file=sys.stderr) + call_body = _build_mcp_call( + request_id="constant-test-1", + tool=GOVERNED_TOOL, + wallet_id=wallet_id, + permit_id=permit_id, + idempotency_key="constant-test-invoke-1", + arguments={"text": "constant test loop"}, + ) + first_call = _post_json( + client, "/mcp/messages", call_body, expected_status=200 + ) + result = _first_jsonrpc_result(first_call) + require(result["isError"] is False, f"tool call failed: {result}") + receipt = result["receipt"] + require(receipt["outcome"] == "success", "receipt outcome != success") + require( + receipt["ledger_entry_id"] is not None, + "success receipt missing ledger entry", + ) + charged = Decimal(str(receipt["credits_charged"])) + require(charged > 0, f"success charged {charged} credits (expected > 0)") + print( + f"[constant-test] success: receipt_id={receipt['receipt_id']} " + f"charged={charged}", + file=sys.stderr, + ) + + # Check ledger debit + print("[constant-test] verifying ledger debit", file=sys.stderr) + ledger = _get_json( + client, f"/v1/billing/ledger/{wallet_id}", expected_status=200 + ) + matching = [ + e for e in ledger["entries"] if e["entry_id"] == receipt["ledger_entry_id"] + ] + require(len(matching) == 1, "receipt ledger_entry_id not in wallet ledger") + debit_amount = Decimal(str(matching[0]["amount"])) + require( + debit_amount == -charged, + f"ledger debit {debit_amount} != receipt charge {-charged}", + ) + debits_before_replay = len( + [e for e in ledger["entries"] if GOVERNED_TOOL in e.get("description", "")] + ) + + # Replay same idempotency key → same receipt_id, no second debit + print("[constant-test] replaying with same idempotency_key", file=sys.stderr) + replay_call = _post_json( + client, "/mcp/messages", call_body, expected_status=200 + ) + replay_result = _first_jsonrpc_result(replay_call) + replay_receipt = replay_result["receipt"] + require( + replay_receipt["receipt_id"] == receipt["receipt_id"], + "replay returned different receipt_id", + ) + ledger_after_replay = _get_json( + client, f"/v1/billing/ledger/{wallet_id}", expected_status=200 + ) + debits_after_replay = len( + [ + e + for e in ledger_after_replay["entries"] + if GOVERNED_TOOL in e.get("description", "") + ] + ) + require( + debits_after_replay == debits_before_replay, + "replay created a second debit", + ) + print( + f"[constant-test] replay OK: same receipt_id, no second debit", + file=sys.stderr, + ) + + # Out-of-scope tool denial check (optional) + # Only run if there are multiple tools available to test against. + # A single-tool environment exercises the core loop but can't test + # permit_tool_not_allowed without registering a second tool. + if len(tool_names) > 1: + print("[constant-test] attempting out-of-scope tool", file=sys.stderr) + # Find a tool that's not the governed one + other_tool = next((t for t in tool_names if t != GOVERNED_TOOL), None) + if other_tool: + denial_body = _build_mcp_call( + request_id="constant-test-deny", + tool=other_tool, + wallet_id=wallet_id, + permit_id=permit_id, + idempotency_key="constant-test-deny-1", + arguments={}, + ) + denial_call = _post_json( + client, "/mcp/messages", denial_body, expected_status=200 + ) + denial_error = _first_jsonrpc_error(denial_call) + require( + denial_error["message"] == "permit_tool_not_allowed", + f"expected permit_tool_not_allowed, got {denial_error['message']}", + ) + denial_receipt = denial_error["data"]["receipt"] + require( + denial_receipt["outcome"] == "denied", + "denial receipt outcome != denied", + ) + denial_charged = Decimal(str(denial_receipt["credits_charged"])) + require( + denial_charged == Decimal("0"), + f"denial charged {denial_charged} credits (expected 0)", + ) + print( + f"[constant-test] denial OK: permit_tool_not_allowed, 0 credits charged", + file=sys.stderr, + ) + else: + print( + "[constant-test] SKIPPED out-of-scope tool check (single tool only)", + file=sys.stderr, + ) + + print("[constant-test] ALL INVARIANTS HELD", file=sys.stderr) + + +def main(argv: list[str] | None = None) -> int: + """Entry point.""" + parser = argparse.ArgumentParser(description=__doc__.splitlines()[0]) + parser.add_argument( + "--api-url", + default=None, + help=( + "API URL to test against (default: $API_URL or " + f"{DEFAULT_API_URL}). Reads $CI_SMOKE_AGENT_KEY, " + "$CI_SMOKE_WALLET_ID, $CI_SMOKE_KEY_ID for auth, or self-provisions." + ), + ) + args = parser.parse_args(argv) + + api_url = args.api_url or _get_api_url() + agent_key, wallet_id, key_id = _get_agent_key() + + try: + run_constant_test(api_url, agent_key, wallet_id, key_id) + except SmokeTestFailure as failure: + print(f"\n[constant-test] FAILED: {failure}", file=sys.stderr) + return 1 + except httpx.HTTPError as network_error: + print( + f"\n[constant-test] network error: {network_error}", + file=sys.stderr, + ) + return 2 + return 0 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/scripts/partner_api_key_bootstrap.py b/scripts/partner_api_key_bootstrap.py index 6080eab7..59c47a5a 100644 --- a/scripts/partner_api_key_bootstrap.py +++ b/scripts/partner_api_key_bootstrap.py @@ -82,7 +82,15 @@ def provision( key_name: str, sponsor_wallet_id: str | None, agent_wallet_id: str | None, + json_mode: bool = False, ) -> dict[str, Any]: + """Provision sponsor → agent → key via bootstrap/admin key. + + Args: + json_mode: If True, print only machine-readable status to stderr (never + the api_key or bootstrap_key). Human text goes to stderr so + `jq -r .api_key` works on stdout JSON output. + """ base = _require_safe_api_url(api_url) headers = { "X-API-Key": bootstrap_key, @@ -152,11 +160,6 @@ def provision( "key_id": key.get("key_id"), "key_prefix": key.get("key_prefix"), "api_key": api_key, - "note": ( - "Store api_key securely. Bootstrap key must not be shared " - "with the partner agent. On partial failure, re-run with " - "--sponsor-wallet-id / --agent-wallet-id to resume." - ), } @@ -212,11 +215,15 @@ def main() -> int: key_name=args.key_name, sponsor_wallet_id=args.sponsor_wallet_id, agent_wallet_id=args.agent_wallet_id, + json_mode=args.json, ) if args.json: + # Machine-readable: JSON to stdout, all status to stderr. + # Never print bootstrap_key. This lets `jq -r .api_key` work. print(json.dumps(result, indent=2)) else: + # Human-readable summary to stdout (legacy behavior). print("Partner API key bootstrap OK") print(f" api_url: {result['api_url']}") print(f" sponsor_wallet_id: {result['sponsor_wallet_id']}") @@ -224,7 +231,6 @@ def main() -> int: print(f" key_id: {result['key_id']}") print(f" key_prefix: {result['key_prefix']}") print(f" api_key (once): {result['api_key']}") - print(f" note: {result['note']}") return 0 diff --git a/tests/test_constant_test_loop.py b/tests/test_constant_test_loop.py new file mode 100644 index 00000000..e6063c9e --- /dev/null +++ b/tests/test_constant_test_loop.py @@ -0,0 +1,318 @@ +"""Tests for the constant smoke test loop and machine-readable bootstrap output. + +Covers: +1. partner_api_key_bootstrap.py --json mode produces pipeable JSON to stdout +2. Human/status text goes to stderr, never stdout (in --json mode) +3. Bootstrap key is never printed to stdout or stderr +4. constant_test_loop.py executes the full loop against a local instance +5. API key is never logged or printed +""" + +from __future__ import annotations + +import json +import os +import signal +import socket +import subprocess +import sys +import tempfile +import time +from pathlib import Path + +import httpx +import pytest + +REPO_ROOT = Path(__file__).resolve().parent.parent +BOOTSTRAP_SCRIPT = REPO_ROOT / "scripts" / "partner_api_key_bootstrap.py" +CONSTANT_TEST_SCRIPT = REPO_ROOT / "scripts" / "constant_test_loop.py" + + +def _free_port() -> int: + """Find a free port on localhost.""" + with socket.socket() as sock: + sock.bind(("127.0.0.1", 0)) + return sock.getsockname()[1] + + +@pytest.fixture(scope="module") +def test_server(tmp_path_factory): + """Boot a minimal test server with a static dev key for bootstrap tests.""" + import base64 + import secrets + + state_dir = tmp_path_factory.mktemp("bootstrap-test-state") + port = _free_port() + static_dev_key = f"amw_dev_{secrets.token_urlsafe(32)}" + signing_seed = base64.b64encode(secrets.token_bytes(32)).decode() + + # Boot uvicorn directly with minimal env for testing + test_env = { + **os.environ, + "PYTHONPATH": str(REPO_ROOT), + "ENVIRONMENT": "local", + "DATABASE_URL": f"sqlite+aiosqlite:///{state_dir / 'api.db'}", + "STATE_BACKEND": "sqlite", + "SQLITE_URL": str(state_dir / "state.db"), + "STATIC_DEV_API_KEYS": static_dev_key, + "VALID_API_KEYS": "", + "TRUST_MODE_ENABLED": "true", + "TRUST_SIGNING_KEY_ID": "test-ed25519", + "TRUST_SIGNING_PRIVATE_KEY_B64": signing_seed, + "ENABLE_DEV_KEY_SELF_PROVISION": "true", + "ENABLE_DOGFOOD_TOOL": "true", + "MCP_UPSTREAM_ENABLED": "false", + "ALLOW_LEGACY_UNPERMITTED_MCP": "false", + "ENABLE_PROOF_SURFACES": "false", + } + + log_path = state_dir / "server.log" + with log_path.open("wb") as log_file: + process = subprocess.Popen( + [ + sys.executable, + "-m", + "uvicorn", + "app.main:app", + "--host", + "127.0.0.1", + "--port", + str(port), + ], + cwd=REPO_ROOT, + env=test_env, + stdout=log_file, + stderr=subprocess.STDOUT, + start_new_session=True, + ) + + base_url = f"http://127.0.0.1:{port}" + try: + deadline = time.monotonic() + 120 + while time.monotonic() < deadline: + if process.poll() is not None: + raise RuntimeError( + "test server exited early:\n" + + log_path.read_text(encoding="utf-8", errors="replace") + ) + try: + if httpx.get(f"{base_url}/health", timeout=2).status_code == 200: + break + except httpx.HTTPError: + pass + time.sleep(0.5) + else: + raise RuntimeError("test server never became healthy") + yield (base_url, static_dev_key) + finally: + process.send_signal(signal.SIGINT) + try: + process.wait(timeout=20) + except subprocess.TimeoutExpired: + os.killpg(process.pid, signal.SIGKILL) + process.wait(timeout=10) + + +def test_bootstrap_json_mode_produces_pipeable_stdout(test_server): + """partner_api_key_bootstrap.py --json prints JSON to stdout, status to stderr.""" + base_url, static_dev_key = test_server + bootstrap_key = static_dev_key + + # Run bootstrap script in --json mode + result = subprocess.run( + [ + sys.executable, + str(BOOTSTRAP_SCRIPT), + "--api-url", + base_url, + "--agent-id", + "json-test-agent", + "--key-name", + "json-test-key", + "--budget-credits", + "100", + "--json", + ], + env={**os.environ, "BOOTSTRAP_KEY": bootstrap_key}, + capture_output=True, + text=True, + timeout=30, + ) + + assert result.returncode == 0, f"bootstrap failed: {result.stderr}" + + # stdout must be valid JSON + stdout_data = json.loads(result.stdout) + assert "api_key" in stdout_data, "JSON output missing api_key" + assert "wallet_id" in stdout_data or "agent_wallet_id" in stdout_data + assert "key_id" in stdout_data + + # stderr should contain human-readable status (or be empty) + # Key invariant: bootstrap_key must NOT appear in stdout or stderr + assert bootstrap_key not in result.stdout, "bootstrap key leaked to stdout" + assert bootstrap_key not in result.stderr, "bootstrap key leaked to stderr" + + +def test_bootstrap_json_mode_jq_pipeable(test_server): + """--json output can be piped to `jq -r .api_key`.""" + base_url, static_dev_key = test_server + bootstrap_key = static_dev_key + + # Run bootstrap --json and pipe to jq + bootstrap_proc = subprocess.Popen( + [ + sys.executable, + str(BOOTSTRAP_SCRIPT), + "--api-url", + base_url, + "--agent-id", + "jq-pipeable-agent", + "--key-name", + "jq-pipeable-key", + "--budget-credits", + "50", + "--json", + ], + env={**os.environ, "BOOTSTRAP_KEY": bootstrap_key}, + stdout=subprocess.PIPE, + stderr=subprocess.PIPE, + text=True, + ) + jq_proc = subprocess.Popen( + ["jq", "-r", ".api_key"], + stdin=bootstrap_proc.stdout, + stdout=subprocess.PIPE, + stderr=subprocess.PIPE, + text=True, + ) + bootstrap_proc.stdout.close() # type: ignore + jq_stdout, jq_stderr = jq_proc.communicate(timeout=30) + bootstrap_proc.wait(timeout=5) + + assert jq_proc.returncode == 0, f"jq failed: {jq_stderr}" + extracted_key = jq_stdout.strip() + assert extracted_key, "jq extracted empty api_key" + # Keys can have amw_ or b2a_ prefix depending on the API version + assert extracted_key.startswith(("amw_", "b2a_")), f"unexpected key prefix: {extracted_key[:10]}" + + # The extracted key should work - test by listing wallets + wallets_response = httpx.get( + f"{base_url}/v1/billing/wallets", + headers={"X-API-Key": extracted_key}, + timeout=10, + ) + assert wallets_response.status_code == 200, "extracted key is not valid" + assert "wallets" in wallets_response.json(), "wallets response malformed" + + +def test_bootstrap_json_never_prints_bootstrap_key(test_server): + """Bootstrap key must never appear in stdout or stderr, even in --json mode.""" + base_url, static_dev_key = test_server + bootstrap_key = static_dev_key + + for json_flag in (True, False): + args = [ + sys.executable, + str(BOOTSTRAP_SCRIPT), + "--api-url", + base_url, + "--agent-id", + f"no-leak-{json_flag}", + "--key-name", + "no-leak-key", + "--budget-credits", + "10", + ] + if json_flag: + args.append("--json") + + result = subprocess.run( + args, + env={**os.environ, "BOOTSTRAP_KEY": bootstrap_key}, + capture_output=True, + text=True, + timeout=30, + ) + + assert result.returncode == 0 + assert bootstrap_key not in result.stdout, ( + f"bootstrap key leaked to stdout (json={json_flag})" + ) + assert bootstrap_key not in result.stderr, ( + f"bootstrap key leaked to stderr (json={json_flag})" + ) + + +def test_constant_test_loop_against_local_instance(test_server): + """constant_test_loop.py runs the full loop against a local instance.""" + base_url, _ = test_server + + # Run the constant test loop - it will self-provision its own key + # Remove CI_SMOKE_AGENT_KEY from env so it self-provisions + env = {k: v for k, v in os.environ.items() + if not k.startswith("CI_SMOKE")} + result = subprocess.run( + [sys.executable, str(CONSTANT_TEST_SCRIPT), "--api-url", base_url], + env=env, + capture_output=True, + text=True, + timeout=60, + ) + + assert result.returncode == 0, ( + f"constant test loop failed:\nstdout: {result.stdout}\nstderr: {result.stderr}" + ) + assert "ALL INVARIANTS HELD" in result.stderr + + +def test_constant_test_loop_self_provisions_when_no_key_provided(test_server): + """constant_test_loop.py self-provisions when CI_SMOKE_AGENT_KEY is not set.""" + base_url, _ = test_server + env = {k: v for k, v in os.environ.items() + if not k.startswith("CI_SMOKE")} + result = subprocess.run( + [sys.executable, str(CONSTANT_TEST_SCRIPT), "--api-url", base_url], + env=env, + capture_output=True, + text=True, + timeout=60, + ) + + assert result.returncode == 0, f"self-provision flow failed: {result.stderr}" + assert "self-provisioning agent key" in result.stderr + assert "ALL INVARIANTS HELD" in result.stderr + + +def test_constant_test_loop_never_logs_api_key(test_server): + """Agent key is never printed or logged during constant test execution.""" + base_url, _ = test_server + # Pre-provision a key to test that it's never logged + provision_response = httpx.post( + f"{base_url}/v1/dev-keys/self-provision", + json={"agent_id": "no-log-test"}, + timeout=10, + ) + assert provision_response.status_code in (200, 201) + provision_data = provision_response.json() + agent_key = provision_data["api_key"] + wallet_id = provision_data["wallet_id"] + key_id = provision_data["key_id"] + + # Run with explicit credentials + result = subprocess.run( + [sys.executable, str(CONSTANT_TEST_SCRIPT), "--api-url", base_url], + env={ + **os.environ, + "CI_SMOKE_AGENT_KEY": agent_key, + "CI_SMOKE_WALLET_ID": wallet_id, + "CI_SMOKE_KEY_ID": key_id, + }, + capture_output=True, + text=True, + timeout=60, + ) + + assert result.returncode == 0, f"failed: {result.stderr}" + # The agent key must never appear anywhere in the output + assert agent_key not in result.stdout, "agent key in stdout" + assert agent_key not in result.stderr, "agent key in stderr" From 814fa62aceb50ab7a08f7cf8d6b98e9015aea3b1 Mon Sep 17 00:00:00 2001 From: Cursor Agent Date: Thu, 20 Aug 2026 09:37:20 +0000 Subject: [PATCH 2/3] fix: address lint failures and review feedback - Remove unnecessary f-string prefixes (ruff F541) - Remove unused tempfile import (ruff F401) - Make idempotency keys unique per run with timestamp - Select partner.echo (production) over partner.notes.write (dogfood) - Add HTTPS validation for non-loopback URLs - Support CI_SMOKE_AGENT_KEY alone (fetch wallet_id/key_id from API) - Fix world-readable /tmp/agent-key.json with umask 077 + chmod 600 - Update docs to clarify wallet_id/key_id are optional Addresses: - Bugbot: fixed idempotency keys, optional wallet_id/key_id, file perms - Codex: production tool selection, dynamic idempotency, HTTPS check - CI lint: 3 ruff violations fixed Co-authored-by: Christopher --- docs/constant-test-loop.md | 4 +- scripts/constant_test_loop.py | 148 +++++++++++++++++++++---------- tests/test_constant_test_loop.py | 1 - 3 files changed, 104 insertions(+), 49 deletions(-) diff --git a/docs/constant-test-loop.md b/docs/constant-test-loop.md index 620771d6..a690f38c 100644 --- a/docs/constant-test-loop.md +++ b/docs/constant-test-loop.md @@ -24,17 +24,19 @@ The script self-provisions an agent key via `/v1/dev-keys/self-provision` when n ## Production Use -Set `CI_SMOKE_AGENT_KEY` (and optionally `CI_SMOKE_WALLET_ID` and `CI_SMOKE_KEY_ID`) for a pre-provisioned agent credential: +Set `CI_SMOKE_AGENT_KEY` for a pre-provisioned agent credential. Optionally provide `CI_SMOKE_WALLET_ID` and `CI_SMOKE_KEY_ID` for faster startup (the script will fetch them from the API if not provided): ```bash # Provision an agent key once (using bootstrap key) export BOOTSTRAP_KEY="amw_live_..." +umask 077 # Ensure restrictive permissions for key file python scripts/partner_api_key_bootstrap.py \ --api-url https://api.thisisatest.tech \ --agent-id ci-smoke-agent \ --key-name constant-test-loop \ --budget-credits 5000 \ --json | tee /tmp/agent-key.json +chmod 600 /tmp/agent-key.json # Restrict to owner-only read/write # Extract and set as CI secret export CI_SMOKE_AGENT_KEY="$(jq -r .api_key /tmp/agent-key.json)" diff --git a/scripts/constant_test_loop.py b/scripts/constant_test_loop.py index 858313ad..e054d78d 100755 --- a/scripts/constant_test_loop.py +++ b/scripts/constant_test_loop.py @@ -1,33 +1,28 @@ #!/usr/bin/env python3 """Production-ready constant smoke test: permit → invoke → receipt → replay → deny. -Exercises the complete governed loop against a live trust plane using the -dogfood tool (``partner.notes.write`` when ``ENABLE_DOGFOOD_TOOL=true``, or -``partner.echo`` when an upstream MCP tool is configured): +Exercises the complete governed loop against a live trust plane. Prefers the +production tool ``partner.echo`` (upstream MCP) when available, falls back to +``partner.notes.write`` (dogfood, requires ``ENABLE_DOGFOOD_TOOL=true`` locally): 1. Permit → invoke the governed tool with scoped permit → signed receipt 2. Replay same idempotency key → same receipt_id, no second debit -3. Out-of-scope tool is denied +3. Out-of-scope tool is denied (when multiple tools exist) The loop reads the agent credential only from $CI_SMOKE_AGENT_KEY (never -hardcoded, never logged, never printed). Exits 0 on success, 1 on any -invariant failure, 2 on configuration error. +hardcoded, never logged, never printed). Optionally reads $CI_SMOKE_WALLET_ID +and $CI_SMOKE_KEY_ID for faster startup; fetches them from the API if missing. +Exits 0 on success, 1 on any invariant failure, 2 on configuration error. -Run locally against quickstart:: +Run locally (auto-provisions agent key):: - export CI_SMOKE_AGENT_KEY="$( - uv run python scripts/quickstart.py --reset & - sleep 5 - curl -s http://127.0.0.1:8000/v1/dev-keys/self-provision \\ - -H 'Content-Type: application/json' \\ - -d '{"agent_id":"constant-test-local"}' | jq -r .api_key - )" - uv run python scripts/constant_test_loop.py + make quickstart # in terminal 1 + python scripts/constant_test_loop.py # in terminal 2 Run against production (once CI_SMOKE_AGENT_KEY is set in CI):: API_URL=https://api.thisisatest.tech \\ - uv run python scripts/constant_test_loop.py + python scripts/constant_test_loop.py """ from __future__ import annotations @@ -44,7 +39,9 @@ ROOT = Path(__file__).resolve().parents[1] DEFAULT_API_URL = "http://127.0.0.1:8000" -GOVERNED_TOOL = "partner.notes.write" # Dogfood tool (enabled by ENABLE_DOGFOOD_TOOL) +# Prefer partner.echo (production governed tool) over partner.notes.write (dogfood) +PRODUCTION_GOVERNED_TOOL = "partner.echo" +DOGFOOD_GOVERNED_TOOL = "partner.notes.write" OUT_OF_SCOPE_TOOL = "some.other.tool" # Dummy tool not in permit scope PERMIT_MAX_CREDITS = 10 @@ -60,9 +57,11 @@ def require(condition: bool, message: str) -> None: def _get_agent_key() -> tuple[str, str, str]: - """Read agent credential from $CI_SMOKE_AGENT_KEY or self-provision one. + """Read agent credential from $CI_SMOKE_AGENT_KEY or signal self-provision. - Returns (api_key, wallet_id, key_id). Never logs or prints the key. + Returns (api_key, wallet_id, key_id). If CI_SMOKE_AGENT_KEY is set but + wallet_id/key_id are missing, returns (key, "", "") to signal they should + be fetched from the API. Never logs or prints the key. """ key = os.environ.get("CI_SMOKE_AGENT_KEY", "").strip() wallet_id = os.environ.get("CI_SMOKE_WALLET_ID", "").strip() @@ -72,19 +71,32 @@ def _get_agent_key() -> tuple[str, str, str]: # All three provided - use them return (key, wallet_id, key_id) elif key: - # Key provided but missing wallet_id or key_id - raise SystemExit( - "error: CI_SMOKE_AGENT_KEY set but CI_SMOKE_WALLET_ID or " - "CI_SMOKE_KEY_ID missing. Provide all three or none." - ) + # Key provided but wallet_id/key_id missing - will fetch from API + return (key, "", "") else: # Nothing provided - will self-provision in run_constant_test return ("", "", "") def _get_api_url() -> str: - """Read API URL from $API_URL or default to local quickstart.""" - return os.environ.get("API_URL", DEFAULT_API_URL).rstrip("/") + """Read API URL from $API_URL or default to local quickstart. + + Validates that non-loopback URLs use HTTPS to prevent cleartext key leaks. + """ + url = os.environ.get("API_URL", DEFAULT_API_URL).rstrip("/") + + # Enforce HTTPS for non-loopback URLs (same as partner_api_key_bootstrap.py) + from urllib.parse import urlparse + parsed = urlparse(url) + is_loopback = parsed.hostname in ("localhost", "127.0.0.1", "::1") + + if parsed.scheme == "http" and not is_loopback: + raise SystemExit( + f"error: refusing to send CI_SMOKE_AGENT_KEY over cleartext HTTP to " + f"non-loopback host {parsed.hostname}. Use https:// or a loopback address." + ) + + return url def _post_json( @@ -164,6 +176,9 @@ def run_constant_test(api_url: str, agent_key: str, wallet_id: str, key_id: str) """Execute the constant test loop and assert all invariants.""" print(f"[constant-test] target: {api_url}", file=sys.stderr) + # Generate a unique run ID for this execution to make idempotency keys unique + run_id = datetime.now(timezone.utc).strftime('%Y%m%d-%H%M%S-%f') + # If credentials not provided, self-provision if not agent_key: print("[constant-test] self-provisioning agent key", file=sys.stderr) @@ -171,7 +186,7 @@ def run_constant_test(api_url: str, agent_key: str, wallet_id: str, key_id: str) provision_resp = _post_json( provision_client, "/v1/dev-keys/self-provision", - {"agent_id": f"constant-test-{datetime.now(timezone.utc).strftime('%Y%m%d-%H%M%S')}"}, + {"agent_id": f"constant-test-{run_id}"}, expected_status=201, ) agent_key = provision_resp["api_key"] @@ -181,6 +196,34 @@ def run_constant_test(api_url: str, agent_key: str, wallet_id: str, key_id: str) f"[constant-test] self-provisioned: wallet_id={wallet_id}", file=sys.stderr, ) + elif not wallet_id or not key_id: + # Key provided but wallet_id/key_id missing - fetch from API + print("[constant-test] fetching wallet_id and key_id from API", file=sys.stderr) + headers = { + "X-API-Key": agent_key, + "Content-Type": "application/json", + } + with httpx.Client(base_url=api_url, headers=headers, timeout=30.0) as fetch_client: + wallets_resp = _get_json(fetch_client, "/v1/billing/wallets", expected_status=200) + wallets = wallets_resp.get("wallets", []) + require(len(wallets) > 0, "no wallets found for this API key") + # Use the first wallet (agent wallet) + wallet_id = wallets[0]["wallet_id"] + + # Fetch key_id from the keys list + keys_resp = _get_json(fetch_client, f"/v1/billing/wallets/{wallet_id}/keys", expected_status=200) + keys = keys_resp.get("keys", []) + require(len(keys) > 0, "no keys found for this wallet") + # Find the key matching our API key prefix + key_prefix = agent_key.split("_")[0] + "_" + agent_key.split("_")[1] + matching_keys = [k for k in keys if k.get("key_prefix") == key_prefix] + require(len(matching_keys) > 0, f"no key found with prefix {key_prefix}") + key_id = matching_keys[0]["key_id"] + + print( + f"[constant-test] fetched: wallet_id={wallet_id}, key_id={key_id}", + file=sys.stderr, + ) # Never log or print the agent_key. headers = { @@ -193,20 +236,30 @@ def run_constant_test(api_url: str, agent_key: str, wallet_id: str, key_id: str) print("[constant-test] discovering tools", file=sys.stderr) tools = _get_json(client, "/mcp/tools.json", expected_status=200) tool_names = [tool["name"] for tool in tools.get("tools", [])] - require( - GOVERNED_TOOL in tool_names, - f"{GOVERNED_TOOL} not discoverable; is ENABLE_DOGFOOD_TOOL=true?", - ) + + # Select governed tool: prefer production partner.echo over dogfood partner.notes.write + governed_tool = None + if PRODUCTION_GOVERNED_TOOL in tool_names: + governed_tool = PRODUCTION_GOVERNED_TOOL + print(f"[constant-test] using production tool: {governed_tool}", file=sys.stderr) + elif DOGFOOD_GOVERNED_TOOL in tool_names: + governed_tool = DOGFOOD_GOVERNED_TOOL + print(f"[constant-test] using dogfood tool: {governed_tool} (ENABLE_DOGFOOD_TOOL=true)", file=sys.stderr) + else: + raise SmokeTestFailure( + f"neither {PRODUCTION_GOVERNED_TOOL} nor {DOGFOOD_GOVERNED_TOOL} discoverable. " + f"Available tools: {', '.join(tool_names)}" + ) # Issue scoped permit print("[constant-test] issuing scoped permit", file=sys.stderr) expires_at = ( datetime.now(timezone.utc) + timedelta(minutes=30) ).strftime("%Y-%m-%dT%H:%M:%SZ") - # Add Idempotency-Key header for permit creation + # Add Idempotency-Key header for permit creation (unique per run) permit_headers = { **headers, - "Idempotency-Key": f"constant-test-permit-{datetime.now(timezone.utc).isoformat()}", + "Idempotency-Key": f"constant-test-permit-{run_id}", } resp = client.post( "/v1/permits", @@ -214,8 +267,8 @@ def run_constant_test(api_url: str, agent_key: str, wallet_id: str, key_id: str) "issuer_wallet_id": wallet_id, "subject_wallet_id": wallet_id, "subject_key_id": key_id, - "allowed_tools": [GOVERNED_TOOL], - "scopes": [f"tool:{GOVERNED_TOOL}:invoke", "billing:charge"], + "allowed_tools": [governed_tool], + "scopes": [f"tool:{governed_tool}:invoke", "billing:charge"], "max_credits": PERMIT_MAX_CREDITS, "expires_at": expires_at, }, @@ -229,14 +282,15 @@ def run_constant_test(api_url: str, agent_key: str, wallet_id: str, key_id: str) permit_id = permit["permit_id"] print(f"[constant-test] permit_id={permit_id}", file=sys.stderr) - # Invoke the governed tool - print(f"[constant-test] invoking {GOVERNED_TOOL}", file=sys.stderr) + # Invoke the governed tool (unique idempotency key per run) + invoke_idempotency_key = f"constant-test-invoke-{run_id}" + print(f"[constant-test] invoking {governed_tool}", file=sys.stderr) call_body = _build_mcp_call( - request_id="constant-test-1", - tool=GOVERNED_TOOL, + request_id=f"constant-test-req-{run_id}", + tool=governed_tool, wallet_id=wallet_id, permit_id=permit_id, - idempotency_key="constant-test-invoke-1", + idempotency_key=invoke_idempotency_key, arguments={"text": "constant test loop"}, ) first_call = _post_json( @@ -273,7 +327,7 @@ def run_constant_test(api_url: str, agent_key: str, wallet_id: str, key_id: str) f"ledger debit {debit_amount} != receipt charge {-charged}", ) debits_before_replay = len( - [e for e in ledger["entries"] if GOVERNED_TOOL in e.get("description", "")] + [e for e in ledger["entries"] if governed_tool in e.get("description", "")] ) # Replay same idempotency key → same receipt_id, no second debit @@ -294,7 +348,7 @@ def run_constant_test(api_url: str, agent_key: str, wallet_id: str, key_id: str) [ e for e in ledger_after_replay["entries"] - if GOVERNED_TOOL in e.get("description", "") + if governed_tool in e.get("description", "") ] ) require( @@ -302,7 +356,7 @@ def run_constant_test(api_url: str, agent_key: str, wallet_id: str, key_id: str) "replay created a second debit", ) print( - f"[constant-test] replay OK: same receipt_id, no second debit", + "[constant-test] replay OK: same receipt_id, no second debit", file=sys.stderr, ) @@ -313,14 +367,14 @@ def run_constant_test(api_url: str, agent_key: str, wallet_id: str, key_id: str) if len(tool_names) > 1: print("[constant-test] attempting out-of-scope tool", file=sys.stderr) # Find a tool that's not the governed one - other_tool = next((t for t in tool_names if t != GOVERNED_TOOL), None) + other_tool = next((t for t in tool_names if t != governed_tool), None) if other_tool: denial_body = _build_mcp_call( - request_id="constant-test-deny", + request_id=f"constant-test-deny-{run_id}", tool=other_tool, wallet_id=wallet_id, permit_id=permit_id, - idempotency_key="constant-test-deny-1", + idempotency_key=f"constant-test-deny-{run_id}", arguments={}, ) denial_call = _post_json( @@ -342,7 +396,7 @@ def run_constant_test(api_url: str, agent_key: str, wallet_id: str, key_id: str) f"denial charged {denial_charged} credits (expected 0)", ) print( - f"[constant-test] denial OK: permit_tool_not_allowed, 0 credits charged", + "[constant-test] denial OK: permit_tool_not_allowed, 0 credits charged", file=sys.stderr, ) else: diff --git a/tests/test_constant_test_loop.py b/tests/test_constant_test_loop.py index e6063c9e..c0b5e1c5 100644 --- a/tests/test_constant_test_loop.py +++ b/tests/test_constant_test_loop.py @@ -16,7 +16,6 @@ import socket import subprocess import sys -import tempfile import time from pathlib import Path From e9f5a64d59688fe03a43b65dc58fe63cbc367e0e Mon Sep 17 00:00:00 2001 From: Cursor Agent Date: Thu, 20 Aug 2026 09:49:33 +0000 Subject: [PATCH 3/3] fix: address CodeRabbit review comments - Add ConfigurationError exception for exit status 2 - Validate HTTPS for both --api-url and $API_URL paths - Emit 'wallet_id' field in bootstrap --json (keep agent_wallet_id for compat) - Improve key prefix validation: use first 8 chars, validate format, require exactly one match - Update docs: use mktemp + cleanup instead of /tmp/agent-key.json - Update docs: use 'wallet_id' field from bootstrap JSON All configuration errors now exit with status 2. Key-only credential path validates key format before API lookup. Co-authored-by: Christopher --- docs/constant-test-loop.md | 42 ++++++++++++++++++------ scripts/constant_test_loop.py | 48 +++++++++++++++++++++++----- scripts/partner_api_key_bootstrap.py | 1 + 3 files changed, 74 insertions(+), 17 deletions(-) diff --git a/docs/constant-test-loop.md b/docs/constant-test-loop.md index a690f38c..1134d93d 100644 --- a/docs/constant-test-loop.md +++ b/docs/constant-test-loop.md @@ -27,21 +27,45 @@ The script self-provisions an agent key via `/v1/dev-keys/self-provision` when n Set `CI_SMOKE_AGENT_KEY` for a pre-provisioned agent credential. Optionally provide `CI_SMOKE_WALLET_ID` and `CI_SMOKE_KEY_ID` for faster startup (the script will fetch them from the API if not provided): ```bash -# Provision an agent key once (using bootstrap key) +# Provision an agent key once (using bootstrap key) and extract credentials in one pipeline export BOOTSTRAP_KEY="amw_live_..." -umask 077 # Ensure restrictive permissions for key file -python scripts/partner_api_key_bootstrap.py \ + +# Extract and set as CI secrets directly from the JSON output (no persistent file) +export CI_SMOKE_AGENT_KEY="$(python scripts/partner_api_key_bootstrap.py \ --api-url https://api.thisisatest.tech \ --agent-id ci-smoke-agent \ --key-name constant-test-loop \ --budget-credits 5000 \ - --json | tee /tmp/agent-key.json -chmod 600 /tmp/agent-key.json # Restrict to owner-only read/write + --json | jq -r .api_key)" -# Extract and set as CI secret -export CI_SMOKE_AGENT_KEY="$(jq -r .api_key /tmp/agent-key.json)" -export CI_SMOKE_WALLET_ID="$(jq -r .agent_wallet_id /tmp/agent-key.json)" -export CI_SMOKE_KEY_ID="$(jq -r .key_id /tmp/agent-key.json)" +export CI_SMOKE_WALLET_ID="$(python scripts/partner_api_key_bootstrap.py \ + --api-url https://api.thisisatest.tech \ + --agent-id ci-smoke-agent \ + --key-name constant-test-loop \ + --budget-credits 5000 \ + --json | jq -r .wallet_id)" + +export CI_SMOKE_KEY_ID="$(python scripts/partner_api_key_bootstrap.py \ + --api-url https://api.thisisatest.tech \ + --agent-id ci-smoke-agent \ + --key-name constant-test-loop \ + --budget-credits 5000 \ + --json | jq -r .key_id)" + +# Or use a restrictive temporary file if needed +KEY_FILE="$(mktemp)" +umask 077 +python scripts/partner_api_key_bootstrap.py \ + --api-url https://api.thisisatest.tech \ + --agent-id ci-smoke-agent \ + --key-name constant-test-loop \ + --budget-credits 5000 \ + --json > "$KEY_FILE" +chmod 600 "$KEY_FILE" +export CI_SMOKE_AGENT_KEY="$(jq -r .api_key "$KEY_FILE")" +export CI_SMOKE_WALLET_ID="$(jq -r .wallet_id "$KEY_FILE")" +export CI_SMOKE_KEY_ID="$(jq -r .key_id "$KEY_FILE")" +rm "$KEY_FILE" # Clean up immediately # Run the constant test API_URL=https://api.thisisatest.tech python scripts/constant_test_loop.py diff --git a/scripts/constant_test_loop.py b/scripts/constant_test_loop.py index e054d78d..25eed014 100755 --- a/scripts/constant_test_loop.py +++ b/scripts/constant_test_loop.py @@ -50,6 +50,10 @@ class SmokeTestFailure(RuntimeError): """Raised when an invariant fails during the smoke test.""" +class ConfigurationError(RuntimeError): + """Raised when configuration is invalid (exits with status 2).""" + + def require(condition: bool, message: str) -> None: """Fail immediately if condition is false.""" if not condition: @@ -91,8 +95,8 @@ def _get_api_url() -> str: is_loopback = parsed.hostname in ("localhost", "127.0.0.1", "::1") if parsed.scheme == "http" and not is_loopback: - raise SystemExit( - f"error: refusing to send CI_SMOKE_AGENT_KEY over cleartext HTTP to " + raise ConfigurationError( + f"refusing to send CI_SMOKE_AGENT_KEY over cleartext HTTP to " f"non-loopback host {parsed.hostname}. Use https:// or a loopback address." ) @@ -214,10 +218,18 @@ def run_constant_test(api_url: str, agent_key: str, wallet_id: str, key_id: str) keys_resp = _get_json(fetch_client, f"/v1/billing/wallets/{wallet_id}/keys", expected_status=200) keys = keys_resp.get("keys", []) require(len(keys) > 0, "no keys found for this wallet") - # Find the key matching our API key prefix - key_prefix = agent_key.split("_")[0] + "_" + agent_key.split("_")[1] + + # Derive key_prefix: first 8 characters (same as generate_api_key format) + # Validate key format before deriving prefix + if len(agent_key) < 8 or "_" not in agent_key: + raise ConfigurationError("malformed API key: expected format _") + key_prefix = agent_key[:8] + matching_keys = [k for k in keys if k.get("key_prefix") == key_prefix] - require(len(matching_keys) > 0, f"no key found with prefix {key_prefix}") + if len(matching_keys) == 0: + raise ConfigurationError(f"no key found with prefix {key_prefix}") + if len(matching_keys) > 1: + raise ConfigurationError(f"ambiguous: {len(matching_keys)} keys with prefix {key_prefix}") key_id = matching_keys[0]["key_id"] print( @@ -408,6 +420,23 @@ def run_constant_test(api_url: str, agent_key: str, wallet_id: str, key_id: str) print("[constant-test] ALL INVARIANTS HELD", file=sys.stderr) +def _validate_api_url(url: str) -> str: + """Validate API URL and enforce HTTPS for non-loopback hosts.""" + from urllib.parse import urlparse + + url = url.rstrip("/") + parsed = urlparse(url) + is_loopback = parsed.hostname in ("localhost", "127.0.0.1", "::1") + + if parsed.scheme == "http" and not is_loopback: + raise ConfigurationError( + f"refusing to send CI_SMOKE_AGENT_KEY over cleartext HTTP to " + f"non-loopback host {parsed.hostname}. Use https:// or a loopback address." + ) + + return url + + def main(argv: list[str] | None = None) -> int: """Entry point.""" parser = argparse.ArgumentParser(description=__doc__.splitlines()[0]) @@ -422,11 +451,14 @@ def main(argv: list[str] | None = None) -> int: ) args = parser.parse_args(argv) - api_url = args.api_url or _get_api_url() - agent_key, wallet_id, key_id = _get_agent_key() - try: + # Validate API URL (applies to both --api-url and $API_URL) + api_url = _validate_api_url(args.api_url or _get_api_url()) + agent_key, wallet_id, key_id = _get_agent_key() run_constant_test(api_url, agent_key, wallet_id, key_id) + except ConfigurationError as config_error: + print(f"\n[constant-test] configuration error: {config_error}", file=sys.stderr) + return 2 except SmokeTestFailure as failure: print(f"\n[constant-test] FAILED: {failure}", file=sys.stderr) return 1 diff --git a/scripts/partner_api_key_bootstrap.py b/scripts/partner_api_key_bootstrap.py index 59c47a5a..7ae85fc3 100644 --- a/scripts/partner_api_key_bootstrap.py +++ b/scripts/partner_api_key_bootstrap.py @@ -156,6 +156,7 @@ def provision( "api_url": base, "sponsor_wallet_id": sponsor_wallet_id, "agent_wallet_id": agent_wallet_id, + "wallet_id": agent_wallet_id, # Documented field name (agent_wallet_id kept for compat) "agent_id": agent_id, "key_id": key.get("key_id"), "key_prefix": key.get("key_prefix"),