mirror of
https://github.com/saymrwulf/proof-aware-crypto-tooling-agent.git
synced 2026-09-04 20:03:40 +00:00
warden cockpit: local read-only custody UI for the human operator
Operator-ordered milestone, built under an explicit no-risk constraint while the paper application is in flight: strictly ADDITIVE (one new module + one CLI subcommand; zero changes to wallet semantics, the public site, or anything the paper describes), LOCAL-ONLY (never deployed; binds 127.0.0.1), and READ-ONLY in v1 (human approve/deny is deliberately deferred — that is a custody-semantics change belonging to a separately reviewed milestone). New: src/pacta/walletui.py — four views over an existing wallet: - Posture: latch state, ledger head with FULL hash-chain re-verification, pinned quorum members, spending policy verbatim. - Signature queue: airgap outbox/inbox, observed never operated. - Incidents & refusals: verbatim records, newest first. - Receipt inspector: paste attestation + receipt + log key; verdict, per-signature results, and diagnostics come verbatim from transparency.verify_receipt — the deployed verifier itself, reused not reimplemented. Design law, stated in the module and enforced by tests: THE COCKPIT RENDERS EVIDENCE, IT NEVER ASSERTS IT. Every panel recomputes at request time and carries a provenance line (function + timestamp); anything unverifiable renders a loud FAILED-TO-VERIFY panel — no cached green, no neutral gray. Panels state what they do NOT prove (e.g. binary pinning vs source-to-binary correspondence). CLI: pacta wallet cockpit --wallet DIR [--host --port]. Tests (10 new, suite 118 -> 128 green): recomputed-evidence rendering, CHAIN BROKEN on tampered ledger, loud failure on collector error, latched-state rendering, airgap queue states, verbatim incident/refusal rendering, fail-closed inspector on garbage, real-evidence roundtrip (skips if example evidence absent), full-sweep BYTE-LEVEL read-only guarantee (wallet dir hash-identical after all GETs + a POST), and fail-fast on non-wallet dirs. Rendering additionally verified visually against a live demo wallet (posture, quorum table, incidents). Docs: docs/cockpit.md; WALLET.md section; ESTATE.md + llms.txt updated. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
This commit is contained in:
parent
609aa2419a
commit
dcb48f453e
7 changed files with 678 additions and 3 deletions
|
|
@ -32,7 +32,7 @@ flowchart LR
|
|||
prov["provider service<br/>check · append · publish · site code · templates (CI-pinned)"]
|
||||
sig["dogfood signer<br/>verified-dalek binary"]
|
||||
lib["consumer library<br/>receipts · pin store · R0–R5"]
|
||||
wal["warden (code)<br/>quorum wallet · MCP"]
|
||||
wal["warden (code)<br/>quorum wallet · MCP · cockpit (local, read-only)"]
|
||||
pap["paper<br/>v0.9 + v0.1/v0.2 archives"]
|
||||
crs["course + llms.txt<br/>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 |
|
||||
|
|
|
|||
11
WALLET.md
11
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 <dir>` 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
|
||||
|
|
|
|||
51
docs/cockpit.md
Normal file
51
docs/cockpit.md
Normal file
|
|
@ -0,0 +1,51 @@
|
|||
# The custody cockpit — a read-only surface for the human operator
|
||||
|
||||
`pacta wallet cockpit --wallet <dir>` 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.
|
||||
2
llms.txt
2
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.
|
||||
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
||||
|
|
|
|||
398
src/pacta/walletui.py
Normal file
398
src/pacta/walletui.py
Normal file
|
|
@ -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'<div class="prov">recomputed {_esc(_now())} via <code>{_esc(via)}</code> — nothing on this panel is cached or asserted.</div>'
|
||||
|
||||
|
||||
def _failed_panel(what: str, via: str, error: Exception) -> str:
|
||||
return (
|
||||
f'<div class="panel bad"><span class="pill bad">FAILED TO VERIFY</span> '
|
||||
f"<strong>{_esc(what)}</strong> could not be recomputed: "
|
||||
f"<code>{_esc(f'{type(error).__name__}: {error}')}</code>. "
|
||||
f"A cockpit that cannot verify shows red, never a stale green."
|
||||
f"{_provenance(via)}</div>"
|
||||
)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# 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'<a href="{href}"{" class=here" if href == active else ""}>{label}</a>'
|
||||
for href, label in _VIEWS)
|
||||
return (
|
||||
"<!doctype html><html><head><meta charset='utf-8'>"
|
||||
f"<title>warden cockpit — {_esc(title)}</title>"
|
||||
f"<style>{_STYLE}</style></head><body>"
|
||||
f"<h1>warden cockpit <span class='muted mono'>{_esc(wallet_dir)}</span></h1>"
|
||||
"<div class='banner'>READ-ONLY. This cockpit observes and recomputes; it cannot "
|
||||
"approve, sign, unlatch, or modify custody state.</div>"
|
||||
f"<nav>{nav}</nav>{body}</body></html>"
|
||||
)
|
||||
|
||||
|
||||
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 = ('<span class="pill bad">LATCHED — outbound custody frozen</span>'
|
||||
if latch.get("latched") else '<span class="pill ok">unlatched</span>')
|
||||
chain_pill = ('<span class="pill ok">chain verified</span>' if ledger["chain_ok"]
|
||||
else '<span class="pill bad">CHAIN BROKEN</span>')
|
||||
members = "".join(
|
||||
f"<tr><td><code>{_esc(m['backend'])}</code></td>"
|
||||
f"<td class='mono'>{_esc(m['component'])}</td>"
|
||||
f"<td>{_esc(m['risk_tier'])}</td>"
|
||||
f"<td class='mono'>{_esc(m['source_commit'][:12])}…</td>"
|
||||
f"<td class='mono'>{_esc(m['binary_sha256'][:16])}…</td></tr>"
|
||||
for m in p["members"])
|
||||
problems = "".join(f"<li>{_esc(x)}</li>" for x in ledger["problems"]) or "<li>none</li>"
|
||||
latch_detail = ""
|
||||
if latch.get("latched"):
|
||||
latch_detail = (f"<p>reason: <code>{_esc(latch.get('reason'))}</code> · "
|
||||
f"incident: <code>{_esc(latch.get('incident'))}</code> · "
|
||||
f"since {_esc(latch.get('at'))} — see the "
|
||||
f"<a href='/incidents'>incident browser</a> and docs/runbook-latch.md.</p>")
|
||||
spending = p.get("spending_policy") or {}
|
||||
return (
|
||||
f"<div class='panel'><h2 style='margin-top:0'>Custody latch {latch_pill}</h2>"
|
||||
f"{latch_detail}{_provenance('Wallet.latch_state()')}</div>"
|
||||
f"<div class='panel'><h2 style='margin-top:0'>Ledger {chain_pill}</h2>"
|
||||
f"<p>{ledger['entries']} entries · head <code>{_esc(ledger['head'][:24])}…</code></p>"
|
||||
f"<ul class='diag'>{problems}</ul>"
|
||||
f"{_provenance('Wallet.verify_ledger() — full hash-chain recomputation')}</div>"
|
||||
f"<div class='panel'><h2 style='margin-top:0'>Quorum members "
|
||||
f"<span class='pill ok'>{len(p['members'])} pinned</span></h2>"
|
||||
"<table><tr><th>backend</th><th>component</th><th>tier</th>"
|
||||
"<th>source commit</th><th>binary sha256</th></tr>"
|
||||
f"{members}</table>"
|
||||
"<p class='muted'>Every member is pinned by binary hash in the capsule; the capsule "
|
||||
f"hash is <code>{_esc(p['capsule_sha256'][:24])}…</code>. 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()')}</div>"
|
||||
f"<div class='panel'><h2 style='margin-top:0'>Spending policy</h2>"
|
||||
f"<pre style='margin:0;font-size:.8rem'>{_esc(json.dumps(spending, indent=2, sort_keys=True))}</pre>"
|
||||
f"{_provenance('Wallet.policy() (policy.json, verbatim)')}</div>"
|
||||
f"<div class='panel'><h2 style='margin-top:0'>Counters</h2>"
|
||||
f"<p>incidents: <strong>{p['incidents']}</strong> · refusal receipts: "
|
||||
f"<strong>{p['refusal_receipts']}</strong> — browse them under "
|
||||
"<a href='/incidents'>Incidents & refusals</a>.</p>"
|
||||
f"{_provenance('directory counts, recomputed')}</div>"
|
||||
)
|
||||
|
||||
|
||||
def render_queue(airgap: dict[str, Any]) -> str:
|
||||
if not airgap["ok"]:
|
||||
return _failed_panel("Signature queue", airgap["via"], airgap["error"])
|
||||
rows = "".join(
|
||||
f"<tr><td class='mono'>{_esc(r['request_id'])}</td>"
|
||||
f"<td>{_esc(r.get('created_at'))}</td>"
|
||||
f"<td class='mono'>{_esc((r.get('payload_sha256') or '')[:24])}…</td>"
|
||||
f"<td>{'<span class=\"pill ok\">answered</span>' if r['answered'] else '<span class=\"pill warn\">awaiting device</span>'}</td></tr>"
|
||||
for r in airgap["data"])
|
||||
body = (f"<table><tr><th>request</th><th>created</th><th>payload sha256</th>"
|
||||
f"<th>state</th></tr>{rows}</table>" if airgap["data"]
|
||||
else "<p class='muted'>No parked signing requests.</p>")
|
||||
return (
|
||||
"<div class='panel'><h2 style='margin-top:0'>Pending airgap signatures</h2>"
|
||||
+ body +
|
||||
"<p class='muted'>This queue is OBSERVED, not operated: completing or refusing a "
|
||||
"request happens through the wallet's own channels (<code>request_signature</code> "
|
||||
"over MCP, or the airgap device flow), never from this page.</p>"
|
||||
+ _provenance("airgap outbox/inbox listing") + "</div>"
|
||||
)
|
||||
|
||||
|
||||
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 = "<p class='muted'>none recorded</p>"
|
||||
else:
|
||||
body = "".join(
|
||||
f"<div class='panel' style='margin:.5rem 0'><code>{_esc(i['_file'])}</code>"
|
||||
f"<pre style='font-size:.76rem;overflow-x:auto'>{_esc(json.dumps({k: v for k, v in i.items() if k != '_file'}, indent=2, sort_keys=True))}</pre></div>"
|
||||
for i in items[:50])
|
||||
return (f"<div class='panel'><h2 style='margin-top:0'>{_esc(title)}</h2>{body}"
|
||||
f"{_provenance(via_note)}</div>")
|
||||
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 = ('<span class="pill ok">ACCEPTED</span>' if result["accepted"]
|
||||
else '<span class="pill bad">REJECTED</span>')
|
||||
sigs = "".join(f"<tr><td><code>{_esc(k)}</code></td><td>{_esc(v)}</td></tr>"
|
||||
for k, v in sorted(result["signatures"].items()))
|
||||
diags = "".join(f"<li>{_esc(x)}</li>" for x in result["diagnostics"]) or "<li>none</li>"
|
||||
verdict = (
|
||||
f"<div class='panel'><h2 style='margin-top:0'>Verdict {pill}</h2>"
|
||||
f"<table><tr><th>signature check</th><th>result</th></tr>{sigs}</table>"
|
||||
f"<h2>Diagnostics</h2><ul class='diag'>{diags}</ul>"
|
||||
f"{_provenance(result['via'])}</div>")
|
||||
return (
|
||||
verdict +
|
||||
"<div class='panel'><h2 style='margin-top:0'>Inspect a receipt</h2>"
|
||||
"<p class='muted'>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.</p>"
|
||||
"<form method='post' action='/inspect'>"
|
||||
f"<p><strong>attestation.json</strong><br><textarea name='attestation'>{_esc(d.get('attestation', ''))}</textarea></p>"
|
||||
f"<p><strong>receipt.json</strong><br><textarea name='receipt'>{_esc(d.get('receipt', ''))}</textarea></p>"
|
||||
f"<p><strong>log public key (PEM)</strong><br><textarea name='pubkey' style='min-height:4rem'>{_esc(d.get('pubkey', ''))}</textarea></p>"
|
||||
"<button type='submit'>Verify (read-only)</button></form></div>"
|
||||
)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# 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", "", "<div class='panel bad'>No such view.</div>", wd), 404)
|
||||
|
||||
def do_POST(self) -> None: # noqa: N802
|
||||
route = urllib.parse.urlparse(self.path).path
|
||||
if route != "/inspect":
|
||||
self._send("<div class='panel bad'>No such action.</div>", 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
|
||||
197
tests/test_walletui.py
Normal file
197
tests/test_walletui.py
Normal file
|
|
@ -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")
|
||||
Loading…
Reference in a new issue