diff --git a/WALLET.md b/WALLET.md index 2798143..7930922 100644 --- a/WALLET.md +++ b/WALLET.md @@ -129,9 +129,11 @@ pacta wallet mcp --wallet ./my-warden # stdio JSON-RPC MCP server ## Agent-native surface (MCP) -`pacta wallet mcp` speaks MCP over stdio JSON-RPC. Seven outcome-first -tools; strict input schemas; results carry evidence; errors are structured -objects, never prose. +`pacta wallet mcp` speaks MCP over stdio JSON-RPC. Eight outcome-first +tools with read-only/destructive annotations; strict input schemas; +results carry evidence; errors are structured objects that include the +signed refusal receipt, never prose. Tool classes are rate-limited +(custody 30/min, verify 120/min, liveness 240/min). | tool | does | |---|---| @@ -142,10 +144,12 @@ objects, never prose. | `posture_challenge` | nonce → firewalled, signed posture attestation with the quorum trail | | `list_incidents` | divergences and quarantines, newest-first | | `explain_refusal` | fetch a refusal receipt by index (or latest) | +| `airgap_pending` | parked gap-signing requests and whether the device answered | Refusal codes (every refusal names one): `EVIDENCE_REQUIRED`, `POLICY_DENIED`, `CUSTODY_LATCHED`, `EVIDENCE_STALE`, `MALFORMED_INTENT`, -`SIGNER_UNAVAILABLE`, `FIREWALL_QUARANTINE`, `PENDING_AIRGAP`. +`SIGNER_UNAVAILABLE`, `FIREWALL_QUARANTINE`, `PENDING_AIRGAP`, +`RATE_LIMITED`. ### The custody card is self-proving @@ -158,6 +162,54 @@ log's own `verify.py` for the ~40-line client side. --- +## The spending policy (POLICY_DENIED) + +Optional `policy.json` in the wallet directory - the rules you would give +a teenager with a debit card, checked before the signer ever runs: + +```json +{ + "outbound": { + "max_amount_per_request": 100.0, + "max_amount_per_day": 500.0, + "counterparty_allowlist": ["alice"], + "counterparty_denylist": ["mallory"] + }, + "identities": { "warden": { "max_amount_per_request": 10.0 } }, + "ledger": { "rotate_at": 100000 } +} +``` + +Amount rules bind on `intent.amount`; list rules bind on +`intent.counterparty` - policy makes those fields **mandatory**, so a +request that omits them is refused, never waved past. Daily ceilings sum +the released amounts in the ledger's last 24 hours per identity. +Per-identity overrides win over `outbound` defaults. No `policy.json` +means unrestricted (and `wallet_status` says so). `pacta wallet policy +--wallet --init-template` writes a starter file. + +## Surface controls and the ledger's diet + +The MCP layer rate-limits by tool class (custody 30/min, verify 120/min, +liveness 240/min) so a hostile counterparty cannot grind the signer or +bloat the audit trail; rate refusals are surface events, not custody +events, and are not ledgered. The ledger itself appends in O(1) (only +the tail is read, under a dedicated lock file, fsynced) and rotates into +hash-chained archive segments at `ledger.rotate_at` entries - +`verify-ledger` walks the whole chain across segments back to genesis. + +## warden-treasury (Solana) + +`pacta wallet treasury-verify` takes a wire-format Solana transaction +(from a file, or fetched by signature via `--rpc-url`), parses it locally +(stdlib; legacy and v0), and quorum-verifies **every required signature** +over the exact message bytes - including through the `anza` member, the +certificate-covered verify path of the code Solana validators run. The +RPC is demoted from oracle to bandwidth: it can withhold transactions +(the verdict names this completeness gap explicitly), but it cannot +manufacture one the quorum will accept. Every check lands in the ledger +with a treasury context. + ## The signing firewall (verify-after-sign) Outbound is: **intent → sign → firewall → release**. @@ -201,7 +253,9 @@ alarm: ## Product lineup warden ships as one core with four production-ready deployment profiles — -see [docs/products.md](docs/products.md). In one line each: +see [docs/products.md](docs/products.md). Operational docs: the attacker +matrix in [docs/threat-model.md](docs/threat-model.md) and the +[latch-recovery runbook](docs/runbook-latch.md). In one line each: - **warden-solo** — a single agent's custody sidecar (local signer). - **warden-airgap** — signing behind a Precursor/Betrusted hardware gap. diff --git a/docs/products.md b/docs/products.md index 94adcbd..37939bb 100644 --- a/docs/products.md +++ b/docs/products.md @@ -73,9 +73,11 @@ the bytes, re-derive the verdict, with a verifier you hold a proof about. compromised or lying RPC can withhold data but cannot manufacture a signature the quorum will accept. -**Ready because:** the anza member is built and tested; wiring it to a -transaction feed is deployment configuration, not new trust surface. (The -chain-adapter layer is the documented integration point.) +**Ready because:** it is built: `pacta wallet treasury-verify` parses +wire-format transactions (legacy + v0, stdlib only) and quorum-verifies +every required signature, with the completeness gap (an RPC can withhold) +named in every verdict. The wire parser is ~120 lines of declared trusted +base, exactly like the forks' own parsers are hypotheses of the theorems. --- @@ -104,9 +106,9 @@ scoped as the next build.) ## Honesty about "production-ready" -Profiles 1 and 2 run end-to-end today on the tested core. Profiles 3 and 4 -are complete *product definitions* on the same core with one documented -integration point each (a chain-transaction adapter; a gossip transport) — +Profiles 1, 2, and 3 run end-to-end today on the tested core. Profile 4 +is a complete *product definition* on the same core with one documented +integration point (a gossip transport) — named here so the boundary between "built and tested" and "wired to your environment" is exact, which is the whole ethos of this project. None of them changes the trust posture; all of them fail closed. diff --git a/docs/runbook-latch.md b/docs/runbook-latch.md new file mode 100644 index 0000000..a9ddc63 --- /dev/null +++ b/docs/runbook-latch.md @@ -0,0 +1,78 @@ +# Runbook: the custody latch fired + +You are here because outbound custody is frozen and every signing request +returns `CUSTODY_LATCHED`. This is the wallet doing its job: an +unexplained quorum divergence or a firewall quarantine occurred, and the +wallet refuses to certify anything — including its own refusals, which +now arrive unsigned on purpose. + +**Do not unlatch first. Diagnose first.** The latch is cheap; a released +forged signature is not. + +## 1. Read what happened (2 minutes) + +```bash +pacta wallet status --wallet # latch reason + incident ref +cat /latch.json +cat /incidents/.json # the full divergence trail +ls /quarantine/ # any withheld signatures +pacta wallet verify-ledger --wallet # is the history itself intact? +``` + +The incident file names, per quorum member, its verdict and its binary +hash at the moment of divergence. That table is your suspect list. + +## 2. Classify (the incident file already did; check its work) + +- `classification: semantic-edge` (severity `note`) — the input hit a + documented degenerate class (small-order R, non-canonical s). This does + NOT latch by itself; if you are latched, something else also happened. +- `classification: unexplained` (severity `tamper`) — members disagreed + with no documented reason, or one errored. Assume fault or tampering + until shown otherwise. + +## 3. Investigate the three usual suspects, in order + +1. **A corrupted/updated member binary.** Compare each member's current + hash against the capsule: + `sha256sum dogfood/state/quorum/pacta-verify-*` vs + `capsule.json` → `members[].binary_sha256`. A mismatch on exactly the + dissenting member is the common benign case (a rebuild happened); + a mismatch you cannot explain is not benign. +2. **Hardware/memory fault.** Re-run the exact input from the incident + file through the quorum (`payload_sha256`, `signature_hex`, + `public_key_hex` are all recorded). A divergence that does not + reproduce points at a transient fault; log that finding in the + unlatch note. +3. **Actual tampering.** Divergence reproduces, hashes match the capsule, + input is not a documented edge → treat the host as suspect: rebuild + members from pinned sources on a machine you trust, re-run, compare. + +## 4. Remediate + +- Benign rebuild drift → rebuild all members (`pacta wallet + build-quorum`), then **re-init or re-seal** the capsule so the pins + match reality again. +- Transient fault → document it; consider the machine's RAM. +- Suspected tamper → do not unlatch on this host. Preserve the wallet + directory (it is the evidence), stand up a fresh wallet from fresh + builds + fresh evidence elsewhere. + +## 5. Unlatch — a deliberate, recorded act + +```bash +pacta wallet unlatch --wallet --note "" +``` + +The note is permanent: it lands in the hash-chained ledger next to the +latch it releases, and shows up in every future audit. Write it for the +auditor you hope never needs it. An empty or lazy note defeats the +design; the CLI requires the flag, your discipline supplies the content. + +## 6. Afterwards + +Re-run a signing smoke test and confirm `unanimous-accept`; check +`pacta wallet status` shows `latched: false`, chain intact, and the +incident count where you expect it. If this wallet participates in a +choir, expect peers to ask about the head gap — that is the system +working. diff --git a/docs/threat-model.md b/docs/threat-model.md new file mode 100644 index 0000000..631a15c --- /dev/null +++ b/docs/threat-model.md @@ -0,0 +1,49 @@ +# warden threat model + +Who can attack the wallet, what each attacker can and cannot achieve, and +which control stops them. "Proven" below means certificate-covered by the +Lean corpus attested in the transparency log; everything else is named +trusted base. The honest summary first: **warden's strongest guarantees +are about deciding what to believe (inbound) and catching its own signer +lying (firewall); an attacker who fully owns the host owns the wallet.** + +## Attacker matrix + +| # | attacker controls | can achieve | cannot achieve | stopped / bounded by | +|---|---|---|---|---| +| 1 | the network path | delay, reorder, withhold traffic | forge an accepted authorization | quorum verification (proven); freshness policy bounds staleness | +| 2 | the RPC provider (treasury) | withhold transactions (completeness) | fabricate a transaction the wallet believes | quorum re-verification of every signature (proven); verdict notes the completeness gap explicitly | +| 3 | a counterparty agent | spam requests; submit garbage | drain funds; grind the ledger unboundedly | MCP rate limiter (surface control); intent binding; spending policy (POLICY_DENIED); refusals cheap, custody calls scarce | +| 4 | ONE quorum member binary on disk (swap/corrupt) | cause divergence | pass a forged signature (needs all four) | capsule hash-pin catches swap at assembly; unexplained divergence latches custody | +| 5 | the signing key (exfiltrated) | sign arbitrary payloads AS this wallet elsewhere | make THIS wallet release out-of-policy signatures | policy + intent gates still bind the wallet's own releases; key custody itself is trusted base — use the airgap profile when this attacker is in scope | +| 6 | the airgap channel (tampered device response) | return a wrong signature | get it released | the firewall: verify-after-sign through the quorum; wrong signature is quarantined and latches | +| 7 | the wallet host (root) | everything: edit ledger, replace binaries AND capsule, read keys | escape detection by an OFF-HOST copy of the ledger head | out of scope for on-host controls — this is the warden-choir rationale (cross-witnessed heads) and the airgap rationale (key not on host) | +| 8 | the log operator (LTL) | attest falsely that proofs re-check | change verdicts (consumers re-derive from cones); rewrite history unnoticed (append-only, witnessed) | observation-not-verdict; STH pinning; witness mirror | +| 9 | the wallet operator (insider) | unlatch carelessly; loosen policy.json | do either invisibly | latch/unlatch and policy live in artifacts; unlatch requires a note recorded permanently in the hash-chained ledger | + +## What is proven vs. trusted, one line each + +- **Proven** (Lean certificates, log-attested): the verify path of each of + the four quorum members — field arithmetic through the full signature + acceptance equation. +- **Trusted base**: the signing path (attested artifact, fenced by the + firewall); SHA-512 (oracle); wire parsers in the forks (hypotheses) and + the ~120-line Solana wire parser in treasury mode; the OS, filesystem, + and Python runtime; compilers; key custody. +- **Deliberately absent**: reproducible builds, side-channel hardening + (R5 frontier); ML-DSA (fail-closed — no proven implementation exists). + +## Design invariants the controls enforce + +1. **Unanimity or nothing**: no majority voting; any divergence fails + closed. A lone honest member is sufficient to block. +2. **A signature that failed the firewall never leaves the process** — + quarantined bytes, latched custody. +3. **Nothing custody-relevant is un-ledgered**; nothing in the ledger can + be edited without breaking the chain (single-flight lock, fsync, + rotation with chained segments). +4. **Refusals are receipts**: machine-actionable, signed when the wallet + still trusts its own boundary, deliberately unsigned when latched. +5. **Policy failures are named** (`POLICY_DENIED`), never silent, and + policy rules make their inputs mandatory rather than waving through + requests that omit them. diff --git a/notebooks/10_verified_custody_wallet.ipynb b/notebooks/10_verified_custody_wallet.ipynb index e3b5133..462f616 100644 --- a/notebooks/10_verified_custody_wallet.ipynb +++ b/notebooks/10_verified_custody_wallet.ipynb @@ -253,6 +253,59 @@ " print(\"The counterparty believed no adjective; it recomputed a Merkle root.\")\n" ] }, + { + "cell_type": "markdown", + "metadata": {}, + "source": [ + "## Corrupt a member, watch the pin catch it (executable)\n", + "\n", + "The wallet seals each member's SHA-256 into its capsule. The\n", + "next cell stages a COPY of a real member binary in a temp\n", + "directory, \"seals\" its hash the way the capsule does, appends\n", + "one byte (a supply-chain attack in miniature), and re-checks.\n", + "Nothing on your machine is modified.\n" + ] + }, + { + "cell_type": "code", + "execution_count": null, + "metadata": {}, + "outputs": [], + "source": [ + "import hashlib, shutil, tempfile, pathlib\n", + "from pacta.quorum import binary_path\n", + "\n", + "member = binary_path(\"dalek\")\n", + "if not member.exists():\n", + " print(\"quorum not built; run pacta wallet build-quorum first\")\n", + "else:\n", + " stage = pathlib.Path(tempfile.mkdtemp()) / member.name\n", + " shutil.copy2(member, stage)\n", + " sealed = hashlib.sha256(stage.read_bytes()).hexdigest() # capsule pin\n", + " print(\"sealed :\", sealed[:24], \"...\")\n", + " with stage.open(\"ab\") as f:\n", + " f.write(b\"\\x00\") # the attack\n", + " current = hashlib.sha256(stage.read_bytes()).hexdigest()\n", + " print(\"current:\", current[:24], \"...\")\n", + " if current != sealed:\n", + " print(\"PIN CAUGHT IT: wallet.quorum() would refuse to assemble ->\")\n", + " print(\" 'quorum member dalek binary hash changed since the capsule was sealed'\")\n", + " else:\n", + " print(\"impossible: SHA-256 collision\")\n" + ] + }, + { + "cell_type": "markdown", + "metadata": {}, + "source": [ + "One appended byte and the wallet refuses to even *assemble* the\n", + "quorum - before any verification runs. Note what this control\n", + "is and is not: it stops binary substitution *between* wallet\n", + "sessions; an attacker with live root outranks it (see\n", + "docs/threat-model.md, attacker #7 - that is what the choir and\n", + "the airgap profiles are for).\n" + ] + }, { "cell_type": "markdown", "metadata": {}, @@ -261,9 +314,11 @@ "\n", "- Change `toy_quorum` to majority voting and write two sentences\n", " on exactly which attack that lets through.\n", - "- Take the real quorum cell and corrupt one member binary on\n", - " disk (append a byte). Predict, then observe, what the wallet's\n", - " capsule hash-pin does the next time it assembles the quorum.\n", + "- Extend the corrupt-a-member cell: corrupt the capsule JSON\n", + " itself instead of the binary. What catches that, and when?\n", + " (Hint: nothing does until the ledger genesis is compared -\n", + " write down the exact trust statement the capsule hash in the\n", + " genesis entry provides.)\n", "- The signing path is trusted base. Write the strongest *true*\n", " sentence you can about warden's outbound safety, and the\n", " strongest *false* one a marketer would write - and name the\n", diff --git a/scripts/build_curriculum_notebooks.py b/scripts/build_curriculum_notebooks.py index 6adeaf6..c806041 100644 --- a/scripts/build_curriculum_notebooks.py +++ b/scripts/build_curriculum_notebooks.py @@ -2245,15 +2245,62 @@ COURSE = { print("The counterparty believed no adjective; it recomputed a Merkle root.") """ ), + md( + """ + ## Corrupt a member, watch the pin catch it (executable) + + The wallet seals each member's SHA-256 into its capsule. The + next cell stages a COPY of a real member binary in a temp + directory, "seals" its hash the way the capsule does, appends + one byte (a supply-chain attack in miniature), and re-checks. + Nothing on your machine is modified. + """ + ), + code( + """ + import hashlib, shutil, tempfile, pathlib + from pacta.quorum import binary_path + + member = binary_path("dalek") + if not member.exists(): + print("quorum not built; run pacta wallet build-quorum first") + else: + stage = pathlib.Path(tempfile.mkdtemp()) / member.name + shutil.copy2(member, stage) + sealed = hashlib.sha256(stage.read_bytes()).hexdigest() # capsule pin + print("sealed :", sealed[:24], "...") + with stage.open("ab") as f: + f.write(b"\\x00") # the attack + current = hashlib.sha256(stage.read_bytes()).hexdigest() + print("current:", current[:24], "...") + if current != sealed: + print("PIN CAUGHT IT: wallet.quorum() would refuse to assemble ->") + print(" 'quorum member dalek binary hash changed since the capsule was sealed'") + else: + print("impossible: SHA-256 collision") + """ + ), + md( + """ + One appended byte and the wallet refuses to even *assemble* the + quorum - before any verification runs. Note what this control + is and is not: it stops binary substitution *between* wallet + sessions; an attacker with live root outranks it (see + docs/threat-model.md, attacker #7 - that is what the choir and + the airgap profiles are for). + """ + ), md( """ ## Exercises - Change `toy_quorum` to majority voting and write two sentences on exactly which attack that lets through. - - Take the real quorum cell and corrupt one member binary on - disk (append a byte). Predict, then observe, what the wallet's - capsule hash-pin does the next time it assembles the quorum. + - Extend the corrupt-a-member cell: corrupt the capsule JSON + itself instead of the binary. What catches that, and when? + (Hint: nothing does until the ledger genesis is compared - + write down the exact trust statement the capsule hash in the + genesis entry provides.) - The signing path is trusted base. Write the strongest *true* sentence you can about warden's outbound safety, and the strongest *false* one a marketer would write - and name the diff --git a/src/pacta/cli.py b/src/pacta/cli.py index abb7534..8297a8a 100644 --- a/src/pacta/cli.py +++ b/src/pacta/cli.py @@ -244,6 +244,19 @@ def build_parser() -> argparse.ArgumentParser: w_unlatch.add_argument("--note", required=True) w_unlatch.set_defaults(func=cmd_wallet_unlatch) + w_policy = wsub.add_parser("policy", help="Show the wallet's spending policy (policy.json), or write a starter template.") + w_policy.add_argument("--wallet", required=True) + w_policy.add_argument("--init-template", action="store_true", help="Write a commented starter policy.json (refuses to overwrite).") + w_policy.set_defaults(func=cmd_wallet_policy) + + w_treasury = wsub.add_parser("treasury-verify", help="Quorum-verify every signature of a Solana transaction (RPC demoted to bandwidth).") + w_treasury.add_argument("--wallet", required=True) + group = w_treasury.add_mutually_exclusive_group(required=True) + group.add_argument("--tx-file", help="File with wire-format transaction bytes (raw or base64).") + group.add_argument("--tx-sig", help="Transaction signature (base58) to fetch via --rpc-url.") + w_treasury.add_argument("--rpc-url", help="Solana JSON-RPC endpoint (required with --tx-sig).") + w_treasury.set_defaults(func=cmd_wallet_treasury_verify) + return parser @@ -677,6 +690,60 @@ def cmd_wallet_unlatch(args: argparse.Namespace) -> int: return 0 +POLICY_TEMPLATE = { + "outbound": { + "max_amount_per_request": 100.0, + "max_amount_per_day": 500.0, + "counterparty_allowlist": ["example-counterparty-id"], + "counterparty_denylist": [], + }, + "identities": {}, + "ledger": {"rotate_at": 100000}, +} + + +def cmd_wallet_policy(args: argparse.Namespace) -> int: + from .wallet import Wallet + + wallet = Wallet(args.wallet) + path = wallet.dir / "policy.json" + if args.init_template: + if path.exists(): + print(f"refusing to overwrite existing {path}", file=sys.stderr) + return 1 + path.write_text(json.dumps(POLICY_TEMPLATE, indent=2, sort_keys=True) + "\n", encoding="utf-8") + print(f"starter policy written: {path} - edit it, the rules you would give a teenager with a debit card") + return 0 + policy = wallet.policy() + if not policy: + print("no policy.json - outbound is unrestricted (run with --init-template for a starter)") + return 0 + print(json.dumps(policy, indent=2, sort_keys=True)) + return 0 + + +def cmd_wallet_treasury_verify(args: argparse.Namespace) -> int: + import base64 as _b64 + + from .treasury import fetch_transaction, verify_transaction + from .wallet import Wallet + + if args.tx_sig and not args.rpc_url: + print("--tx-sig requires --rpc-url", file=sys.stderr) + return 2 + if args.tx_file: + raw = Path(args.tx_file).read_bytes() + try: + tx = _b64.b64decode(raw, validate=True) + except Exception: # noqa: BLE001 - not base64, treat as wire bytes + tx = raw + else: + tx = fetch_transaction(args.rpc_url, args.tx_sig) + verdict = verify_transaction(Wallet(args.wallet), tx) + print(json.dumps(verdict.to_dict(), indent=2, sort_keys=True)) + return 0 if verdict.authentic else 1 + + def cmd_agent(args: argparse.Namespace) -> int: card = _card_for_agent(args) decision = run_agent_action( diff --git a/src/pacta/treasury.py b/src/pacta/treasury.py new file mode 100644 index 0000000..eecc538 --- /dev/null +++ b/src/pacta/treasury.py @@ -0,0 +1,196 @@ +"""warden-treasury: trust-minimized Solana transaction watching. + +An agent that must believe on-chain state ("did my deposit land?") today +asks an RPC provider and trusts the answer. Treasury mode demotes the RPC +from oracle to bandwidth: it takes the raw transaction bytes, parses the +wire format locally (stdlib only), and re-verifies every required +signature through the wallet's quorum - which includes the ``anza`` +member, the certificate-covered verify path of the code Solana +validators themselves run. A lying RPC can withhold a transaction, but it +cannot manufacture one the quorum will accept. + +Honesty ledger for this module: signature verification is custody-grade +(the quorum); the WIRE PARSING here (base58, compact-u16, the message +header) is ~120 lines of stdlib Python and is trusted base, exactly like +the wire parsers inside the forks are hypotheses of the theorems. A +malicious RPC also controls *which* transactions you see (completeness); +treasury verification establishes authenticity, not completeness. +""" + +from __future__ import annotations + +import json +import urllib.request +from dataclasses import dataclass, field +from typing import Any + +_B58_ALPHABET = "123456789ABCDEFGHJKLMNPQRSTUVWXYZabcdefghijkmnopqrstuvwxyz" +_B58_INDEX = {c: i for i, c in enumerate(_B58_ALPHABET)} + + +def b58decode(text: str) -> bytes: + """Base58 (Bitcoin/Solana alphabet), stdlib only.""" + value = 0 + for char in text: + if char not in _B58_INDEX: + raise ValueError(f"invalid base58 character {char!r}") + value = value * 58 + _B58_INDEX[char] + raw = value.to_bytes((value.bit_length() + 7) // 8, "big") if value else b"" + pad = len(text) - len(text.lstrip("1")) + return b"\x00" * pad + raw + + +def _compact_u16(data: bytes, offset: int) -> tuple[int, int]: + """Solana's compact-u16 (shortvec) length encoding.""" + result = 0 + shift = 0 + while True: + if offset >= len(data): + raise ValueError("truncated compact-u16") + byte = data[offset] + offset += 1 + result |= (byte & 0x7F) << shift + if byte & 0x80 == 0: + return result, offset + shift += 7 + if shift > 14: + raise ValueError("compact-u16 too long") + + +@dataclass(slots=True) +class ParsedTransaction: + signatures: list[bytes] + message: bytes # the exact bytes the signatures sign + account_keys: list[bytes] # 32-byte Ed25519 keys, signers first + num_required_signatures: int + version: int | None = None # None = legacy, 0 = v0 + + +def parse_transaction(tx: bytes) -> ParsedTransaction: + """Parse a wire-format Solana transaction (legacy or v0). + + Layout: compact-u16 count of 64-byte signatures, then the message. + Message: optional version byte (high bit set), 3-byte header + (num_required_signatures, num_readonly_signed, num_readonly_unsigned), + compact-u16 count of 32-byte account keys, 32-byte recent blockhash, + instructions (not needed here - the signed payload is the whole + message, byte for byte). + """ + count, offset = _compact_u16(tx, 0) + if count == 0 or count > 12: + raise ValueError(f"implausible signature count {count}") + signatures = [] + for _ in range(count): + if offset + 64 > len(tx): + raise ValueError("truncated signature section") + signatures.append(tx[offset:offset + 64]) + offset += 64 + message = tx[offset:] + if not message: + raise ValueError("empty message") + pos = 0 + version: int | None = None + if message[0] & 0x80: + version = message[0] & 0x7F + if version != 0: + raise ValueError(f"unsupported transaction version {version}") + pos = 1 + if pos + 3 > len(message): + raise ValueError("truncated message header") + num_required = message[pos] + pos += 3 + key_count, pos = _compact_u16(message, pos) + if key_count < num_required: + raise ValueError("fewer account keys than required signatures") + keys = [] + for _ in range(key_count): + if pos + 32 > len(message): + raise ValueError("truncated account keys") + keys.append(message[pos:pos + 32]) + pos += 32 + if len(signatures) != num_required: + raise ValueError( + f"signature count {len(signatures)} != header num_required_signatures {num_required}" + ) + return ParsedTransaction(signatures, message, keys, num_required, version) + + +@dataclass(slots=True) +class TreasuryVerdict: + authentic: bool + signer_results: list[dict[str, Any]] = field(default_factory=list) + version: int | None = None + + def to_dict(self) -> dict[str, Any]: + return { + "type": "pacta.wallet.treasury_verdict.v1", + "authentic": self.authentic, + "signer_results": self.signer_results, + "transaction_version": self.version, + "note": ( + "authenticity only: every required signature was quorum-verified over the " + "message bytes; an RPC can still withhold transactions (no completeness claim)" + ), + } + + +def verify_transaction(wallet: Any, tx: bytes, state_dir: Any = None) -> TreasuryVerdict: + """Quorum-verify every required signature of a wire-format transaction. + + Fail-closed: authentic only if EVERY required signature is a unanimous + quorum accept. Each check lands in the wallet ledger with a treasury + context, so the audit trail shows exactly which chain facts this + wallet chose to believe, and on whose mathematics. + """ + parsed = parse_transaction(tx) + results = [] + authentic = True + for i, signature in enumerate(parsed.signatures): + signer_key = parsed.account_keys[i] + outcome = wallet.verify_inbound( + parsed.message, + signature, + signer_key, + context=f"treasury: tx signer {i}", + state_dir=state_dir, + ) + results.append({ + "signer_index": i, + "signer_key_hex": signer_key.hex(), + "classification": outcome.classification, + "accepted": outcome.accepted, + }) + authentic = authentic and outcome.accepted + return TreasuryVerdict(authentic, results, parsed.version) + + +def fetch_transaction(rpc_url: str, signature_b58: str, timeout: int = 20) -> bytes: + """Fetch raw transaction bytes from a Solana JSON-RPC endpoint. + + The response is used as BYTES ONLY - nothing the RPC says about + status, slots, or balances is consumed here. Trust demotion is the + point: the RPC hands over an envelope; the quorum decides. + """ + import base64 + + request = json.dumps({ + "jsonrpc": "2.0", + "id": 1, + "method": "getTransaction", + "params": [ + signature_b58, + {"encoding": "base64", "maxSupportedTransactionVersion": 0}, + ], + }).encode("utf-8") + req = urllib.request.Request( + rpc_url, data=request, headers={"Content-Type": "application/json"} + ) + with urllib.request.urlopen(req, timeout=timeout) as response: + body = json.loads(response.read()) + result = body.get("result") + if not result: + raise ValueError(f"RPC returned no transaction for {signature_b58}") + encoded = (result.get("transaction") or [None])[0] + if not isinstance(encoded, str): + raise ValueError("RPC response has no base64 transaction body") + return base64.b64decode(encoded) diff --git a/src/pacta/wallet.py b/src/pacta/wallet.py index c1afd00..9766724 100644 --- a/src/pacta/wallet.py +++ b/src/pacta/wallet.py @@ -77,6 +77,7 @@ REFUSAL_CODES = ( "SIGNER_UNAVAILABLE", # signer backend cannot serve the request "FIREWALL_QUARANTINE", # produced signature failed the quorum firewall "PENDING_AIRGAP", # request parked in the airgap outbox + "RATE_LIMITED", # surface-level throttle (issued by the MCP layer) ) @@ -215,8 +216,9 @@ class PendingAirgap(Exception): class Wallet: - def __init__(self, wallet_dir: str | Path) -> None: + def __init__(self, wallet_dir: str | Path, state_dir: str | Path | None = None) -> None: self.dir = Path(wallet_dir) + self.state_dir = state_dir # default quorum binary location override self.capsule_path = self.dir / "capsule.json" self.ledger_path = self.dir / "ledger.jsonl" self.latch_path = self.dir / "latch.json" @@ -451,40 +453,105 @@ class Wallet: if line.strip() ] + @staticmethod + def _last_ledger_line(handle: Any) -> dict[str, Any] | None: + """Read only the final line - O(1) in ledger size, not O(n). + + Seeks back from EOF in one chunk (ledger lines are well under 64 KiB; + a single quarantine-size body is ~2 KiB) and parses the last + newline-terminated record.""" + handle.seek(0, os.SEEK_END) + size = handle.tell() + if size == 0: + return None + back = min(size, 65536) + handle.seek(size - back) + tail = handle.read(back) + lines = [line for line in tail.splitlines() if line.strip()] + return json.loads(lines[-1]) if lines else None + def _append_ledger(self, entry_type: str, body: dict[str, Any]) -> dict[str, Any]: # Single-flight: the chain is read-modify-append, so two concurrent # writers (threads, or two processes sharing a wallet dir) could both # read the same tail, compute the same prev_hash, and fork the chain. - # An advisory exclusive lock over the whole critical section - taken - # AFTER reopening under the lock so the prev-read reflects any writer - # that just finished - makes appends serialize. fsync so a crash can't - # leave a torn line that verify_ledger would read as tampering. - self.ledger_path.touch(exist_ok=True) - with self.ledger_path.open("r+", encoding="utf-8") as handle: - fcntl.flock(handle.fileno(), fcntl.LOCK_EX) - try: - existing = [json.loads(line) for line in handle.read().splitlines() if line.strip()] - prev_hash = existing[-1]["entry_hash"] if existing else "0" * 64 - entry = { - "index": len(existing), + # The exclusive lock lives on a dedicated lock FILE (not the ledger + # fd) so it survives the rotation rename; fsync so a crash can't + # leave a torn line that verify_ledger would read as tampering. Only + # the LAST line is read per append (O(1)); when a segment reaches + # the rotation threshold it is archived and a `rotation` entry + # carries the chain across files, keeping history verifiable end to + # end. + lock_path = self.dir / "ledger.lock" + with lock_path.open("w") as lock: + fcntl.flock(lock.fileno(), fcntl.LOCK_EX) + self.ledger_path.touch(exist_ok=True) + with self.ledger_path.open("rb") as handle: + last = self._last_ledger_line(handle) + prev_hash = last["entry_hash"] if last else "0" * 64 + next_index = (last["index"] + 1) if last else 0 + rotate_at = int((self.policy().get("ledger") or {}).get("rotate_at", 100_000)) + if last is not None and next_index % rotate_at == 0 and last.get("entry_type") != "rotation": + archive = self.dir / f"ledger-{next_index:08d}-{prev_hash[:12]}.jsonl" + self.ledger_path.rename(archive) + rotation = { + "index": next_index, "timestamp": _now(), - "entry_type": entry_type, - "body": body, + "entry_type": "rotation", + "body": { + "archived_file": archive.name, + "archived_head": prev_hash, + "archived_through_index": next_index - 1, + }, "prev_hash": prev_hash, } - entry["entry_hash"] = _sha256(_canonical(entry)) - handle.seek(0, os.SEEK_END) - handle.write(json.dumps(entry, sort_keys=True) + "\n") - handle.flush() - os.fsync(handle.fileno()) - return entry - finally: - fcntl.flock(handle.fileno(), fcntl.LOCK_UN) + rotation["entry_hash"] = _sha256(_canonical(rotation)) + with self.ledger_path.open("w", encoding="utf-8") as fresh: + fresh.write(json.dumps(rotation, sort_keys=True) + "\n") + fresh.flush() + os.fsync(fresh.fileno()) + prev_hash = rotation["entry_hash"] + next_index += 1 + with self.ledger_path.open("ab") as handle: + return self._write_entry(handle, entry_type, body, prev_hash, next_index) + + @staticmethod + def _write_entry(handle: Any, entry_type: str, body: dict[str, Any], prev_hash: str, index: int) -> dict[str, Any]: + entry = { + "index": index, + "timestamp": _now(), + "entry_type": entry_type, + "body": body, + "prev_hash": prev_hash, + } + entry["entry_hash"] = _sha256(_canonical(entry)) + handle.seek(0, os.SEEK_END) + handle.write((json.dumps(entry, sort_keys=True) + "\n").encode("utf-8")) + handle.flush() + os.fsync(handle.fileno()) + return entry def verify_ledger(self) -> tuple[bool, list[str]]: + """Re-check the full chain, walking archived segments through their + `rotation` links, oldest first back to genesis.""" problems: list[str] = [] + segments: list[list[dict[str, Any]]] = [] + entries = self._ledger_entries() + while True: + segments.insert(0, entries) + first = entries[0] if entries else None + if not first or first.get("entry_type") != "rotation": + break + archive = self.dir / str((first.get("body") or {}).get("archived_file", "")) + if not archive.is_file(): + problems.append(f"rotation at index {first.get('index')}: archive {archive.name} missing") + break + entries = [ + json.loads(line) + for line in archive.read_text(encoding="utf-8").splitlines() + if line.strip() + ] prev = "0" * 64 - for entry in self._ledger_entries(): + for entry in (e for segment in segments for e in segment): claimed = entry.get("entry_hash") body = {k: v for k, v in entry.items() if k != "entry_hash"} if entry.get("prev_hash") != prev: @@ -495,8 +562,11 @@ class Wallet: return (not problems), problems def ledger_head(self) -> str: - entries = self._ledger_entries() - return entries[-1]["entry_hash"] if entries else "0" * 64 + if not self.ledger_path.exists(): + return "0" * 64 + with self.ledger_path.open("rb") as handle: + last = self._last_ledger_line(handle) + return last["entry_hash"] if last else "0" * 64 # -- quorum assembly ------------------------------------------------------- @@ -509,6 +579,7 @@ class Wallet: # is the right granularity: an attacker who can rewrite the on-disk # binary already outranks this check, and re-hashing per call buys # nothing against them while taxing every honest verification. + state_dir = state_dir if state_dir is not None else self.state_dir cache_key = str(state_dir) if state_dir is not None else "__default__" cached = self._quorum_cache.get(cache_key) if cached is not None: @@ -639,35 +710,166 @@ class Wallet: ) -> dict[str, Any] | Refusal: """The outbound path. Returns a release dict on success, or a Refusal. - Order matters and is deliberate: latch, policy freshness, intent - binding, signer, then the firewall - the quorum verifies the fresh - signature and only unanimity releases it. A signature that fails the - firewall is quarantined and never returned to the caller; that is - the verify-after-sign fault-injection countermeasure, custody-grade. + Gate order is deliberate and each gate is its own method: latch, + evidence freshness, intent binding, spending policy, signer + resolution, signing, then the firewall - the quorum verifies the + fresh signature and only unanimity releases it. A signature that + fails the firewall is quarantined and never returned to the caller; + that is the verify-after-sign fault-injection countermeasure, + custody-grade. """ request = {"intent": intent, "payload_sha256": _sha256(payload), "key_name": key_name} + refusal = self._gate_latch(request) + if refusal is not None: + return refusal + refusal = self._check_freshness(self.capsule()) + if refusal is not None: + return refusal + refusal = self._gate_intent(intent, payload, request) + if refusal is not None: + return refusal + refusal = self._gate_policy(intent, key_name, request) + if refusal is not None: + return refusal + resolved = self._resolve_signer(signer, key_name, request) + if isinstance(resolved, Refusal): + return resolved + signer, key, pub = resolved + signature = self._obtain_signature(signer, payload, key, intent, request_id, request) + if isinstance(signature, Refusal): + return signature + return self._run_firewall( + payload, signature, pub, intent, request, state_dir, + key_name=key_name, signer_name=getattr(signer, "name", "unknown"), + ) + + # -- outbound gates, one method each --------------------------------------- + + def _gate_latch(self, request: dict[str, Any]) -> Refusal | None: latch = self.latch_state() - if latch.get("latched"): - return self._refuse( - "CUSTODY_LATCHED", - f"custody latch engaged: {latch.get('reason')}", - [f"operator unlatch with written note (incident {latch.get('incident')})"], - "resolve the incident, then `pacta wallet unlatch --note `", - request, - ) - capsule = self.capsule() - stale = self._check_freshness(capsule) - if stale is not None: - return stale + if not latch.get("latched"): + return None + return self._refuse( + "CUSTODY_LATCHED", + f"custody latch engaged: {latch.get('reason')}", + [f"operator unlatch with written note (incident {latch.get('incident')})"], + "resolve the incident, then `pacta wallet unlatch --note `", + request, + ) + + def _gate_intent(self, intent: dict[str, Any], payload: bytes, request: dict[str, Any]) -> Refusal | None: problem = self._validate_intent(intent, payload) - if problem: - return self._refuse( - "MALFORMED_INTENT", - problem, - ["intent.purpose (non-empty string)", "intent.payload_sha256 matching the payload"], - "resend with a well-formed intent envelope; see WALLET.md#intent", - request, + if problem is None: + return None + return self._refuse( + "MALFORMED_INTENT", + problem, + ["intent.purpose (non-empty string)", "intent.payload_sha256 matching the payload"], + "resend with a well-formed intent envelope; see WALLET.md#intent", + request, + ) + + def _gate_policy(self, intent: dict[str, Any], key_name: str, request: dict[str, Any]) -> Refusal | None: + """The lightweight spending-policy engine (POLICY_DENIED consequences). + + Rules live in ``policy.json`` in the wallet directory - the rules you + would give a teenager with a debit card: per-request and per-day + amount ceilings and counterparty allow/deny lists, with per-identity + overrides. No policy file means no restrictions (and the posture + reports as much). Amount rules bind on ``intent.amount``; list rules + bind on ``intent.counterparty`` - policy makes those intent fields + mandatory, so a request that omits them is refused, not waved past. + """ + rules = self._policy_rules(key_name) + if not rules: + return None + + def deny(reason: str, missing: list[str], remediation: str) -> Refusal: + return self._refuse("POLICY_DENIED", reason, missing, remediation, request) + + allow = rules.get("counterparty_allowlist") + denylist = rules.get("counterparty_denylist") or [] + counterparty = intent.get("counterparty") + if (allow is not None or denylist) and not isinstance(counterparty, str): + return deny( + "policy has counterparty rules but the intent names no counterparty", + ["intent.counterparty"], + "resend with intent.counterparty set", ) + if denylist and counterparty in denylist: + return deny( + f"counterparty {counterparty!r} is on the denylist", + [], + "this counterparty is blocked by wallet policy; change the policy deliberately if wrong", + ) + if allow is not None and counterparty not in allow: + return deny( + f"counterparty {counterparty!r} is not on the allowlist", + [], + "add the counterparty to policy.json outbound.counterparty_allowlist if intended", + ) + max_request = rules.get("max_amount_per_request") + max_day = rules.get("max_amount_per_day") + if max_request is not None or max_day is not None: + amount = intent.get("amount") + if not isinstance(amount, (int, float)) or amount <= 0: + return deny( + "policy has amount ceilings but the intent carries no positive intent.amount", + ["intent.amount (positive number)"], + "resend with intent.amount set; amounts are policy units, recorded in the ledger", + ) + if max_request is not None and amount > float(max_request): + return deny( + f"amount {amount} exceeds the per-request ceiling {max_request}", + [], + "split the request or raise the ceiling deliberately in policy.json", + ) + if max_day is not None: + spent = self._spent_last_24h(key_name) + if spent + amount > float(max_day): + return deny( + f"amount {amount} would exceed the daily ceiling {max_day} " + f"(already released in the last 24h: {spent})", + [], + "wait for the window to roll, or raise the ceiling deliberately in policy.json", + ) + return None + + def policy(self) -> dict[str, Any]: + path = self.dir / "policy.json" + if not path.exists(): + return {} + return json.loads(path.read_text(encoding="utf-8")) + + def _policy_rules(self, key_name: str) -> dict[str, Any]: + policy = self.policy() + rules = dict(policy.get("outbound") or {}) + rules.update((policy.get("identities") or {}).get(key_name) or {}) + return rules + + def _spent_last_24h(self, key_name: str) -> float: + cutoff = datetime.now(timezone.utc).timestamp() - 86400 + spent = 0.0 + for entry in self._ledger_entries(): + if entry.get("entry_type") != "outbound-sign": + continue + body = entry.get("body") or {} + if body.get("identity") != key_name: + continue + try: + stamp = datetime.fromisoformat(str(entry.get("timestamp", "")).replace("Z", "+00:00")).timestamp() + except ValueError: + continue + if stamp < cutoff: + continue + amount = (body.get("intent") or {}).get("amount") + if isinstance(amount, (int, float)): + spent += float(amount) + return spent + + def _resolve_signer( + self, signer: Any, key_name: str, request: dict[str, Any] + ) -> tuple[Any, Path, Path] | Refusal: key = self.keys_dir / f"{key_name}.key.pem" pub = self.keys_dir / f"{key_name}.pub.pem" if signer is None: @@ -688,11 +890,21 @@ class Wallet: "use the airgap signer for this identity", request, ) + return signer, key, pub + + def _obtain_signature( + self, + signer: Any, + payload: bytes, + key: Path, + intent: dict[str, Any], + request_id: str | None, + request: dict[str, Any], + ) -> bytes | Refusal: try: if isinstance(signer, AirgapSigner): - signature = signer.sign(payload, key, request_id=request_id, intent=intent) - else: - signature = signer.sign(payload, key) + return signer.sign(payload, key, request_id=request_id, intent=intent) + return signer.sign(payload, key) except PendingAirgap as pending: return self._refuse( "PENDING_AIRGAP", @@ -706,15 +918,24 @@ class Wallet: return self._refuse( "SIGNER_UNAVAILABLE", f"signer failed: {exc}", [], "check the signer backend", request ) - # THE FIREWALL: the fresh signature faces the full quorum. + + def _run_firewall( + self, + payload: bytes, + signature: bytes, + pub: Path, + intent: dict[str, Any], + request: dict[str, Any], + state_dir: str | Path | None, + key_name: str, + signer_name: str, + ) -> dict[str, Any] | Refusal: + """The fresh signature faces the full quorum; only unanimity releases.""" public_key = pem_public_key_to_raw(pub) result = self.quorum(state_dir).verify(payload, signature, public_key) if not result.accepted: quarantine_ref = self._quarantine(payload, signature, public_key, intent, result) - if result.incident: - incident_ref = self._write_incident(result.incident) - else: - incident_ref = None + incident_ref = self._write_incident(result.incident) if result.incident else None self._latch( "outbound firewall rejected a signature this wallet just produced " f"(quarantine {quarantine_ref})", @@ -735,7 +956,7 @@ class Wallet: "signature_hex": signature.hex(), "public_key_hex": public_key.hex(), "identity": key_name, - "signer": getattr(signer, "name", "unknown"), + "signer": signer_name, "intent": intent, "payload_sha256": _sha256(payload), "firewall": result.to_dict(), @@ -830,6 +1051,7 @@ class Wallet: for m in capsule["members"] ], "policy": capsule["policy"], + "spending_policy": self.policy() or {"note": "no policy.json - outbound is unrestricted"}, "latch": self.latch_state(), "ledger": { "entries": len(self._ledger_entries()), diff --git a/src/pacta/walletmcp.py b/src/pacta/walletmcp.py index 9c4f93d..e0708be 100644 --- a/src/pacta/walletmcp.py +++ b/src/pacta/walletmcp.py @@ -38,9 +38,76 @@ def _b64_to_bytes(field: str, value: Any) -> bytes: class _ToolError(Exception): - def __init__(self, code: str, reason: str, missing: list[str], remediation: str) -> None: + def __init__( + self, + code: str, + reason: str, + missing: list[str], + remediation: str, + receipt: dict[str, Any] | None = None, + receipt_path: str | None = None, + ) -> None: super().__init__(reason) self.payload = {"code": code, "reason": reason, "missing": missing, "remediation": remediation} + if receipt is not None: + # The signed refusal receipt travels WITH the error: the refused + # agent can hand its principal a provable, stamped "the wallet + # said no, and this is why" instead of an unsigned anecdote. + self.payload["receipt"] = receipt + self.payload["receipt_path"] = receipt_path + + +def _refusal_error(refusal: Any) -> _ToolError: + return _ToolError( + refusal.code, + refusal.reason, + refusal.missing, + refusal.remediation, + receipt=refusal.receipt, + receipt_path=str(refusal.receipt_path) if refusal.receipt_path else None, + ) + + +class _RateLimiter: + """Sliding-window throttle per tool class. Liveness reads are cheap and + generous; custody mutations are scarce on purpose. Limits exist so a + hostile counterparty cannot grind the ledger or the signer; they are a + surface control, not a custody event - refusals here are NOT ledgered.""" + + WINDOW = 60.0 + LIMITS = {"custody": 30, "verify": 120, "liveness": 240} + + def __init__(self) -> None: + self._hits: dict[str, list[float]] = {} + + def check(self, category: str) -> None: + import time + + now = time.monotonic() + hits = self._hits.setdefault(category, []) + cutoff = now - self.WINDOW + while hits and hits[0] < cutoff: + hits.pop(0) + if len(hits) >= self.LIMITS[category]: + raise _ToolError( + "RATE_LIMITED", + f"{category} budget exhausted ({self.LIMITS[category]} calls / {int(self.WINDOW)}s)", + [], + "back off and retry after the window rolls; liveness reads have a higher budget than custody calls", + ) + hits.append(now) + + +_TOOL_CATEGORY = { + "request_signature": "custody", + "verify_inbound": "verify", + "posture_challenge": "custody", # it produces a firewalled signature + "wallet_status": "liveness", + "custody_card": "liveness", + "list_incidents": "liveness", + "explain_refusal": "liveness", + "airgap_pending": "liveness", +} TOOLS: list[dict[str, Any]] = [ @@ -48,6 +115,7 @@ 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}, + "annotations": {"readOnlyHint": True}, }, { "name": "verify_inbound", @@ -63,25 +131,38 @@ TOOLS: list[dict[str, Any]] = [ "required": ["payload_b64", "signature_b64", "public_key_b64"], "additionalProperties": False, }, + "annotations": {"readOnlyHint": False, "destructiveHint": False, "idempotentHint": True}, }, { "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.", + "description": "Outbound signing with intent binding, spending policy, and the quorum firewall. Provide intent.purpose (recorded as WHY) and the payload; amount/counterparty feed the policy engine when rules exist; signer 'airgap' parks the request for the gap device (resend with the returned request_id after the device answers). The produced signature is quorum-verified 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)"}, + "amount": {"type": "number", "description": "policy units for spending rules (required when the policy has amount ceilings)"}, + "counterparty": {"type": "string", "description": "who this benefits (required when the policy has counterparty lists)"}, + "signer": {"type": "string", "enum": ["local", "airgap"], "description": "signing backend (default local)"}, + "request_id": {"type": "string", "description": "airgap request id when completing a parked request"}, }, "required": ["payload_b64", "purpose"], "additionalProperties": False, }, + "annotations": {"readOnlyHint": False, "destructiveHint": False, "idempotentHint": False}, + }, + { + "name": "airgap_pending", + "description": "List parked airgap signing requests (outbox) and whether the device has answered (inbox). Use the request_id with request_signature to complete one.", + "inputSchema": {"type": "object", "properties": {}, "additionalProperties": False}, + "annotations": {"readOnlyHint": True}, }, { "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}, + "annotations": {"readOnlyHint": True}, }, { "name": "posture_challenge", @@ -92,11 +173,13 @@ TOOLS: list[dict[str, Any]] = [ "required": ["nonce"], "additionalProperties": False, }, + "annotations": {"readOnlyHint": False, "destructiveHint": 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}, + "annotations": {"readOnlyHint": True}, }, { "name": "explain_refusal", @@ -106,18 +189,26 @@ TOOLS: list[dict[str, Any]] = [ "properties": {"index": {"type": "integer", "description": "receipt number; omit for latest"}}, "additionalProperties": False, }, + "annotations": {"readOnlyHint": True}, }, ] class WalletMCP: - def __init__(self, wallet_dir: str | Path, log_url: str = "https://ltl.zkdefi.org") -> None: - self.wallet = Wallet(wallet_dir) + def __init__( + self, + wallet_dir: str | Path, + log_url: str = "https://ltl.zkdefi.org", + state_dir: str | Path | None = None, + ) -> None: + self.wallet = Wallet(wallet_dir, state_dir=state_dir) self.log_url = log_url + self.limiter = _RateLimiter() 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, + "airgap_pending": self._airgap_pending, "custody_card": self._custody_card, "posture_challenge": self._posture_challenge, "list_incidents": self._list_incidents, @@ -147,16 +238,43 @@ class WalletMCP: 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()} + intent: dict[str, Any] = {"purpose": purpose, "payload_sha256": hashlib.sha256(payload).hexdigest()} + if args.get("amount") is not None: + intent["amount"] = args["amount"] + if args.get("counterparty") is not None: + intent["counterparty"] = args["counterparty"] + signer = None + if args.get("signer") == "airgap": + from .wallet import AirgapSigner + + signer = AirgapSigner(self.wallet.airgap_dir) result = self.wallet.request_signature( - intent, payload, key_name=str(args.get("identity", "warden")) + intent, + payload, + signer=signer, + key_name=str(args.get("identity", "warden")), + request_id=args.get("request_id"), ) if isinstance(result, Refusal): - raise _ToolError( - result.code, result.reason, result.missing, result.remediation - ) + raise _refusal_error(result) return result + def _airgap_pending(self, _: dict[str, Any]) -> dict[str, Any]: + pending = [] + outbox = self.wallet.airgap_dir / "outbox" + inbox = self.wallet.airgap_dir / "inbox" + for req in sorted(outbox.glob("*.request.json")): + request_id = req.name.removesuffix(".request.json") + body = json.loads(req.read_text(encoding="utf-8")) + pending.append({ + "request_id": request_id, + "created_at": body.get("created_at"), + "payload_sha256": body.get("payload_sha256"), + "intent": body.get("intent"), + "device_answered": (inbox / f"{request_id}.response.json").exists(), + }) + return {"pending": pending, "count": len(pending)} + def _custody_card(self, _: dict[str, Any]) -> dict[str, Any]: return build_custody_card(self.wallet, self.log_url) @@ -166,7 +284,7 @@ class WalletMCP: 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) + raise _refusal_error(result) return result def _list_incidents(self, _: dict[str, Any]) -> dict[str, Any]: @@ -212,6 +330,7 @@ class WalletMCP: if handler is None: return self._err(msg_id, -32602, f"unknown tool {name}") try: + self.limiter.check(_TOOL_CATEGORY.get(name, "custody")) payload = handler(args) return self._ok(msg_id, { "content": [{"type": "text", "text": json.dumps(payload, indent=2, sort_keys=True)}], diff --git a/tests/test_wallet_hardening.py b/tests/test_wallet_hardening.py new file mode 100644 index 0000000..fab8227 --- /dev/null +++ b/tests/test_wallet_hardening.py @@ -0,0 +1,347 @@ +"""Hardening round: policy engine, ledger rotation, MCP rate limits + +receipt-bearing errors, airgap MCP flow, treasury parsing/verification. +""" + +import base64 +import hashlib +import json +import stat +from pathlib import Path + +import pytest + +from pacta.quorum import binary_path +from pacta.signing import generate_ed25519_keypair +from pacta.wallet import Refusal, Wallet +from pacta.walletmcp import WalletMCP, _RateLimiter + + +def _sha256(data: bytes) -> str: + return hashlib.sha256(data).hexdigest() + + +def _fake_member(path: Path, verdict: str) -> None: + code = {"accept": 0, "reject": 1}[verdict] + out = {"accept": "OK", "reject": "INVALID"}[verdict] + path.write_text(f"#!/bin/sh\necho {out}\nexit {code}\n") + path.chmod(path.stat().st_mode | stat.S_IEXEC) + + +def _seal_wallet(tmp_path: Path, verdicts: dict, state_dir: Path) -> Wallet: + state_dir.mkdir(parents=True, exist_ok=True) + members = [] + for name, verdict in verdicts.items(): + binary = binary_path(name, state_dir) + _fake_member(binary, verdict) + members.append({ + "backend": name, "component": f"{name}-ed25519-verified", "semantics": "test", + "entry_point": "test", "source_commit": "de" * 20, "repo_commit": "ca" * 20, + "binary_sha256": _sha256(binary.read_bytes()), "backend_cfg": "test", "risk_tier": "R4", + "evidence": {"leaf_hash": "00", "leaf_index": 0, "tree_size": 1, "inclusion_proof": [], + "sth": {"timestamp": "2099-01-01T00:00:00Z"}}, + }) + wallet = Wallet(tmp_path / "w") + for sub in (wallet.keys_dir, wallet.incidents_dir, wallet.receipts_dir, + wallet.quarantine_dir, wallet.airgap_dir / "outbox", wallet.airgap_dir / "inbox"): + sub.mkdir(parents=True, exist_ok=True) + capsule = {"type": "pacta.wallet.custody_capsule.v1", "created_at": "2026-07-06T00:00:00Z", + "members": members, + "policy": {"require_unanimity": True, "min_members": 2, "require_tier": "R4", + "freshness_max_age_days": 0}, + "signing": {"backend": "test"}, "problems_at_init": []} + wallet.capsule_path.write_text(json.dumps(capsule, indent=2, sort_keys=True) + "\n") + wallet._append_ledger("genesis", {"capsule_sha256": "x"}) + generate_ed25519_keypair(wallet.keys_dir / "warden.key.pem", wallet.keys_dir / "warden.pub.pem") + wallet._test_state_dir = state_dir # type: ignore[attr-defined] + return wallet + + +def _intent(payload: bytes, **extra) -> dict: + intent = {"purpose": "test", "payload_sha256": _sha256(payload)} + intent.update(extra) + return intent + + +def _needs_signer(): + from pacta.dogfood import locate_verifier + + if locate_verifier() is None: + pytest.skip("dogfood signer not built") + + +# -- policy engine ----------------------------------------------------------- + + +def test_policy_absent_means_unrestricted(tmp_path): + _needs_signer() + wallet = _seal_wallet(tmp_path, {"a": "accept", "b": "accept"}, tmp_path / "state") + payload = b"no policy" + result = wallet.request_signature(_intent(payload), payload, state_dir=wallet._test_state_dir) + assert not isinstance(result, Refusal) + + +def test_policy_amount_ceiling_denies(tmp_path): + wallet = _seal_wallet(tmp_path, {"a": "accept", "b": "accept"}, tmp_path / "state") + (wallet.dir / "policy.json").write_text(json.dumps( + {"outbound": {"max_amount_per_request": 50}})) + payload = b"big spender" + # missing amount -> denied (policy makes the field mandatory) + r1 = wallet.request_signature(_intent(payload), payload, state_dir=wallet._test_state_dir) + assert isinstance(r1, Refusal) and r1.code == "POLICY_DENIED" + # over ceiling -> denied + r2 = wallet.request_signature(_intent(payload, amount=51), payload, state_dir=wallet._test_state_dir) + assert isinstance(r2, Refusal) and r2.code == "POLICY_DENIED" + + +def test_policy_under_ceiling_releases_and_daily_cap_accumulates(tmp_path): + _needs_signer() + wallet = _seal_wallet(tmp_path, {"a": "accept", "b": "accept"}, tmp_path / "state") + (wallet.dir / "policy.json").write_text(json.dumps( + {"outbound": {"max_amount_per_request": 50, "max_amount_per_day": 80}})) + payload = b"pay 40" + ok1 = wallet.request_signature(_intent(payload, amount=40), payload, state_dir=wallet._test_state_dir) + assert not isinstance(ok1, Refusal) + ok2 = wallet.request_signature(_intent(payload, amount=40), payload, state_dir=wallet._test_state_dir) + assert not isinstance(ok2, Refusal) + # 80 spent; one more coin breaks the day + r = wallet.request_signature(_intent(payload, amount=1), payload, state_dir=wallet._test_state_dir) + assert isinstance(r, Refusal) and r.code == "POLICY_DENIED" + assert "daily ceiling" in r.reason + + +def test_policy_counterparty_lists(tmp_path): + _needs_signer() + wallet = _seal_wallet(tmp_path, {"a": "accept", "b": "accept"}, tmp_path / "state") + (wallet.dir / "policy.json").write_text(json.dumps( + {"outbound": {"counterparty_allowlist": ["alice"], "counterparty_denylist": ["mallory"]}})) + payload = b"to someone" + r1 = wallet.request_signature(_intent(payload), payload, state_dir=wallet._test_state_dir) + assert isinstance(r1, Refusal) and "names no counterparty" in r1.reason + r2 = wallet.request_signature(_intent(payload, counterparty="mallory"), payload, state_dir=wallet._test_state_dir) + assert isinstance(r2, Refusal) and "denylist" in r2.reason + r3 = wallet.request_signature(_intent(payload, counterparty="bob"), payload, state_dir=wallet._test_state_dir) + assert isinstance(r3, Refusal) and "allowlist" in r3.reason + ok = wallet.request_signature(_intent(payload, counterparty="alice"), payload, state_dir=wallet._test_state_dir) + assert not isinstance(ok, Refusal) + + +def test_policy_identity_override(tmp_path): + wallet = _seal_wallet(tmp_path, {"a": "accept", "b": "accept"}, tmp_path / "state") + (wallet.dir / "policy.json").write_text(json.dumps({ + "outbound": {"max_amount_per_request": 100}, + "identities": {"warden": {"max_amount_per_request": 10}}, + })) + payload = b"identity override" + r = wallet.request_signature(_intent(payload, amount=50), payload, state_dir=wallet._test_state_dir) + assert isinstance(r, Refusal) and r.code == "POLICY_DENIED" + + +# -- ledger rotation + O(1) head ------------------------------------------------ + + +def test_ledger_rotation_keeps_chain_verifiable(tmp_path): + wallet = _seal_wallet(tmp_path, {"a": "accept", "b": "accept"}, tmp_path / "state") + (wallet.dir / "policy.json").write_text(json.dumps({"ledger": {"rotate_at": 5}})) + for i in range(12): + wallet._append_ledger("stress", {"i": i}) + archives = sorted(wallet.dir.glob("ledger-*.jsonl")) + assert archives, "rotation never happened" + ok, problems = wallet.verify_ledger() + assert ok, problems + # head is O(1)-readable and matches the last line + entries = wallet._ledger_entries() + assert wallet.ledger_head() == entries[-1]["entry_hash"] + # indices keep increasing across the rotation boundary + assert entries[-1]["index"] == 12 + len(archives) # genesis + 12 + rotation entries + + +def test_ledger_rotation_detects_archive_tampering(tmp_path): + wallet = _seal_wallet(tmp_path, {"a": "accept", "b": "accept"}, tmp_path / "state") + (wallet.dir / "policy.json").write_text(json.dumps({"ledger": {"rotate_at": 4}})) + for i in range(9): + wallet._append_ledger("stress", {"i": i}) + archive = sorted(wallet.dir.glob("ledger-*.jsonl"))[0] + lines = archive.read_text().splitlines() + doctored = json.loads(lines[1]) + doctored["body"]["i"] = 999 + lines[1] = json.dumps(doctored, sort_keys=True) + archive.write_text("\n".join(lines) + "\n") + ok, problems = wallet.verify_ledger() + assert not ok and problems + + +# -- MCP: receipts in errors, airgap flow, rate limiting ------------------------- + + +def _call(srv, method, params=None, msg_id=1): + return srv.handle({"jsonrpc": "2.0", "id": msg_id, "method": method, "params": params or {}}) + + +def test_mcp_refusal_carries_signed_receipt(tmp_path): + _needs_signer() + wallet = _seal_wallet(tmp_path, {"a": "accept", "b": "accept"}, tmp_path / "state") + (wallet.dir / "policy.json").write_text(json.dumps( + {"outbound": {"max_amount_per_request": 1}})) + srv = WalletMCP(wallet.dir, state_dir=wallet._test_state_dir) + res = _call(srv, "tools/call", {"name": "request_signature", "arguments": { + "payload_b64": base64.b64encode(b"x").decode(), "purpose": "test", "amount": 9, + }})["result"] + assert res["isError"] + body = res["structuredContent"] + assert body["code"] == "POLICY_DENIED" + # the signed receipt travels with the error + assert body["receipt"]["type"] == "pacta.wallet.refusal_receipt.v1" + assert body["receipt"]["signature"].get("scheme") == "ed25519-dogfood" + assert body["receipt_path"] + + +def test_mcp_airgap_park_list_complete(tmp_path): + _needs_signer() + wallet = _seal_wallet(tmp_path, {"a": "accept", "b": "accept"}, tmp_path / "state") + srv = WalletMCP(wallet.dir, state_dir=wallet._test_state_dir) + payload = b"gap payload" + res = _call(srv, "tools/call", {"name": "request_signature", "arguments": { + "payload_b64": base64.b64encode(payload).decode(), "purpose": "gap test", + "signer": "airgap", "request_id": "reqmcp1", + }})["result"] + assert res["structuredContent"]["code"] == "PENDING_AIRGAP" + listing = _call(srv, "tools/call", {"name": "airgap_pending"})["result"]["structuredContent"] + assert listing["count"] == 1 and listing["pending"][0]["request_id"] == "reqmcp1" + assert listing["pending"][0]["device_answered"] is False + # the "device" signs (with the wallet's own key, for the test) and answers + from pacta.dogfood import locate_verifier, sign_payload_dogfood + + signature = sign_payload_dogfood(payload, wallet.keys_dir / "warden.key.pem", locate_verifier()) + (wallet.airgap_dir / "inbox" / "reqmcp1.response.json").write_text( + json.dumps({"signature_hex": signature.hex()})) + done = _call(srv, "tools/call", {"name": "request_signature", "arguments": { + "payload_b64": base64.b64encode(payload).decode(), "purpose": "gap test", + "signer": "airgap", "request_id": "reqmcp1", + }})["result"] + assert not done["isError"], done["structuredContent"] + assert done["structuredContent"]["firewall"]["classification"] == "unanimous-accept" + assert done["structuredContent"]["signer"] == "airgap" + + +def test_rate_limiter_trips_and_recovers(): + limiter = _RateLimiter() + limiter.LIMITS = {"custody": 3, "verify": 120, "liveness": 240} + for _ in range(3): + limiter.check("custody") + from pacta.walletmcp import _ToolError + + with pytest.raises(_ToolError) as excinfo: + limiter.check("custody") + assert excinfo.value.payload["code"] == "RATE_LIMITED" + + +def test_mcp_tools_have_annotations(tmp_path): + from pacta.walletmcp import TOOLS + + for tool in TOOLS: + assert "annotations" in tool, tool["name"] + readonly = {t["name"] for t in TOOLS if t["annotations"].get("readOnlyHint")} + assert "wallet_status" in readonly and "custody_card" in readonly + assert "request_signature" not in readonly + + +# -- treasury ------------------------------------------------------------------- + + +def _make_message(signer_key: bytes, extra_key: bytes, version0: bool = False) -> bytes: + header = bytes([1, 0, 1]) + keys = bytes([2]) + signer_key + extra_key # compact-u16(2) then two keys + blockhash = bytes(range(32)) + instructions = bytes([0]) # compact-u16(0): none + prefix = bytes([0x80]) if version0 else b"" + return prefix + header + keys + blockhash + instructions + + +def _assemble_tx(signature: bytes, message: bytes) -> bytes: + return bytes([1]) + signature + message + + +def test_treasury_parse_roundtrip_legacy_and_v0(): + from pacta.treasury import parse_transaction + + signer, extra = bytes([7] * 32), bytes([9] * 32) + for v0 in (False, True): + message = _make_message(signer, extra, version0=v0) + tx = _assemble_tx(b"\xab" * 64, message) + parsed = parse_transaction(tx) + assert parsed.num_required_signatures == 1 + assert parsed.account_keys[0] == signer + assert parsed.message == message + assert parsed.version == (0 if v0 else None) + + +def test_treasury_parse_rejects_garbage(): + from pacta.treasury import parse_transaction + + with pytest.raises(ValueError): + parse_transaction(b"") + with pytest.raises(ValueError): + parse_transaction(bytes([1]) + b"\x00" * 10) # truncated signature + # header/signature-count mismatch + signer, extra = bytes([7] * 32), bytes([9] * 32) + message = bytes([2, 0, 1]) + bytes([2]) + signer + extra + bytes(32) + bytes([0]) + with pytest.raises(ValueError): + parse_transaction(_assemble_tx(b"\xab" * 64, message)) + + +def test_treasury_b58decode_vectors(): + from pacta.treasury import b58decode + + assert b58decode("") == b"" + assert b58decode("1") == b"\x00" + assert b58decode("2g") == b"a" + assert b58decode("ZiCa") == b"abc" + assert b58decode("11ZiCa") == b"\x00\x00abc" + with pytest.raises(ValueError): + b58decode("0OIl") + + +def test_treasury_verify_transaction_real_quorum(tmp_path): + """The showcase: a synthetic Solana transaction signed with the wallet's + key, quorum-verified through the REAL four proven forks.""" + if not all(binary_path(b).exists() for b in ("dalek", "anza")): + pytest.skip("real quorum not built") + _needs_signer() + from pacta.dogfood import locate_verifier, pem_public_key_to_raw, sign_payload_dogfood + from pacta.treasury import verify_transaction + + # seal against the REAL binaries + real = {b: binary_path(b) for b in ("dalek", "anza", "risc0", "betrusted") if binary_path(b).exists()} + wallet = Wallet(tmp_path / "w") + for sub in (wallet.keys_dir, wallet.incidents_dir, wallet.receipts_dir, + wallet.quarantine_dir, wallet.airgap_dir / "outbox", wallet.airgap_dir / "inbox"): + sub.mkdir(parents=True, exist_ok=True) + members = [{ + "backend": b, "component": f"{b}-ed25519-verified", "semantics": "t", "entry_point": "t", + "source_commit": "x", "repo_commit": "y", "binary_sha256": _sha256(p.read_bytes()), + "backend_cfg": "t", "risk_tier": "R4", + "evidence": {"leaf_hash": "00", "leaf_index": 0, "tree_size": 1, "inclusion_proof": [], + "sth": {"timestamp": "2099-01-01T00:00:00Z"}}, + } for b, p in real.items()] + capsule = {"type": "pacta.wallet.custody_capsule.v1", "created_at": "2026-07-06T00:00:00Z", + "members": members, + "policy": {"require_unanimity": True, "min_members": 2, "require_tier": "R4", + "freshness_max_age_days": 0}, + "signing": {"backend": "test"}, "problems_at_init": []} + wallet.capsule_path.write_text(json.dumps(capsule, indent=2, sort_keys=True) + "\n") + wallet._append_ledger("genesis", {"capsule_sha256": "x"}) + generate_ed25519_keypair(wallet.keys_dir / "warden.key.pem", wallet.keys_dir / "warden.pub.pem") + + signer_key = pem_public_key_to_raw(wallet.keys_dir / "warden.pub.pem") + message = _make_message(signer_key, bytes([9] * 32)) + signature = sign_payload_dogfood(message, wallet.keys_dir / "warden.key.pem", locate_verifier()) + verdict = verify_transaction(wallet, _assemble_tx(signature, message)) + assert verdict.authentic + assert verdict.signer_results[0]["classification"] == "unanimous-accept" + # a flipped byte in the signature: not authentic + bad = verify_transaction(wallet, _assemble_tx(bytes([signature[0] ^ 1]) + signature[1:], message)) + assert not bad.authentic + # and the checks are in the ledger with treasury context + contexts = [e["body"].get("context", "") for e in wallet._ledger_entries() + if e["entry_type"] == "inbound-verify"] + assert any("treasury" in c for c in contexts)