diff --git a/src/pacta/cli.py b/src/pacta/cli.py index af4e320..c00b557 100644 --- a/src/pacta/cli.py +++ b/src/pacta/cli.py @@ -197,6 +197,52 @@ def build_parser() -> argparse.ArgumentParser: agent.add_argument("--max-sth-age-seconds", type=int, help="Reject signed tree heads older than this.") agent.add_argument("--require-verified-verifier", action="store_true", help="Fail closed unless Ed25519 verification ran on the dogfood (certificate-covered) verifier.") agent.set_defaults(func=cmd_agent) + + wallet = sub.add_parser("wallet", help="warden: the verified-custody wallet (quorum boundary + signing firewall).") + wsub = wallet.add_subparsers(dest="wallet_cmd", required=True) + + w_build = wsub.add_parser("build-quorum", help="Build the quorum verifier members from the pinned proven source workspaces.") + w_build.add_argument("--sources-root", required=True, help="Directory holding the pinned fork source checkouts.") + w_build.add_argument("--backends", nargs="*", help="Subset of dalek anza risc0 betrusted (default: all).") + w_build.add_argument("--timeout", type=int, default=900) + w_build.set_defaults(func=cmd_wallet_build_quorum) + + w_init = wsub.add_parser("init", help="Create a wallet - the R4 gate in executable form (refuses below end-to-end coverage).") + w_init.add_argument("--wallet", required=True) + w_init.add_argument("--evidence", required=True, help="Directory of .attestation.json + .receipt.json (e.g. from `pacta log-fetch`).") + w_init.add_argument("--log-public-key", required=True) + w_init.add_argument("--trusted-provider", required=True, help="Whose observations you trust (verdicts are always re-derived locally).") + w_init.add_argument("--repos-config", default="examples/repos.yaml") + w_init.add_argument("--backends", nargs="*") + w_init.add_argument("--require-tier", default="R4") + w_init.add_argument("--min-members", type=int, default=2) + w_init.add_argument("--freshness-days", type=int, default=30) + w_init.set_defaults(func=cmd_wallet_init) + + w_status = wsub.add_parser("status", help="Show wallet custody posture.") + w_status.add_argument("--wallet", required=True) + w_status.set_defaults(func=cmd_wallet_status) + + w_card = wsub.add_parser("card", help="Emit the self-proving custody card (optionally to a .well-known dir).") + w_card.add_argument("--wallet", required=True) + w_card.add_argument("--out-dir", help="Write .well-known/custody-card.json under this directory.") + w_card.add_argument("--log-url", default="https://ltl.zkdefi.org") + w_card.set_defaults(func=cmd_wallet_card) + + w_mcp = wsub.add_parser("mcp", help="Serve the agent-native MCP surface over stdio JSON-RPC.") + w_mcp.add_argument("--wallet", required=True) + w_mcp.add_argument("--log-url", default="https://ltl.zkdefi.org") + w_mcp.set_defaults(func=cmd_wallet_mcp) + + w_ledger = wsub.add_parser("verify-ledger", help="Re-check the wallet's hash-chained ledger integrity.") + w_ledger.add_argument("--wallet", required=True) + w_ledger.set_defaults(func=cmd_wallet_verify_ledger) + + w_unlatch = wsub.add_parser("unlatch", help="Release a tamper latch (deliberate operator act; the note is recorded permanently).") + w_unlatch.add_argument("--wallet", required=True) + w_unlatch.add_argument("--note", required=True) + w_unlatch.set_defaults(func=cmd_wallet_unlatch) + return parser @@ -544,6 +590,92 @@ def _log_accountability_checks( return diagnostics +def cmd_wallet_build_quorum(args: argparse.Namespace) -> int: + from .quorum import QUORUM_BACKENDS, build_quorum_member + + names = args.backends or list(QUORUM_BACKENDS) + failures = 0 + for name in names: + try: + prov = build_quorum_member(name, args.sources_root, timeout=args.timeout) + print(f"{name}: built source={(prov['source_commit'] or '?')[:12]} sha256={prov['binary_sha256'][:12]}") + except (RuntimeError, KeyError) as exc: + failures += 1 + print(f"{name}: FAILED {exc}", file=sys.stderr) + return 1 if failures else 0 + + +def cmd_wallet_init(args: argparse.Namespace) -> int: + from .wallet import Wallet, WalletError + + try: + wallet = Wallet.init( + args.wallet, + args.evidence, + args.log_public_key, + repos_config=args.repos_config, + trusted_provider=args.trusted_provider, + backends=args.backends, + require_tier=args.require_tier, + min_members=args.min_members, + freshness_max_age_days=args.freshness_days, + ) + except WalletError as exc: + print(f"R4 gate refused: {exc}", file=sys.stderr) + return 1 + capsule = wallet.capsule() + print(f"wallet created: {args.wallet}") + for member in capsule["members"]: + print(f" {member['backend']:>10} {member['risk_tier']} leaf {member['evidence']['leaf_index']} src {(member['source_commit'] or '?')[:12]}") + print(f" policy: unanimity of >={capsule['policy']['min_members']} at tier {capsule['policy']['require_tier']}") + return 0 + + +def cmd_wallet_status(args: argparse.Namespace) -> int: + from .wallet import Wallet + + print(json.dumps(Wallet(args.wallet).posture(), indent=2, sort_keys=True)) + return 0 + + +def cmd_wallet_card(args: argparse.Namespace) -> int: + from .custodycard import build_custody_card, write_well_known + from .wallet import Wallet + + wallet = Wallet(args.wallet) + if args.out_dir: + path = write_well_known(wallet, args.out_dir, args.log_url) + print(f"custody card: {path}") + else: + print(json.dumps(build_custody_card(wallet, args.log_url), indent=2, sort_keys=True)) + return 0 + + +def cmd_wallet_mcp(args: argparse.Namespace) -> int: + from .walletmcp import WalletMCP + + WalletMCP(args.wallet, log_url=args.log_url).serve_stdio() + return 0 + + +def cmd_wallet_verify_ledger(args: argparse.Namespace) -> int: + from .wallet import Wallet + + ok, problems = Wallet(args.wallet).verify_ledger() + print("ledger chain: " + ("intact" if ok else "BROKEN")) + for problem in problems: + print(f" - {problem}", file=sys.stderr) + return 0 if ok else 1 + + +def cmd_wallet_unlatch(args: argparse.Namespace) -> int: + from .wallet import Wallet + + Wallet(args.wallet).unlatch(args.note) + print("latch released; note recorded in the ledger") + return 0 + + def cmd_agent(args: argparse.Namespace) -> int: card = _card_for_agent(args) decision = run_agent_action( diff --git a/src/pacta/custodycard.py b/src/pacta/custodycard.py new file mode 100644 index 0000000..7bd57bd --- /dev/null +++ b/src/pacta/custodycard.py @@ -0,0 +1,196 @@ +"""The custody card and the proof-of-posture challenge. + +A2A publishes agent cards at ``/.well-known/agent-card.json`` and (since +v1.0) signs them. warden's card goes one step further: it is +**self-proving**. Alongside the description sits, for every quorum +member, the transparency-log inclusion proof binding the member's +attested source to a signed tree head. A counterparty does not have to +believe the card's adjectives - it can recompute the Merkle roots with +~40 lines of stdlib code and check the STH signature against the log key +it already pins. + +The proof-of-posture (PoP) challenge is the live counterpart: send a +nonce, get back a signed posture attestation (capsule hash, ledger head, +latch state, incident count) whose signature just passed the wallet's +own outbound firewall - with the quorum trail attached. A heartbeat you +can audit instead of trust. +""" + +from __future__ import annotations + +import hashlib +import json +from datetime import datetime, timezone +from pathlib import Path +from typing import Any + +from .wallet import Refusal, Wallet + +CARD_TYPE = "pacta.wallet.custody_card.v1" +POP_TYPE = "pacta.wallet.posture_attestation.v1" +WELL_KNOWN_PATH = ".well-known/custody-card.json" + + +def _now() -> str: + return datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ") + + +def _canonical(document: Any) -> bytes: + return json.dumps(document, sort_keys=True, separators=(",", ":")).encode("utf-8") + + +def build_custody_card(wallet: Wallet, log_url: str = "https://ltl.zkdefi.org") -> dict[str, Any]: + """The self-proving business card. Everything a counterparty agent needs + to evaluate this wallet is embedded or Merkle-bound; nothing requires + believing the operator.""" + capsule = wallet.capsule() + posture = wallet.posture() + identities = sorted(p.name.removesuffix(".pub.pem") for p in wallet.keys_dir.glob("*.pub.pem")) + return { + "type": CARD_TYPE, + "name": "warden", + "description": ( + "Verified-custody Ed25519 wallet. Inbound acceptance requires unanimity " + "of quorum members compiled from sources whose correctness certificates " + "are machine-checked in Lean 4 and replay-attested in a public RFC 9162 " + "transparency log. Outbound signatures pass the same quorum as a firewall " + "before release." + ), + "protocol": { + "mcp": { + "transport": "stdio", + "command": "pacta wallet mcp --wallet ", + "tools": [ + "wallet_status", + "verify_inbound", + "request_signature", + "custody_card", + "posture_challenge", + "list_incidents", + "explain_refusal", + ], + }, + "refusal_codes": [ + "EVIDENCE_REQUIRED", "POLICY_DENIED", "CUSTODY_LATCHED", + "EVIDENCE_STALE", "MALFORMED_INTENT", "SIGNER_UNAVAILABLE", + "FIREWALL_QUARANTINE", "PENDING_AIRGAP", + ], + }, + "identities": identities, + "quorum": { + "policy": capsule["policy"], + "members": [ + { + "backend": m["backend"], + "component": m["component"], + "semantics": m["semantics"], + "entry_point": m["entry_point"], + "source_commit": m["source_commit"], + "binary_sha256": m["binary_sha256"], + "risk_tier": m["risk_tier"], + # The self-proving part: recompute, don't believe. + "transparency_evidence": m["evidence"], + } + for m in capsule["members"] + ], + }, + "signing": capsule["signing"], + "state": { + "latch": posture["latch"], + "ledger_head": posture["ledger"]["head"], + "ledger_entries": posture["ledger"]["entries"], + "incidents": posture["incidents"], + }, + "evidence_endpoints": { + "log": log_url, + "log_docs": f"{log_url}/docs", + "verify_hint": ( + "for each member: leaf_hash = SHA256(0x00 || canonical attestation leaf); " + "walk inclusion_proof to sth.root_hash; verify sth signature against the " + "log public key you pin. Stdlib only; see the log's verify.py." + ), + }, + "honesty": [ + "verification path certificate-covered; signing path trusted base (attested artifact)", + "SHA-512 enters the theorems as an opaque oracle", + "wire parser outcomes are hypotheses; side channels and reproducible builds are R5, not claimed", + "ML-DSA (PQC) slot fail-closed: no proven implementation exists", + ], + "generated_at": _now(), + } + + +def write_well_known(wallet: Wallet, out_dir: str | Path, log_url: str = "https://ltl.zkdefi.org") -> Path: + out = Path(out_dir) / WELL_KNOWN_PATH + out.parent.mkdir(parents=True, exist_ok=True) + card = build_custody_card(wallet, log_url) + out.write_text(json.dumps(card, indent=2, sort_keys=True) + "\n", encoding="utf-8") + return out + + +def posture_challenge( + wallet: Wallet, + nonce: str, + key_name: str = "warden", +) -> dict[str, Any] | Refusal: + """Answer a counterparty's nonce with a firewalled, signed posture. + + The signature on the attestation is produced by the wallet's own + outbound path, so it carries the full quorum trail - the counterparty + sees not only *that* the wallet signed, but that four provably + -equivalent verifiers unanimously accepted the signature before it + left the building. If custody is latched, the challenge honestly + fails with a CUSTODY_LATCHED refusal instead of a heartbeat. + """ + if not isinstance(nonce, str) or not (8 <= len(nonce) <= 128): + raise ValueError("nonce must be a string of 8..128 characters") + posture = wallet.posture() + body = { + "type": POP_TYPE, + "nonce": nonce, + "posture": posture, + } + payload = _canonical(body) + intent = { + "purpose": f"proof-of-posture challenge response (nonce {nonce[:16]}...)", + "payload_sha256": hashlib.sha256(payload).hexdigest(), + } + result = wallet.request_signature(intent, payload, key_name=key_name) + if isinstance(result, Refusal): + return result + return { + "type": POP_TYPE, + "nonce": nonce, + "posture": posture, + "payload_hex": payload.hex(), + "signature_hex": result["signature_hex"], + "public_key_hex": result["public_key_hex"], + "identity": result["identity"], + "firewall": result["firewall"], + "issued_at": result["issued_at"], + } + + +def verify_posture_attestation(attestation: dict[str, Any], expected_nonce: str) -> tuple[bool, list[str]]: + """Counterparty-side check, stdlib only: nonce echo, payload binding. + + Signature verification is left to the counterparty's own verifier + (ideally its own quorum); this helper checks the structure so a + counterparty without crypto can still catch replay and splicing. + """ + problems: list[str] = [] + if attestation.get("type") != POP_TYPE: + problems.append(f"unexpected type {attestation.get('type')}") + if attestation.get("nonce") != expected_nonce: + problems.append("nonce mismatch (replay?)") + try: + payload = bytes.fromhex(str(attestation.get("payload_hex", ""))) + body = json.loads(payload) + except ValueError: + problems.append("payload_hex is not hex/JSON") + return False, problems + if body.get("nonce") != expected_nonce: + problems.append("signed payload does not bind the nonce") + if body.get("posture") != attestation.get("posture"): + problems.append("posture shown does not match posture signed") + return (not problems), problems diff --git a/src/pacta/walletmcp.py b/src/pacta/walletmcp.py new file mode 100644 index 0000000..9c4f93d --- /dev/null +++ b/src/pacta/walletmcp.py @@ -0,0 +1,260 @@ +"""warden's Model Context Protocol surface - the agent-native front door. + +A dependency-free MCP server over stdio JSON-RPC 2.0. It follows the AX +canon distilled in docs/agent-native.md: outcome-first tools, strict +input schemas, and results that carry their own evidence so a calling +agent never has to trust an adjective. Errors are structured objects +(``code`` / ``missing`` / ``remediation``), not prose - a refused agent +gets a machine-actionable receipt it can hand its principal. + +This is intentionally tiny and stdlib-only: an agent should be able to +read the whole trust surface in one sitting. It speaks enough of MCP +(``initialize``, ``tools/list``, ``tools/call``) to be driven by any MCP +client, and degrades to a plain JSON-RPC endpoint for scripts. +""" + +from __future__ import annotations + +import base64 +import json +import sys +from pathlib import Path +from typing import Any, Callable + +from .custodycard import build_custody_card, posture_challenge +from .wallet import Refusal, Wallet + +PROTOCOL_VERSION = "2025-06-18" +SERVER_INFO = {"name": "warden", "version": "1.0.0"} + + +def _b64_to_bytes(field: str, value: Any) -> bytes: + if not isinstance(value, str): + raise _ToolError("MALFORMED_INTENT", f"{field} must be base64 string", [field], "send base64") + try: + return base64.b64decode(value, validate=True) + except Exception as exc: # noqa: BLE001 + raise _ToolError("MALFORMED_INTENT", f"{field} is not valid base64: {exc}", [field], "re-encode as base64") + + +class _ToolError(Exception): + def __init__(self, code: str, reason: str, missing: list[str], remediation: str) -> None: + super().__init__(reason) + self.payload = {"code": code, "reason": reason, "missing": missing, "remediation": remediation} + + +TOOLS: list[dict[str, Any]] = [ + { + "name": "wallet_status", + "description": "Custody posture: quorum members and tiers, latch state, ledger head and chain integrity, incident and refusal counts. Read this first.", + "inputSchema": {"type": "object", "properties": {}, "additionalProperties": False}, + }, + { + "name": "verify_inbound", + "description": "Run the quorum on an inbound (payload, signature, public_key). Acceptance requires unanimity of the proven verifiers; divergence is classified and, if unexplained, latches custody. All inputs base64.", + "inputSchema": { + "type": "object", + "properties": { + "payload_b64": {"type": "string", "description": "message bytes, base64"}, + "signature_b64": {"type": "string", "description": "64-byte Ed25519 signature, base64"}, + "public_key_b64": {"type": "string", "description": "32-byte raw Ed25519 public key, base64"}, + "context": {"type": "string", "description": "free-text label recorded in the ledger"}, + }, + "required": ["payload_b64", "signature_b64", "public_key_b64"], + "additionalProperties": False, + }, + }, + { + "name": "request_signature", + "description": "Outbound signing with intent binding and the quorum firewall. Provide an intent.purpose (recorded as WHY) and the payload; the produced signature is verified by the quorum before release and quarantined if it fails.", + "inputSchema": { + "type": "object", + "properties": { + "payload_b64": {"type": "string", "description": "message bytes to sign, base64"}, + "purpose": {"type": "string", "description": "why this signature is requested (recorded in the ledger)"}, + "identity": {"type": "string", "description": "wallet identity name (default: warden)"}, + }, + "required": ["payload_b64", "purpose"], + "additionalProperties": False, + }, + }, + { + "name": "custody_card", + "description": "The self-proving business card: quorum membership with embedded transparency-log inclusion proofs a counterparty can recompute, signing provenance, honesty ledger. No trust required.", + "inputSchema": {"type": "object", "properties": {}, "additionalProperties": False}, + }, + { + "name": "posture_challenge", + "description": "Proof-of-posture: send a nonce (8..128 chars), receive a signed posture attestation whose signature passed the outbound firewall, with the full quorum trail attached. An auditable heartbeat.", + "inputSchema": { + "type": "object", + "properties": {"nonce": {"type": "string"}}, + "required": ["nonce"], + "additionalProperties": False, + }, + }, + { + "name": "list_incidents", + "description": "Quorum divergences and firewall quarantines recorded by this wallet, newest-first, with severities and trails.", + "inputSchema": {"type": "object", "properties": {}, "additionalProperties": False}, + }, + { + "name": "explain_refusal", + "description": "Fetch a previously issued refusal receipt by index, or the latest. Returns the machine-actionable receipt (code, missing, remediation, signature).", + "inputSchema": { + "type": "object", + "properties": {"index": {"type": "integer", "description": "receipt number; omit for latest"}}, + "additionalProperties": False, + }, + }, +] + + +class WalletMCP: + def __init__(self, wallet_dir: str | Path, log_url: str = "https://ltl.zkdefi.org") -> None: + self.wallet = Wallet(wallet_dir) + self.log_url = log_url + self.handlers: dict[str, Callable[[dict[str, Any]], dict[str, Any]]] = { + "wallet_status": self._wallet_status, + "verify_inbound": self._verify_inbound, + "request_signature": self._request_signature, + "custody_card": self._custody_card, + "posture_challenge": self._posture_challenge, + "list_incidents": self._list_incidents, + "explain_refusal": self._explain_refusal, + } + + # -- tool implementations ------------------------------------------------ + + def _wallet_status(self, _: dict[str, Any]) -> dict[str, Any]: + return self.wallet.posture() + + def _verify_inbound(self, args: dict[str, Any]) -> dict[str, Any]: + payload = _b64_to_bytes("payload_b64", args.get("payload_b64")) + signature = _b64_to_bytes("signature_b64", args.get("signature_b64")) + public_key = _b64_to_bytes("public_key_b64", args.get("public_key_b64")) + if len(signature) != 64: + raise _ToolError("MALFORMED_INTENT", "signature must be 64 bytes", ["signature_b64"], "send a 64-byte Ed25519 signature") + if len(public_key) != 32: + raise _ToolError("MALFORMED_INTENT", "public key must be 32 bytes", ["public_key_b64"], "send a 32-byte raw Ed25519 key") + result = self.wallet.verify_inbound(payload, signature, public_key, context=str(args.get("context", ""))) + return result.to_dict() + + def _request_signature(self, args: dict[str, Any]) -> dict[str, Any]: + import hashlib + + payload = _b64_to_bytes("payload_b64", args.get("payload_b64")) + purpose = args.get("purpose") + if not isinstance(purpose, str) or not purpose.strip(): + raise _ToolError("MALFORMED_INTENT", "purpose is required", ["purpose"], "state why the signature is requested") + intent = {"purpose": purpose, "payload_sha256": hashlib.sha256(payload).hexdigest()} + result = self.wallet.request_signature( + intent, payload, key_name=str(args.get("identity", "warden")) + ) + if isinstance(result, Refusal): + raise _ToolError( + result.code, result.reason, result.missing, result.remediation + ) + return result + + def _custody_card(self, _: dict[str, Any]) -> dict[str, Any]: + return build_custody_card(self.wallet, self.log_url) + + def _posture_challenge(self, args: dict[str, Any]) -> dict[str, Any]: + try: + result = posture_challenge(self.wallet, str(args.get("nonce", ""))) + except ValueError as exc: + raise _ToolError("MALFORMED_INTENT", str(exc), ["nonce"], "send an 8..128 character nonce") + if isinstance(result, Refusal): + raise _ToolError(result.code, result.reason, result.missing, result.remediation) + return result + + def _list_incidents(self, _: dict[str, Any]) -> dict[str, Any]: + incidents = [] + for path in sorted(self.wallet.incidents_dir.glob("*.json"), reverse=True): + incidents.append(json.loads(path.read_text(encoding="utf-8"))) + return {"incidents": incidents, "count": len(incidents)} + + def _explain_refusal(self, args: dict[str, Any]) -> dict[str, Any]: + receipts = sorted(self.wallet.receipts_dir.glob("*.json")) + if not receipts: + raise _ToolError("EVIDENCE_REQUIRED", "no refusal receipts issued yet", [], "there is nothing to explain") + idx = args.get("index") + if idx is None: + path = receipts[-1] + else: + match = [p for p in receipts if p.stem == f"{int(idx):04d}"] + if not match: + raise _ToolError("EVIDENCE_REQUIRED", f"no receipt {idx}", [], f"valid indices 0..{len(receipts)-1}") + path = match[0] + return json.loads(path.read_text(encoding="utf-8")) + + # -- JSON-RPC plumbing --------------------------------------------------- + + def handle(self, message: dict[str, Any]) -> dict[str, Any] | None: + method = message.get("method") + msg_id = message.get("id") + if method == "initialize": + return self._ok(msg_id, { + "protocolVersion": PROTOCOL_VERSION, + "capabilities": {"tools": {"listChanged": False}}, + "serverInfo": SERVER_INFO, + }) + if method in ("notifications/initialized", "initialized"): + return None + if method == "tools/list": + return self._ok(msg_id, {"tools": TOOLS}) + if method == "tools/call": + params = message.get("params") or {} + name = params.get("name") + args = params.get("arguments") or {} + handler = self.handlers.get(name) + if handler is None: + return self._err(msg_id, -32602, f"unknown tool {name}") + try: + payload = handler(args) + return self._ok(msg_id, { + "content": [{"type": "text", "text": json.dumps(payload, indent=2, sort_keys=True)}], + "structuredContent": payload, + "isError": False, + }) + except _ToolError as exc: + return self._ok(msg_id, { + "content": [{"type": "text", "text": json.dumps(exc.payload, indent=2, sort_keys=True)}], + "structuredContent": exc.payload, + "isError": True, + }) + except Exception as exc: # noqa: BLE001 - never crash the server + err = {"code": "INTERNAL", "reason": f"{type(exc).__name__}: {exc}", "missing": [], "remediation": "inspect the wallet directory"} + return self._ok(msg_id, { + "content": [{"type": "text", "text": json.dumps(err, indent=2, sort_keys=True)}], + "structuredContent": err, + "isError": True, + }) + if msg_id is None: + return None + return self._err(msg_id, -32601, f"method not found: {method}") + + @staticmethod + def _ok(msg_id: Any, result: dict[str, Any]) -> dict[str, Any]: + return {"jsonrpc": "2.0", "id": msg_id, "result": result} + + @staticmethod + def _err(msg_id: Any, code: int, message: str) -> dict[str, Any]: + return {"jsonrpc": "2.0", "id": msg_id, "error": {"code": code, "message": message}} + + def serve_stdio(self, stdin: Any = None, stdout: Any = None) -> None: + stdin = stdin or sys.stdin + stdout = stdout or sys.stdout + for line in stdin: + line = line.strip() + if not line: + continue + try: + message = json.loads(line) + except json.JSONDecodeError: + continue + response = self.handle(message) + if response is not None: + stdout.write(json.dumps(response) + "\n") + stdout.flush()