saymrwulf/proof-aware-crypto-tooling-agent
1
0
Fork 0
mirror of https://github.com/saymrwulf/proof-aware-crypto-tooling-agent.git synced 2026-09-03 19:53:43 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions

warden hardening round: policy engine, ledger rotation, MCP UX, treasury LIVE, ops docs

Browse source 
Tier 2:
- request_signature decomposed into named gates (latch, freshness,
  intent, policy, signer, firewall)
- ledger: O(1) tail-read appends under a dedicated lock file (survives
  rotation rename); hash-chained segment rotation at policy
  ledger.rotate_at; verify-ledger walks all segments to genesis;
  archive tampering detected (tested)
- docs/threat-model.md (attacker matrix 1-9, proven-vs-trusted, design
  invariants) + docs/runbook-latch.md (diagnose-first recovery)
- lecture 10: executable corrupt-a-member exercise (capsule pin catches
  one appended byte), honest note on what the pin does NOT stop

Lightweight policy engine (POLICY_DENIED wired):
- policy.json: per-request/per-day amount ceilings, counterparty
  allow/deny lists, per-identity overrides; rules make their intent
  fields mandatory; daily sums from the ledger
-

Agent UX:
- signed refusal receipts travel inside MCP errors (receipt +
  receipt_path in structuredContent)
- airgap over MCP: request_signature signer=airgap + request_id,
  new airgap_pending tool; park -> list -> device answers -> complete
  (tested end-to-end)
- all 8 tools carry readOnly/destructive annotations
- sliding-window rate limiter per tool class (custody/verify/liveness);
  RATE_LIMITED refusal code; surface control, not ledgered

warden-treasury LIVE:
- treasury.py: stdlib base58, compact-u16, legacy+v0 wire parsing;
  every required signature quorum-verified over exact message bytes;
  completeness gap named in every verdict; RPC fetch uses response as
  bytes only
-
- live-quorum test: synthetic Solana tx signed with wallet key ->
  authentic via 4 proven forks; flipped byte -> not authentic

100 tests green (was 85).

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
This commit is contained in:
mrwulf 2026-07-07 13:38:42 +02:00
parent b867ec2342
commit 65772b3d2e
11 changed files with 1322 additions and 86 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

64
WALLET.md
View file

@ -129,9 +129,11 @@ pacta wallet mcp --wallet ./my-warden # stdio JSON-RPC MCP server
## Agent-native surface (MCP) ## Agent-native surface (MCP)
`pacta wallet mcp` speaks MCP over stdio JSON-RPC. Seven outcome-first `pacta wallet mcp` speaks MCP over stdio JSON-RPC. Eight outcome-first
tools; strict input schemas; results carry evidence; errors are structured tools with read-only/destructive annotations; strict input schemas;
objects, never prose. 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 | | tool | does |
|---|---| |---|---|
@ -142,10 +144,12 @@ objects, never prose.
| `posture_challenge` | nonce → firewalled, signed posture attestation with the quorum trail | | `posture_challenge` | nonce → firewalled, signed posture attestation with the quorum trail |
| `list_incidents` | divergences and quarantines, newest-first | | `list_incidents` | divergences and quarantines, newest-first |
| `explain_refusal` | fetch a refusal receipt by index (or latest) | | `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`, Refusal codes (every refusal names one): `EVIDENCE_REQUIRED`,
`POLICY_DENIED`, `CUSTODY_LATCHED`, `EVIDENCE_STALE`, `MALFORMED_INTENT`, `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 ### 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 <dir> --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) ## The signing firewall (verify-after-sign)
Outbound is: **intent → sign → firewall → release**. Outbound is: **intent → sign → firewall → release**.
@ -201,7 +253,9 @@ alarm:
## Product lineup ## Product lineup
warden ships as one core with four production-ready deployment profiles — 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-solo** — a single agent's custody sidecar (local signer).
- **warden-airgap** — signing behind a Precursor/Betrusted hardware gap. - **warden-airgap** — signing behind a Precursor/Betrusted hardware gap.

14
docs/products.md
View file

@ -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 compromised or lying RPC can withhold data but cannot manufacture a
signature the quorum will accept. signature the quorum will accept.
**Ready because:** the anza member is built and tested; wiring it to a **Ready because:** it is built: `pacta wallet treasury-verify` parses
transaction feed is deployment configuration, not new trust surface. (The wire-format transactions (legacy + v0, stdlib only) and quorum-verifies
chain-adapter layer is the documented integration point.) 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" ## Honesty about "production-ready"
Profiles 1 and 2 run end-to-end today on the tested core. Profiles 3 and 4 Profiles 1, 2, and 3 run end-to-end today on the tested core. Profile 4
are complete *product definitions* on the same core with one documented is a complete *product definition* on the same core with one documented
integration point each (a chain-transaction adapter; a gossip transport) — integration point (a gossip transport) —
named here so the boundary between "built and tested" and "wired to your 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 environment" is exact, which is the whole ethos of this project. None of
them changes the trust posture; all of them fail closed. them changes the trust posture; all of them fail closed.

78
docs/runbook-latch.md Normal file
View file

@ -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 <dir> # latch reason + incident ref
cat <dir>/latch.json
cat <dir>/incidents/<ref>.json # the full divergence trail
ls <dir>/quarantine/ # any withheld signatures
pacta wallet verify-ledger --wallet <dir> # 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 <dir> --note "<what happened, what you checked, why it is safe now>"
```
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.

49
docs/threat-model.md Normal file
View file

@ -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.

61
notebooks/10_verified_custody_wallet.ipynb
View file

@ -253,6 +253,59 @@
" print(\"The counterparty believed no adjective; it recomputed a Merkle root.\")\n" " 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", "cell_type": "markdown",
"metadata": {}, "metadata": {},
@ -261,9 +314,11 @@
"\n", "\n",
"- Change `toy_quorum` to majority voting and write two sentences\n", "- Change `toy_quorum` to majority voting and write two sentences\n",
" on exactly which attack that lets through.\n", " on exactly which attack that lets through.\n",
"- Take the real quorum cell and corrupt one member binary on\n", "- Extend the corrupt-a-member cell: corrupt the capsule JSON\n",
" disk (append a byte). Predict, then observe, what the wallet's\n", " itself instead of the binary. What catches that, and when?\n",
" capsule hash-pin does the next time it assembles the quorum.\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", "- The signing path is trusted base. Write the strongest *true*\n",
" sentence you can about warden's outbound safety, and the\n", " sentence you can about warden's outbound safety, and the\n",
" strongest *false* one a marketer would write - and name the\n", " strongest *false* one a marketer would write - and name the\n",

53
scripts/build_curriculum_notebooks.py
View file

@ -2245,15 +2245,62 @@ COURSE = {
print("The counterparty believed no adjective; it recomputed a Merkle root.") 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( md(
""" """
## Exercises ## Exercises
- Change `toy_quorum` to majority voting and write two sentences - Change `toy_quorum` to majority voting and write two sentences
on exactly which attack that lets through. on exactly which attack that lets through.
- Take the real quorum cell and corrupt one member binary on - Extend the corrupt-a-member cell: corrupt the capsule JSON
disk (append a byte). Predict, then observe, what the wallet's itself instead of the binary. What catches that, and when?
capsule hash-pin does the next time it assembles the quorum. (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* - The signing path is trusted base. Write the strongest *true*
sentence you can about warden's outbound safety, and the sentence you can about warden's outbound safety, and the
strongest *false* one a marketer would write - and name the strongest *false* one a marketer would write - and name the

67
src/pacta/cli.py
View file

@ -244,6 +244,19 @@ def build_parser() -> argparse.ArgumentParser:
w_unlatch.add_argument("--note", required=True) w_unlatch.add_argument("--note", required=True)
w_unlatch.set_defaults(func=cmd_wallet_unlatch) 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 return parser
@ -677,6 +690,60 @@ def cmd_wallet_unlatch(args: argparse.Namespace) -> int:
return 0 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: def cmd_agent(args: argparse.Namespace) -> int:
card = _card_for_agent(args) card = _card_for_agent(args)
decision = run_agent_action( decision = run_agent_action(

196
src/pacta/treasury.py Normal file
View file

@ -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)

340
src/pacta/wallet.py
View file

@ -77,6 +77,7 @@ REFUSAL_CODES = (
"SIGNER_UNAVAILABLE", # signer backend cannot serve the request "SIGNER_UNAVAILABLE", # signer backend cannot serve the request
"FIREWALL_QUARANTINE", # produced signature failed the quorum firewall "FIREWALL_QUARANTINE", # produced signature failed the quorum firewall
"PENDING_AIRGAP", # request parked in the airgap outbox "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: 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.dir = Path(wallet_dir)
self.state_dir = state_dir # default quorum binary location override
self.capsule_path = self.dir / "capsule.json" self.capsule_path = self.dir / "capsule.json"
self.ledger_path = self.dir / "ledger.jsonl" self.ledger_path = self.dir / "ledger.jsonl"
self.latch_path = self.dir / "latch.json" self.latch_path = self.dir / "latch.json"
@ -451,40 +453,105 @@ class Wallet:
if line.strip() 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]: 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 # Single-flight: the chain is read-modify-append, so two concurrent
# writers (threads, or two processes sharing a wallet dir) could both # writers (threads, or two processes sharing a wallet dir) could both
# read the same tail, compute the same prev_hash, and fork the chain. # read the same tail, compute the same prev_hash, and fork the chain.
# An advisory exclusive lock over the whole critical section - taken # The exclusive lock lives on a dedicated lock FILE (not the ledger
# AFTER reopening under the lock so the prev-read reflects any writer # fd) so it survives the rotation rename; fsync so a crash can't
# that just finished - makes appends serialize. fsync so a crash can't # leave a torn line that verify_ledger would read as tampering. Only
# leave a torn line that verify_ledger would read as tampering. # the LAST line is read per append (O(1)); when a segment reaches
self.ledger_path.touch(exist_ok=True) # the rotation threshold it is archived and a `rotation` entry
with self.ledger_path.open("r+", encoding="utf-8") as handle: # carries the chain across files, keeping history verifiable end to
fcntl.flock(handle.fileno(), fcntl.LOCK_EX) # end.
try: lock_path = self.dir / "ledger.lock"
existing = [json.loads(line) for line in handle.read().splitlines() if line.strip()] with lock_path.open("w") as lock:
prev_hash = existing[-1]["entry_hash"] if existing else "0" * 64 fcntl.flock(lock.fileno(), fcntl.LOCK_EX)
entry = { self.ledger_path.touch(exist_ok=True)
"index": len(existing), 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(), "timestamp": _now(),
"entry_type": entry_type, "entry_type": "rotation",
"body": body, "body": {
"archived_file": archive.name,
"archived_head": prev_hash,
"archived_through_index": next_index - 1,
},
"prev_hash": prev_hash, "prev_hash": prev_hash,
} }
entry["entry_hash"] = _sha256(_canonical(entry)) rotation["entry_hash"] = _sha256(_canonical(rotation))
handle.seek(0, os.SEEK_END) with self.ledger_path.open("w", encoding="utf-8") as fresh:
handle.write(json.dumps(entry, sort_keys=True) + "\n") fresh.write(json.dumps(rotation, sort_keys=True) + "\n")
handle.flush() fresh.flush()
os.fsync(handle.fileno()) os.fsync(fresh.fileno())
return entry prev_hash = rotation["entry_hash"]
finally: next_index += 1
fcntl.flock(handle.fileno(), fcntl.LOCK_UN) 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]]: 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] = [] 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 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") claimed = entry.get("entry_hash")
body = {k: v for k, v in entry.items() if k != "entry_hash"} body = {k: v for k, v in entry.items() if k != "entry_hash"}
if entry.get("prev_hash") != prev: if entry.get("prev_hash") != prev:
@ -495,8 +562,11 @@ class Wallet:
return (not problems), problems return (not problems), problems
def ledger_head(self) -> str: def ledger_head(self) -> str:
entries = self._ledger_entries() if not self.ledger_path.exists():
return entries[-1]["entry_hash"] if entries else "0" * 64 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 ------------------------------------------------------- # -- quorum assembly -------------------------------------------------------
@ -509,6 +579,7 @@ class Wallet:
# is the right granularity: an attacker who can rewrite the on-disk # is the right granularity: an attacker who can rewrite the on-disk
# binary already outranks this check, and re-hashing per call buys # binary already outranks this check, and re-hashing per call buys
# nothing against them while taxing every honest verification. # 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__" cache_key = str(state_dir) if state_dir is not None else "__default__"
cached = self._quorum_cache.get(cache_key) cached = self._quorum_cache.get(cache_key)
if cached is not None: if cached is not None:
@ -639,35 +710,166 @@ class Wallet:
) -> dict[str, Any] | Refusal: ) -> dict[str, Any] | Refusal:
"""The outbound path. Returns a release dict on success, or a Refusal. """The outbound path. Returns a release dict on success, or a Refusal.
Order matters and is deliberate: latch, policy freshness, intent Gate order is deliberate and each gate is its own method: latch,
binding, signer, then the firewall - the quorum verifies the fresh evidence freshness, intent binding, spending policy, signer
signature and only unanimity releases it. A signature that fails the resolution, signing, then the firewall - the quorum verifies the
firewall is quarantined and never returned to the caller; that is fresh signature and only unanimity releases it. A signature that
the verify-after-sign fault-injection countermeasure, custody-grade. 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} 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() latch = self.latch_state()
if latch.get("latched"): if not latch.get("latched"):
return self._refuse( return None
"CUSTODY_LATCHED", return self._refuse(
f"custody latch engaged: {latch.get('reason')}", "CUSTODY_LATCHED",
[f"operator unlatch with written note (incident {latch.get('incident')})"], f"custody latch engaged: {latch.get('reason')}",
"resolve the incident, then `pacta wallet unlatch --note <why>`", [f"operator unlatch with written note (incident {latch.get('incident')})"],
request, "resolve the incident, then `pacta wallet unlatch --note <why>`",
) request,
capsule = self.capsule() )
stale = self._check_freshness(capsule)
if stale is not None: def _gate_intent(self, intent: dict[str, Any], payload: bytes, request: dict[str, Any]) -> Refusal | None:
return stale
problem = self._validate_intent(intent, payload) problem = self._validate_intent(intent, payload)
if problem: if problem is None:
return self._refuse( return None
"MALFORMED_INTENT", return self._refuse(
problem, "MALFORMED_INTENT",
["intent.purpose (non-empty string)", "intent.payload_sha256 matching the payload"], problem,
"resend with a well-formed intent envelope; see WALLET.md#intent", ["intent.purpose (non-empty string)", "intent.payload_sha256 matching the payload"],
request, "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" key = self.keys_dir / f"{key_name}.key.pem"
pub = self.keys_dir / f"{key_name}.pub.pem" pub = self.keys_dir / f"{key_name}.pub.pem"
if signer is None: if signer is None:
@ -688,11 +890,21 @@ class Wallet:
"use the airgap signer for this identity", "use the airgap signer for this identity",
request, 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: try:
if isinstance(signer, AirgapSigner): if isinstance(signer, AirgapSigner):
signature = signer.sign(payload, key, request_id=request_id, intent=intent) return signer.sign(payload, key, request_id=request_id, intent=intent)
else: return signer.sign(payload, key)
signature = signer.sign(payload, key)
except PendingAirgap as pending: except PendingAirgap as pending:
return self._refuse( return self._refuse(
"PENDING_AIRGAP", "PENDING_AIRGAP",
@ -706,15 +918,24 @@ class Wallet:
return self._refuse( return self._refuse(
"SIGNER_UNAVAILABLE", f"signer failed: {exc}", [], "check the signer backend", request "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) public_key = pem_public_key_to_raw(pub)
result = self.quorum(state_dir).verify(payload, signature, public_key) result = self.quorum(state_dir).verify(payload, signature, public_key)
if not result.accepted: if not result.accepted:
quarantine_ref = self._quarantine(payload, signature, public_key, intent, result) quarantine_ref = self._quarantine(payload, signature, public_key, intent, result)
if result.incident: incident_ref = self._write_incident(result.incident) if result.incident else None
incident_ref = self._write_incident(result.incident)
else:
incident_ref = None
self._latch( self._latch(
"outbound firewall rejected a signature this wallet just produced " "outbound firewall rejected a signature this wallet just produced "
f"(quarantine {quarantine_ref})", f"(quarantine {quarantine_ref})",
@ -735,7 +956,7 @@ class Wallet:
"signature_hex": signature.hex(), "signature_hex": signature.hex(),
"public_key_hex": public_key.hex(), "public_key_hex": public_key.hex(),
"identity": key_name, "identity": key_name,
"signer": getattr(signer, "name", "unknown"), "signer": signer_name,
"intent": intent, "intent": intent,
"payload_sha256": _sha256(payload), "payload_sha256": _sha256(payload),
"firewall": result.to_dict(), "firewall": result.to_dict(),
@ -830,6 +1051,7 @@ class Wallet:
for m in capsule["members"] for m in capsule["members"]
], ],
"policy": capsule["policy"], "policy": capsule["policy"],
"spending_policy": self.policy() or {"note": "no policy.json - outbound is unrestricted"},
"latch": self.latch_state(), "latch": self.latch_state(),
"ledger": { "ledger": {
"entries": len(self._ledger_entries()), "entries": len(self._ledger_entries()),

139
src/pacta/walletmcp.py
View file

@ -38,9 +38,76 @@ def _b64_to_bytes(field: str, value: Any) -> bytes:
class _ToolError(Exception): 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) super().__init__(reason)
self.payload = {"code": code, "reason": reason, "missing": missing, "remediation": remediation} 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]] = [ TOOLS: list[dict[str, Any]] = [
@ -48,6 +115,7 @@ TOOLS: list[dict[str, Any]] = [
"name": "wallet_status", "name": "wallet_status",
"description": "Custody posture: quorum members and tiers, latch state, ledger head and chain integrity, incident and refusal counts. Read this first.", "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}, "inputSchema": {"type": "object", "properties": {}, "additionalProperties": False},
"annotations": {"readOnlyHint": True},
}, },
{ {
"name": "verify_inbound", "name": "verify_inbound",
@ -63,25 +131,38 @@ TOOLS: list[dict[str, Any]] = [
"required": ["payload_b64", "signature_b64", "public_key_b64"], "required": ["payload_b64", "signature_b64", "public_key_b64"],
"additionalProperties": False, "additionalProperties": False,
}, },
"annotations": {"readOnlyHint": False, "destructiveHint": False, "idempotentHint": True},
}, },
{ {
"name": "request_signature", "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": { "inputSchema": {
"type": "object", "type": "object",
"properties": { "properties": {
"payload_b64": {"type": "string", "description": "message bytes to sign, base64"}, "payload_b64": {"type": "string", "description": "message bytes to sign, base64"},
"purpose": {"type": "string", "description": "why this signature is requested (recorded in the ledger)"}, "purpose": {"type": "string", "description": "why this signature is requested (recorded in the ledger)"},
"identity": {"type": "string", "description": "wallet identity name (default: warden)"}, "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"], "required": ["payload_b64", "purpose"],
"additionalProperties": False, "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", "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.", "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}, "inputSchema": {"type": "object", "properties": {}, "additionalProperties": False},
"annotations": {"readOnlyHint": True},
}, },
{ {
"name": "posture_challenge", "name": "posture_challenge",
@ -92,11 +173,13 @@ TOOLS: list[dict[str, Any]] = [
"required": ["nonce"], "required": ["nonce"],
"additionalProperties": False, "additionalProperties": False,
}, },
"annotations": {"readOnlyHint": False, "destructiveHint": False},
}, },
{ {
"name": "list_incidents", "name": "list_incidents",
"description": "Quorum divergences and firewall quarantines recorded by this wallet, newest-first, with severities and trails.", "description": "Quorum divergences and firewall quarantines recorded by this wallet, newest-first, with severities and trails.",
"inputSchema": {"type": "object", "properties": {}, "additionalProperties": False}, "inputSchema": {"type": "object", "properties": {}, "additionalProperties": False},
"annotations": {"readOnlyHint": True},
}, },
{ {
"name": "explain_refusal", "name": "explain_refusal",
@ -106,18 +189,26 @@ TOOLS: list[dict[str, Any]] = [
"properties": {"index": {"type": "integer", "description": "receipt number; omit for latest"}}, "properties": {"index": {"type": "integer", "description": "receipt number; omit for latest"}},
"additionalProperties": False, "additionalProperties": False,
}, },
"annotations": {"readOnlyHint": True},
}, },
] ]
class WalletMCP: class WalletMCP:
def __init__(self, wallet_dir: str | Path, log_url: str = "https://ltl.zkdefi.org") -> None: def __init__(
self.wallet = Wallet(wallet_dir) 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.log_url = log_url
self.limiter = _RateLimiter()
self.handlers: dict[str, Callable[[dict[str, Any]], dict[str, Any]]] = { self.handlers: dict[str, Callable[[dict[str, Any]], dict[str, Any]]] = {
"wallet_status": self._wallet_status, "wallet_status": self._wallet_status,
"verify_inbound": self._verify_inbound, "verify_inbound": self._verify_inbound,
"request_signature": self._request_signature, "request_signature": self._request_signature,
"airgap_pending": self._airgap_pending,
"custody_card": self._custody_card, "custody_card": self._custody_card,
"posture_challenge": self._posture_challenge, "posture_challenge": self._posture_challenge,
"list_incidents": self._list_incidents, "list_incidents": self._list_incidents,
@ -147,16 +238,43 @@ class WalletMCP:
purpose = args.get("purpose") purpose = args.get("purpose")
if not isinstance(purpose, str) or not purpose.strip(): if not isinstance(purpose, str) or not purpose.strip():
raise _ToolError("MALFORMED_INTENT", "purpose is required", ["purpose"], "state why the signature is requested") 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( 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): if isinstance(result, Refusal):
raise _ToolError( raise _refusal_error(result)
result.code, result.reason, result.missing, result.remediation
)
return 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]: def _custody_card(self, _: dict[str, Any]) -> dict[str, Any]:
return build_custody_card(self.wallet, self.log_url) return build_custody_card(self.wallet, self.log_url)
@ -166,7 +284,7 @@ class WalletMCP:
except ValueError as exc: except ValueError as exc:
raise _ToolError("MALFORMED_INTENT", str(exc), ["nonce"], "send an 8..128 character nonce") raise _ToolError("MALFORMED_INTENT", str(exc), ["nonce"], "send an 8..128 character nonce")
if isinstance(result, Refusal): if isinstance(result, Refusal):
raise _ToolError(result.code, result.reason, result.missing, result.remediation) raise _refusal_error(result)
return result return result
def _list_incidents(self, _: dict[str, Any]) -> dict[str, Any]: def _list_incidents(self, _: dict[str, Any]) -> dict[str, Any]:
@ -212,6 +330,7 @@ class WalletMCP:
if handler is None: if handler is None:
return self._err(msg_id, -32602, f"unknown tool {name}") return self._err(msg_id, -32602, f"unknown tool {name}")
try: try:
self.limiter.check(_TOOL_CATEGORY.get(name, "custody"))
payload = handler(args) payload = handler(args)
return self._ok(msg_id, { return self._ok(msg_id, {
"content": [{"type": "text", "text": json.dumps(payload, indent=2, sort_keys=True)}], "content": [{"type": "text", "text": json.dumps(payload, indent=2, sort_keys=True)}],

347
tests/test_wallet_hardening.py Normal file
View file

@ -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)