diff --git a/ESTATE.md b/ESTATE.md index f7ac60d..6381223 100644 --- a/ESTATE.md +++ b/ESTATE.md @@ -32,7 +32,7 @@ flowchart LR prov["provider service
check · append · publish · site code · templates (CI-pinned)"] sig["dogfood signer
verified-dalek binary"] lib["consumer library
receipts · pin store · R0–R5"] - wal["warden (code)
quorum wallet · MCP"] + wal["warden (code)
quorum wallet · MCP · cockpit (local, read-only)"] pap["paper
v0.9 + v0.1/v0.2 archives"] crs["course + llms.txt
14 notebooks"] key["SIGNING KEY (offline)"] @@ -118,7 +118,7 @@ deployed verifier; see the corpus KNOWN-GAPS ledger). | `dalek-` / `anza-` / `risc0-` / `betrusted-ed25519-verified` | subject | Rust source + Lean proofs; 16 certs each; attested (leaves 8–11, generations at 0–7) | frozen at attested commits; branch moves only for docs | | `pasta-pallas-verified` | subject | field layer proven; curve layer pending; **not attested** | changes freely | | `ltl-accumulator-verified` | subject | 61-cert corpus about the log's accumulator model; **entry-13 subject**, frozen `172a1d0` | frozen; doc-only commits allowed | -| `proof-aware-crypto-tooling-agent` (this repo) | machinery | provider service, consumer library, warden, dogfood signer, paper, course, tests | **changes freely — the hub** | +| `proof-aware-crypto-tooling-agent` (this repo) | machinery | provider service, consumer library, warden (+ local read-only cockpit), dogfood signer, paper, course, tests | **changes freely — the hub** | | `lean-transparency-log` | published | the public mirror: leaves, heads, receipts, fail-closed `verify.py` + selftest | **generated by publish** — canonical files here, templates in pacta, CI-pinned | | `verifying-crypto-with-lean` | published | undergraduate book; zero coupling to log state | changes freely | | `swisspost-evoting-go-poc` | consumer | operator's PoC; prospective consumer (family-level dalek match only) | independent | diff --git a/WALLET.md b/WALLET.md index 4bf8a2c..5158d19 100644 --- a/WALLET.md +++ b/WALLET.md @@ -143,6 +143,17 @@ pacta wallet mcp --wallet ./my-warden # stdio JSON-RPC MCP server --- +## The custody cockpit (human surface, read-only) + +`pacta wallet cockpit --wallet ` serves a local web UI for the +operator: posture (latch, ledger chain re-verified, pinned quorum), +the airgap signature queue (observed, never operated), the incident and +refusal browser, and a receipt inspector driven by the deployed +verifier. Design law: it renders recomputed evidence with provenance +lines, never cached status; it cannot approve, sign, unlatch, or modify +custody state (byte-level read-only guarantee in +`tests/test_walletui.py`). Details: [docs/cockpit.md](docs/cockpit.md). + ## Agent-native surface (MCP) `pacta wallet mcp` speaks MCP over stdio JSON-RPC. Eight outcome-first diff --git a/docs/cockpit.md b/docs/cockpit.md new file mode 100644 index 0000000..fdf071c --- /dev/null +++ b/docs/cockpit.md @@ -0,0 +1,51 @@ +# The custody cockpit — a read-only surface for the human operator + +`pacta wallet cockpit --wallet ` serves a local web UI +(default `http://127.0.0.1:8471`) over an existing warden wallet. +warden has always been agent-native (MCP) and CLI-native; the cockpit is +the third surface — for the human who ultimately answers for the money. + +## The design law + +**The cockpit renders evidence; it never asserts it.** Every panel is +recomputed at request time by the same functions the wallet itself uses +(`Wallet.posture()`, `Wallet.verify_ledger()`, directory listings, +`transparency.verify_receipt`), and every panel carries a provenance +line naming the function and the timestamp. Anything that cannot be +recomputed renders as a loud red FAILED-TO-VERIFY panel. There is no +cached green and no neutral gray — a cockpit that shows unverified green +lights would be the anti-warden. + +## The read-only guarantee + +The cockpit cannot approve, sign, unlatch, or modify custody state. It +calls only read paths; the one POST route (the receipt inspector) parses +submitted artifacts in memory and throwaway temp files, never near the +wallet directory. `tests/test_walletui.py` asserts this at the byte +level: a full request sweep, POST included, leaves every file in the +wallet directory hash-identical. Human approve/deny is deliberately NOT +here — that would be a custody-semantics change, which belongs to a +separate, explicitly reviewed milestone. + +## The four views + +| view | shows | recomputed by | +|---|---|---| +| **Posture** (`/`) | custody latch state, ledger head with full hash-chain re-verification, the pinned quorum members (backend, component, tier, source commit, binary hash), spending policy verbatim, incident/refusal counts | `Wallet.posture()` / `Wallet.verify_ledger()` | +| **Signature queue** (`/queue`) | parked airgap signing requests (outbox) and whether the device has answered (inbox) — observed, never operated | airgap outbox/inbox listing | +| **Incidents & refusals** (`/incidents`) | incident records and signed refusal receipts, verbatim, newest first | `incidents/*.json`, `receipts/*.json` | +| **Receipt inspector** (`/inspect`) | paste an attestation + transparency receipt + log public key; the verdict, per-signature results, and diagnostics come verbatim from the deployed verifier | `pacta.transparency.verify_receipt` | + +Every panel also states what it does **not** prove (e.g. the quorum +table says binary hashes are pinned but source-to-binary correspondence +is out of scope until reproducible builds). + +## Serving + +```bash +pacta wallet cockpit --wallet ~/my-wallet # 127.0.0.1:8471 +pacta wallet cockpit --wallet ~/my-wallet --port 9000 +``` + +The server binds localhost by default and is not meant to be exposed; +there is no authentication because there is nothing to operate. diff --git a/llms.txt b/llms.txt index b98ec5c..11e6d7d 100644 --- a/llms.txt +++ b/llms.txt @@ -13,7 +13,7 @@ - [ESTATE.md](ESTATE.md): the map of the whole endeavour — every repo, service, mirror, operator-held entity, and the two self-referential loops. - [README.md](README.md): what pacta is, the R0–R5 risk model, the dogfood loop. -- [WALLET.md](WALLET.md): warden, the verified-custody wallet — the quorum boundary, the signing firewall, the R4 gate, the MCP surface. +- [WALLET.md](WALLET.md): warden, the verified-custody wallet — the quorum boundary, the signing firewall, the R4 gate, the MCP surface, and the local read-only custody cockpit for the human operator (docs/cockpit.md). - [docs/agent-native.md](docs/agent-native.md): why the wallet is agent-native first (AX, MCP, A2A, AP2, x402, ERC-8004) and what each idea became. - [docs/products.md](docs/products.md): the four warden deployment profiles. diff --git a/src/pacta/cli.py b/src/pacta/cli.py index aae8fd9..86eaaf6 100644 --- a/src/pacta/cli.py +++ b/src/pacta/cli.py @@ -235,6 +235,12 @@ def build_parser() -> argparse.ArgumentParser: w_mcp.add_argument("--log-url", default="https://ltl.zkdefi.org") w_mcp.set_defaults(func=cmd_wallet_mcp) + w_cockpit = wsub.add_parser("cockpit", help="Serve the read-only custody cockpit (local web UI) for the human operator.") + w_cockpit.add_argument("--wallet", required=True) + w_cockpit.add_argument("--host", default="127.0.0.1", help="Bind address (default localhost; the cockpit is not meant to be exposed).") + w_cockpit.add_argument("--port", type=int, default=8471) + w_cockpit.set_defaults(func=cmd_wallet_cockpit) + w_ledger = wsub.add_parser("verify-ledger", help="Re-check the wallet's hash-chained ledger integrity.") w_ledger.add_argument("--wallet", required=True) w_ledger.set_defaults(func=cmd_wallet_verify_ledger) @@ -673,6 +679,18 @@ def cmd_wallet_card(args: argparse.Namespace) -> int: return 0 +def cmd_wallet_cockpit(args: argparse.Namespace) -> int: + from .walletui import serve + server = serve(args.wallet, host=args.host, port=args.port) + host, port = server.server_address[0], server.server_address[1] + print(f"warden cockpit (READ-ONLY) on http://{host}:{port} - Ctrl-C to stop") + try: + server.serve_forever() + except KeyboardInterrupt: + server.shutdown() + return 0 + + def cmd_wallet_mcp(args: argparse.Namespace) -> int: from .walletmcp import WalletMCP diff --git a/src/pacta/walletui.py b/src/pacta/walletui.py new file mode 100644 index 0000000..d195443 --- /dev/null +++ b/src/pacta/walletui.py @@ -0,0 +1,398 @@ +"""walletui - the warden custody cockpit (local, read-only). + +A localhost web surface over an existing wallet directory, for the human +operator who ultimately answers for the money. Four views: posture, the +pending-signature queue (airgap outbox), the incident & refusal browser, +and a receipt inspector. + +Design law: THE COCKPIT RENDERS EVIDENCE, IT NEVER ASSERTS IT. Every +panel is recomputed from wallet state or submitted artifacts at request +time by the same functions the wallet itself uses, and every panel names +the function and timestamp that produced it. Anything that cannot be +recomputed renders as a loud FAILED-TO-VERIFY panel - there is no cached +green and no neutral gray. + +Read-only guarantee: this module calls only read paths (``Wallet.posture``, +``verify_ledger``, directory listings) and ``transparency.verify_receipt`` +on submitted artifacts (parsed in memory / temp files outside the wallet). +It cannot approve, sign, unlatch, or modify custody state; the HTTP surface +exposes no mutating route. Human approve/deny is deliberately NOT here - +that would be a custody-semantics change, which belongs to a separate, +explicitly reviewed milestone. + +The server binds 127.0.0.1 by default and is not meant to be exposed. +""" +from __future__ import annotations + +import html +import json +import tempfile +import urllib.parse +from datetime import datetime, timezone +from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer +from pathlib import Path +from typing import Any, Callable + +from .attestation import load_attestation +from .transparency import load_receipt, verify_receipt +from .wallet import Wallet + +_STYLE = """ + :root{--ink:#1c2430;--ink2:#5a6675;--line:#dde2e9;--ok:#1e7f4f;--okbg:#e2f2e9; + --bad:#a3242c;--badbg:#fbe4e6;--warn:#a86a10;--warnbg:#fdf0da; + --accent:#3b4d8f;--accentbg:#eef0f7;--bg:#f8f9fa} + *{box-sizing:border-box} + body{font-family:system-ui,sans-serif;max-width:62rem;margin:0 auto; + padding:1.4rem 1.2rem 4rem;color:var(--ink);line-height:1.55;background:var(--bg)} + h1{font-size:1.35rem;margin:.2rem 0 0} + h2{font-size:1.05rem;margin:1.6rem 0 .5rem} + code{font-family:ui-monospace,Menlo,Consolas,monospace;background:#eef0f3; + border-radius:4px;padding:.08rem .3rem;font-size:.88em} + nav{margin:.7rem 0 1rem;display:flex;gap:.5rem;flex-wrap:wrap} + nav a{color:var(--accent);text-decoration:none;border:1px solid var(--line); + background:#fff;border-radius:6px;padding:.25rem .7rem;font-size:.85rem} + nav a.here{border-color:var(--accent);font-weight:600} + .banner{background:var(--warnbg);border:1px solid var(--warn);color:var(--warn); + border-radius:6px;padding:.45rem .8rem;font-size:.82rem;font-weight:600} + .panel{background:#fff;border:1px solid var(--line);border-radius:8px; + padding:.9rem 1.1rem;margin:.7rem 0} + .panel.bad{border-color:var(--bad);background:var(--badbg)} + .prov{color:var(--ink2);font-size:.72rem;margin-top:.6rem;border-top:1px dashed var(--line); + padding-top:.35rem} + .pill{display:inline-block;border-radius:9px;padding:.06rem .55rem;font-size:.76rem; + font-weight:700} + .pill.ok{background:var(--okbg);color:var(--ok)} + .pill.bad{background:var(--badbg);color:var(--bad)} + .pill.warn{background:var(--warnbg);color:var(--warn)} + table{border-collapse:collapse;width:100%;font-size:.88rem;background:#fff} + td,th{border:1px solid var(--line);padding:.4rem .6rem;text-align:left;vertical-align:top} + th{background:var(--accentbg)} + ul.diag{margin:.4rem 0 0;padding-left:1.2rem} + ul.diag li{font-size:.85rem;margin:.2rem 0} + textarea{width:100%;min-height:7.5rem;font-family:ui-monospace,monospace;font-size:.8rem; + border:1px solid var(--line);border-radius:6px;padding:.5rem} + button{background:var(--accent);color:#fff;border:0;border-radius:6px; + padding:.5rem 1.1rem;font-size:.9rem;cursor:pointer} + .muted{color:var(--ink2);font-size:.85rem} + .mono{font-family:ui-monospace,monospace} +""" + + +def _now() -> str: + return datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ") + + +def _esc(value: Any) -> str: + return html.escape(str(value)) + + +def _provenance(via: str) -> str: + return f'
recomputed {_esc(_now())} via {_esc(via)} — nothing on this panel is cached or asserted.
' + + +def _failed_panel(what: str, via: str, error: Exception) -> str: + return ( + f'
FAILED TO VERIFY ' + f"{_esc(what)} could not be recomputed: " + f"{_esc(f'{type(error).__name__}: {error}')}. " + f"A cockpit that cannot verify shows red, never a stale green." + f"{_provenance(via)}
" + ) + + +# --------------------------------------------------------------------------- +# collectors - read-only, one wallet function each, exceptions contained +# --------------------------------------------------------------------------- + +def collect(via: str, fn: Callable[[], Any]) -> dict[str, Any]: + try: + return {"ok": True, "via": via, "data": fn()} + except Exception as error: # noqa: BLE001 - fail-closed rendering is the point + return {"ok": False, "via": via, "error": error} + + +def collect_incidents(wallet: Wallet) -> dict[str, Any]: + def read() -> list[dict[str, Any]]: + items = [] + for path in sorted(wallet.incidents_dir.glob("*.json"), reverse=True): + record = json.loads(path.read_text(encoding="utf-8")) + record["_file"] = path.name + items.append(record) + return items + return collect("incidents/*.json (verbatim files)", read) + + +def collect_refusals(wallet: Wallet) -> dict[str, Any]: + def read() -> list[dict[str, Any]]: + items = [] + for path in sorted(wallet.receipts_dir.glob("*.json"), reverse=True): + record = json.loads(path.read_text(encoding="utf-8")) + record["_file"] = path.name + items.append(record) + return items + return collect("receipts/*.json (refusal receipts, verbatim)", read) + + +def collect_airgap(wallet: Wallet) -> dict[str, Any]: + def read() -> list[dict[str, Any]]: + pending = [] + outbox = wallet.airgap_dir / "outbox" + inbox = 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"), + "answered": (inbox / f"{request_id}.response.json").exists(), + }) + return pending + return collect("airgap/outbox + inbox listing", read) + + +def inspect_receipt(attestation_text: str, receipt_text: str, + public_key_pem: str) -> dict[str, Any]: + """Run the SAME verification the wallet and CLI use on pasted artifacts. + + Nothing is written anywhere near the wallet; artifacts live in a + throwaway temp directory for the duration of the call. + """ + try: + with tempfile.TemporaryDirectory() as tmp: + att_path = Path(tmp) / "attestation.json" + rec_path = Path(tmp) / "receipt.json" + key_path = Path(tmp) / "log.pub" + att_path.write_text(attestation_text, encoding="utf-8") + rec_path.write_text(receipt_text, encoding="utf-8") + key_path.write_text(public_key_pem, encoding="utf-8") + attestation = load_attestation(att_path) + receipt = load_receipt(rec_path) + result = verify_receipt(attestation, receipt, key_path, + require_signatures="ed25519") + return { + "ok": True, + "via": "pacta.transparency.verify_receipt (the deployed verifier itself)", + "accepted": bool(result.accepted), + "signatures": dict(result.signatures), + "diagnostics": list(result.diagnostics), + } + except Exception as error: # noqa: BLE001 + return {"ok": False, + "via": "pacta.transparency.verify_receipt", + "error": f"{type(error).__name__}: {error}"} + + +# --------------------------------------------------------------------------- +# renderers - pure string builders over collector output +# --------------------------------------------------------------------------- + +_VIEWS = [("/", "Posture"), ("/queue", "Signature queue"), + ("/incidents", "Incidents & refusals"), ("/inspect", "Receipt inspector")] + + +def _page(title: str, active: str, body: str, wallet_dir: str) -> str: + nav = "".join( + f'{label}' + for href, label in _VIEWS) + return ( + "" + f"warden cockpit — {_esc(title)}" + f"" + f"

warden cockpit {_esc(wallet_dir)}

" + "" + f"{body}" + ) + + +def render_posture(posture: dict[str, Any]) -> str: + if not posture["ok"]: + return _failed_panel("Custody posture", posture["via"], posture["error"]) + p = posture["data"] + latch = p["latch"] + ledger = p["ledger"] + latch_pill = ('LATCHED — outbound custody frozen' + if latch.get("latched") else 'unlatched') + chain_pill = ('chain verified' if ledger["chain_ok"] + else 'CHAIN BROKEN') + members = "".join( + f"{_esc(m['backend'])}" + f"{_esc(m['component'])}" + f"{_esc(m['risk_tier'])}" + f"{_esc(m['source_commit'][:12])}…" + f"{_esc(m['binary_sha256'][:16])}…" + for m in p["members"]) + problems = "".join(f"
  • {_esc(x)}
  • " for x in ledger["problems"]) or "
  • none
  • " + latch_detail = "" + if latch.get("latched"): + latch_detail = (f"

    reason: {_esc(latch.get('reason'))} · " + f"incident: {_esc(latch.get('incident'))} · " + f"since {_esc(latch.get('at'))} — see the " + f"incident browser and docs/runbook-latch.md.

    ") + spending = p.get("spending_policy") or {} + return ( + f"

    Custody latch {latch_pill}

    " + f"{latch_detail}{_provenance('Wallet.latch_state()')}
    " + f"

    Ledger {chain_pill}

    " + f"

    {ledger['entries']} entries · head {_esc(ledger['head'][:24])}…

    " + f"
      {problems}
    " + f"{_provenance('Wallet.verify_ledger() — full hash-chain recomputation')}
    " + f"

    Quorum members " + f"{len(p['members'])} pinned

    " + "" + "" + f"{members}
    backendcomponenttiersource commitbinary sha256
    " + "

    Every member is pinned by binary hash in the capsule; the capsule " + f"hash is {_esc(p['capsule_sha256'][:24])}…. What this table does NOT " + "prove: that the binaries correspond to the attested sources (reproducible builds " + "are out of scope, stated in the paper and the claim cards)." + f"{_provenance('Wallet.capsule() / Wallet.posture()')}

    " + f"

    Spending policy

    " + f"
    {_esc(json.dumps(spending, indent=2, sort_keys=True))}
    " + f"{_provenance('Wallet.policy() (policy.json, verbatim)')}
    " + f"

    Counters

    " + f"

    incidents: {p['incidents']} · refusal receipts: " + f"{p['refusal_receipts']} — browse them under " + "Incidents & refusals.

    " + f"{_provenance('directory counts, recomputed')}
    " + ) + + +def render_queue(airgap: dict[str, Any]) -> str: + if not airgap["ok"]: + return _failed_panel("Signature queue", airgap["via"], airgap["error"]) + rows = "".join( + f"{_esc(r['request_id'])}" + f"{_esc(r.get('created_at'))}" + f"{_esc((r.get('payload_sha256') or '')[:24])}…" + f"{'answered' if r['answered'] else 'awaiting device'}" + for r in airgap["data"]) + body = (f"" + f"{rows}
    requestcreatedpayload sha256state
    " if airgap["data"] + else "

    No parked signing requests.

    ") + return ( + "

    Pending airgap signatures

    " + + body + + "

    This queue is OBSERVED, not operated: completing or refusing a " + "request happens through the wallet's own channels (request_signature " + "over MCP, or the airgap device flow), never from this page.

    " + + _provenance("airgap outbox/inbox listing") + "
    " + ) + + +def render_incidents(incidents: dict[str, Any], refusals: dict[str, Any]) -> str: + def block(title: str, coll: dict[str, Any], via_note: str) -> str: + if not coll["ok"]: + return _failed_panel(title, coll["via"], coll["error"]) + items = coll["data"] + if not items: + body = "

    none recorded

    " + else: + body = "".join( + f"
    {_esc(i['_file'])}" + f"
    {_esc(json.dumps({k: v for k, v in i.items() if k != '_file'}, indent=2, sort_keys=True))}
    " + for i in items[:50]) + return (f"

    {_esc(title)}

    {body}" + f"{_provenance(via_note)}
    ") + return (block("Incidents (quorum divergences, quarantines)", incidents, + "incidents/*.json, verbatim, newest first") + + block("Refusal receipts (signed, machine-actionable)", refusals, + "receipts/*.json, verbatim, newest first")) + + +def render_inspect(result: dict[str, Any] | None, + defaults: dict[str, str] | None = None) -> str: + d = defaults or {} + verdict = "" + if result is not None: + if not result["ok"]: + verdict = _failed_panel("Receipt verification", result["via"], + RuntimeError(result["error"])) + else: + pill = ('ACCEPTED' if result["accepted"] + else 'REJECTED') + sigs = "".join(f"{_esc(k)}{_esc(v)}" + for k, v in sorted(result["signatures"].items())) + diags = "".join(f"
  • {_esc(x)}
  • " for x in result["diagnostics"]) or "
  • none
  • " + verdict = ( + f"

    Verdict {pill}

    " + f"{sigs}
    signature checkresult
    " + f"

    Diagnostics

      {diags}
    " + f"{_provenance(result['via'])}
    ") + return ( + verdict + + "

    Inspect a receipt

    " + "

    Paste an attestation, its transparency receipt, and the log's " + "public key. The verdict is produced by the wallet's own deployed verifier — " + "this page adds nothing and hides nothing; the diagnostics list is verbatim.

    " + "
    " + f"

    attestation.json

    " + f"

    receipt.json

    " + f"

    log public key (PEM)

    " + "
    " + ) + + +# --------------------------------------------------------------------------- +# server +# --------------------------------------------------------------------------- + +def make_handler(wallet_dir: Path): + class CockpitHandler(BaseHTTPRequestHandler): + server_version = "warden-cockpit/1" + + def _send(self, body: str, status: int = 200) -> None: + data = body.encode("utf-8") + self.send_response(status) + self.send_header("Content-Type", "text/html; charset=utf-8") + self.send_header("Content-Length", str(len(data))) + self.send_header("Cache-Control", "no-store") + self.end_headers() + self.wfile.write(data) + + def _wallet(self) -> Wallet: + return Wallet(wallet_dir) + + def do_GET(self) -> None: # noqa: N802 - http.server API + route = urllib.parse.urlparse(self.path).path + wd = str(wallet_dir) + if route == "/": + wallet = self._wallet() + body = render_posture(collect("Wallet.posture()", wallet.posture)) + self._send(_page("posture", "/", body, wd)) + elif route == "/queue": + body = render_queue(collect_airgap(self._wallet())) + self._send(_page("signature queue", "/queue", body, wd)) + elif route == "/incidents": + wallet = self._wallet() + body = render_incidents(collect_incidents(wallet), collect_refusals(wallet)) + self._send(_page("incidents", "/incidents", body, wd)) + elif route == "/inspect": + self._send(_page("receipt inspector", "/inspect", render_inspect(None), wd)) + else: + self._send(_page("not found", "", "
    No such view.
    ", wd), 404) + + def do_POST(self) -> None: # noqa: N802 + route = urllib.parse.urlparse(self.path).path + if route != "/inspect": + self._send("
    No such action.
    ", 404) + return + length = int(self.headers.get("Content-Length", "0")) + form = urllib.parse.parse_qs(self.rfile.read(length).decode("utf-8")) + fields = {k: form.get(k, [""])[0] for k in ("attestation", "receipt", "pubkey")} + result = inspect_receipt(fields["attestation"], fields["receipt"], fields["pubkey"]) + self._send(_page("receipt inspector", "/inspect", + render_inspect(result, fields), str(wallet_dir))) + + def log_message(self, fmt: str, *args: Any) -> None: # quiet + return + + return CockpitHandler + + +def serve(wallet_dir: str | Path, host: str = "127.0.0.1", port: int = 8471) -> ThreadingHTTPServer: + wallet_dir = Path(wallet_dir).resolve() + Wallet(wallet_dir).capsule() # fail fast if this is not a wallet + server = ThreadingHTTPServer((host, port), make_handler(wallet_dir)) + return server diff --git a/tests/test_walletui.py b/tests/test_walletui.py new file mode 100644 index 0000000..788b978 --- /dev/null +++ b/tests/test_walletui.py @@ -0,0 +1,197 @@ +"""Cockpit tests: the design law is testable — every panel renders +recomputed evidence with provenance, failures render loud, and the HTTP +surface cannot mutate wallet state (asserted by hashing the wallet +directory before and after a full request sweep, including a POST).""" + +import hashlib +import json +import stat +import threading +import urllib.request +import urllib.parse +from pathlib import Path + +import pytest + +from pacta.quorum import binary_path +from pacta.signing import generate_ed25519_keypair +from pacta.wallet import Wallet +from pacta.walletui import (collect, collect_airgap, collect_incidents, + collect_refusals, inspect_receipt, render_incidents, + render_inspect, render_posture, render_queue, serve) + + +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) -> Wallet: + state_dir = tmp_path / "state" + state_dir.mkdir(parents=True, exist_ok=True) + members = [] + for name in ("dalek", "anza"): + binary = binary_path(name, state_dir) + _fake_member(binary, "accept") + members.append({ + "backend": name, + "component": f"{name}-ed25519-verified", + "semantics": "test", "entry_point": "test", + "source_commit": "deadbeef" * 5, "repo_commit": "cafe" * 10, + "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": [], + } + capsule_bytes = json.dumps(capsule, sort_keys=True, separators=(",", ":")).encode() + wallet.capsule_path.write_text(json.dumps(capsule, indent=2, sort_keys=True) + "\n") + wallet._append_ledger("genesis", {"type": "pacta.wallet.ledger_genesis.v1", + "capsule_sha256": _sha256(capsule_bytes)}) + generate_ed25519_keypair(wallet.keys_dir / "warden.key.pem", + wallet.keys_dir / "warden.pub.pem") + return wallet + + +def _dir_fingerprint(root: Path) -> dict[str, str]: + out = {} + for path in sorted(root.rglob("*")): + if path.is_file(): + out[str(path.relative_to(root))] = _sha256(path.read_bytes()) + return out + + +def test_posture_renders_recomputed_evidence(tmp_path): + wallet = _seal_wallet(tmp_path) + html = render_posture(collect("Wallet.posture()", wallet.posture)) + assert "chain verified" in html + assert "unlatched" in html + assert "dalek-ed25519-verified" in html and "anza-ed25519-verified" in html + assert "recomputed" in html and "Wallet.verify_ledger()" in html + # the panel states its own honesty boundary + assert "does NOT" in html and "reproducible builds" in html.lower() + + +def test_broken_ledger_renders_red(tmp_path): + wallet = _seal_wallet(tmp_path) + ledger = wallet.dir / "ledger.jsonl" + ledger.write_text(ledger.read_text().replace("genesis", "gene-sis"), encoding="utf-8") + html = render_posture(collect("Wallet.posture()", wallet.posture)) + assert "CHAIN BROKEN" in html + + +def test_collector_failure_is_loud_not_gray(tmp_path): + def boom(): + raise RuntimeError("evidence unavailable") + html = render_posture(collect("Wallet.posture()", boom)) + assert "FAILED TO VERIFY" in html and "evidence unavailable" in html + assert "pill ok" not in html # no green anywhere on a failed panel + + +def test_latched_state_renders_frozen(): + posture = {"ok": True, "via": "x", "data": { + "capsule_sha256": "ab" * 32, + "members": [], "policy": {}, "spending_policy": {}, + "latch": {"latched": True, "reason": "quorum divergence", + "incident": "incident-1.json", "at": "2026-07-19T00:00:00Z"}, + "ledger": {"entries": 1, "head": "cd" * 32, "chain_ok": True, "problems": []}, + "incidents": 1, "refusal_receipts": 0, "generated_at": "now"}} + html = render_posture(posture) + assert "LATCHED" in html and "outbound custody frozen" in html + assert "quorum divergence" in html and "runbook-latch" in html + + +def test_queue_lists_airgap_requests(tmp_path): + wallet = _seal_wallet(tmp_path) + outbox = wallet.airgap_dir / "outbox" + (outbox / "req-1.request.json").write_text(json.dumps( + {"created_at": "2026-07-19T00:00:00Z", "payload_sha256": "aa" * 32})) + html = render_queue(collect_airgap(wallet)) + assert "req-1" in html and "awaiting device" in html + (wallet.airgap_dir / "inbox" / "req-1.response.json").write_text("{}") + html = render_queue(collect_airgap(wallet)) + assert "answered" in html + # observed, never operated + assert "OBSERVED" in html + + +def test_incidents_and_refusals_render_verbatim(tmp_path): + wallet = _seal_wallet(tmp_path) + (wallet.incidents_dir / "incident-1.json").write_text(json.dumps( + {"severity": "tamper", "detail": "member disagreement on payload"})) + (wallet.receipts_dir / "refusal-1.json").write_text(json.dumps( + {"code": "CUSTODY_LATCHED", "remediation": "see runbook"})) + html = render_incidents(collect_incidents(wallet), collect_refusals(wallet)) + assert "incident-1.json" in html and "member disagreement" in html + assert "refusal-1.json" in html and "CUSTODY_LATCHED" in html + + +def test_inspect_garbage_fails_closed(): + result = inspect_receipt("not json at all", "{}", "") + assert result["ok"] is False + html = render_inspect(result) + assert "FAILED TO VERIFY" in html + assert "ACCEPTED" not in html + + +def test_inspect_real_evidence_roundtrip(): + evidence = Path("examples") / "wallet-evidence" + key = evidence / "log.pub" + attestations = sorted(evidence.glob("*attestation*.json")) if evidence.exists() else [] + receipts = sorted(evidence.glob("*receipt*.json")) if evidence.exists() else [] + if not (key.exists() and attestations and receipts): + pytest.skip("example wallet evidence not present") + result = inspect_receipt(attestations[0].read_text(), receipts[0].read_text(), + key.read_text()) + assert result["ok"] is True + html = render_inspect(result) + assert ("ACCEPTED" in html) or ("REJECTED" in html) + assert "verify_receipt" in html # provenance names the deployed verifier + + +def test_server_routes_and_read_only_guarantee(tmp_path): + wallet = _seal_wallet(tmp_path) + before = _dir_fingerprint(wallet.dir) + server = serve(wallet.dir, host="127.0.0.1", port=0) + port = server.server_address[1] + thread = threading.Thread(target=server.serve_forever, daemon=True) + thread.start() + try: + for route in ("/", "/queue", "/incidents", "/inspect"): + with urllib.request.urlopen(f"http://127.0.0.1:{port}{route}") as resp: + body = resp.read().decode() + assert resp.status == 200 + assert "READ-ONLY" in body + data = urllib.parse.urlencode( + {"attestation": "junk", "receipt": "junk", "pubkey": "junk"}).encode() + with urllib.request.urlopen(f"http://127.0.0.1:{port}/inspect", data=data) as resp: + assert resp.status == 200 + assert "FAILED TO VERIFY" in resp.read().decode() + finally: + server.shutdown() + thread.join(timeout=5) + # the whole sweep, POST included, changed not one byte of wallet state + assert _dir_fingerprint(wallet.dir) == before + + +def test_serve_refuses_non_wallet(tmp_path): + with pytest.raises(Exception): + serve(tmp_path / "empty")