Compare commits

...

4 commits

Author SHA1 Message Date
dcb48f453e 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>
2026-07-19 18:02:17 +02:00
609aa2419a ESTATE.md: unname the private infrastructure layer (operator review)
A public map should list only entities whose existence is already
public or must be public for trust. The private infra repo is now 'the
deployment configuration, maintained privately' (node, edge, and
inventory row removed); the signing-key row no longer describes backup
media; the evidence archive row no longer enumerates contents. Stated
as an explicit rule in the maintenance section.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
2026-07-19 16:56:47 +02:00
0f094066f4 ESTATE.md: the one-page estate map (operator order 2026-07-19)
The endeavour spans ~12 active repos, a generated mirror, a droplet
with three containers, a full-account Forgejo mirror, operator-held
state, and two self-referential loops — enough that oversight was
genuinely lost. ESTATE.md is the canonical committed map: five lanes of
custody (upstream inputs -> verified subjects -> machinery &
operator-held -> published faces -> consumers) as a GitHub-rendered
Mermaid diagram, plus the two loops explained first (dogfood signer;
entry-13 self-attestation), a repository inventory with mutability
classes (frozen / generated / free / operator-only), the services and
operator-held table, an edge glossary, and maintenance triggers.

Lives HERE because pacta is the machinery hub and the only repo that
changes freely; the mirror is generated, the subjects are frozen, and
the book is independent. Linked from README and llms.txt. An
interactive operator's version of the same model exists outside the
repos.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
2026-07-19 16:51:20 +02:00
2587cc018b publish assets: sync the fail-open time bomb; llms/course/test docs refreshed (doc audit 2026-07-19)
REAL DEFECT found by the operator-ordered doc-freshness audit:
published_assets.py still carried the PRE-HARDENING fail-open verify.py
and the pre-Tier-2 README as the templates that log-publish drops into
the mirror — the next publish would have silently overwritten the
round-13-hardened fail-closed verifier and the corrected README with
the old versions. Fixed:
- published_assets.py regenerated from the canonical mirror files
  (byte-identity verified by round-trip exec), now also carrying
  verify_selftest.py; SYNC RULE documented in the module docstring.
- transparency_log.publish() now writes verify_selftest.py too.
- NEW tests/test_published_assets.py pins the security-critical markers
  (fail-closed FATAL, RECEIPT_TYPE, verify_receipt, --all receipt
  coverage, required fingerprint) so template drift fails CI instead of
  shipping.
- test_web_and_witness updated to the hardened verifier's markers —
  the published test log now passes FULL signature mode end to end
  ('RESULT: OK [full]'), a stronger assertion than the old string.

Doc refresh in the same pass:
- llms.txt: thirteen leaves + entry-13 self-attestation + fail-closed
  verifier; paper line -> new title, 23 pages, v0.2/v0.1 archives.
- Course (generator + generated 06b notebook): 'the git hash IS the
  content hash' -> 'the commit pins the exact source tree';
  'irrevocably part of the log every other agent sees' -> 'committed to
  the log's signed view, which any agent can compare' (the two Tier-2
  scope corrections had never reached the teaching material).
- test_paper_verifiers.py docstring rescoped: its 164k counts are the
  archived v0.2 report's citation; the current paper cites the corpus
  harness and makes no extensional-equality claim.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
2026-07-19 13:10:55 +02:00
15 changed files with 1176 additions and 56 deletions

157
ESTATE.md Normal file
View file

@ -0,0 +1,157 @@
# The estate map
One page holding the entire mental model of the Lean Transparency Log
endeavour: every repository, service, mirror, and operator-held entity,
and — most importantly — the *typed* relationships between them,
including the two self-referential loops that make the estate hard to
keep in one head. Maintained here in pacta because pacta is the
machinery hub and the only repo that changes freely.
State snapshot (2026-07-19): log **13 leaves**, root `3488a2d0…`, key
fingerprint `874c8a00…`, paper **v0.9 camera-ready (23 pp)**, five
attested components, pacta suite 118 green.
```mermaid
flowchart LR
subgraph S["UPSTREAM INPUTS (frozen clones)"]
s1["curve25519-dalek-source"]
s2["anza-cryptography-source"]
s3["risc0-…-dalek-source"]
s4["betrusted-…-dalek-source"]
s5["pasta_curves-source"]
end
subgraph V["VERIFIED SUBJECTS"]
d["dalek-ed25519-verified<br/>16 certs · leaf 8 · signer source"]
a["anza-ed25519-verified<br/>16 certs · leaf 9"]
r["risc0-ed25519-verified<br/>16 certs · leaf 10"]
b["betrusted-ed25519-verified<br/>16 certs · leaf 11"]
p["pasta-pallas-verified<br/>field layer only · NOT attested"]
c["ltl-accumulator-verified<br/>61 certs · entry-13 subject · frozen 172a1d0"]
end
subgraph M["MACHINERY — pacta + operator-held"]
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 · R0R5"]
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)"]
ops["operational log state<br/>the true accumulator"]
sd["evidence archive (offline)<br/>kits · stamps"]
end
subgraph P["PUBLISHED FACES"]
mir["lean-transparency-log<br/>GENERATED mirror · fail-closed verify.py + selftest"]
site["ltl.zkdefi.org<br/>homepage · /v1 API · /paper"]
fj["Forgejo (droplet)<br/>nightly full-account mirror"]
book["verifying-crypto-with-lean<br/>undergrad book (independent)"]
end
subgraph C2["CONSUMERS"]
cl["offline cloner<br/>verify.py --all"]
wr["warden (runtime)"]
ag["agents<br/>MCP · custody card"]
sw["swisspost-evoting-go-poc<br/>prospective, family-level only"]
rev["external reviewers<br/>GPT-5.6 + Claude"]
end
s1 --> d
s2 --> a
s3 --> r
s4 --> b
s5 --> p
d -->|attest| prov
a -->|attest| prov
r -->|attest| prov
b -->|attest| prov
prov -->|append| ops
key -->|signs heads| ops
ops -->|publish| mir
prov -.->|"templates (CI-pinned)"| mir
prov -->|app code| site
mir -->|published copy| site
pap -->|/paper| site
mir -.->|nightly| fj
mir -->|clone + verify| cl
site -->|API · custody card| ag
mir -->|receipts · quorum| wr
site -.->|prospective| sw
sd -->|review kits| rev
d ==>|"LOOP 1: built from"| sig
sig ==>|"LOOP 1: signs the log"| ops
mir ==>|"LOOP 1: contains the signer's own attestation (leaf 8)"| d
c ==>|"LOOP 2: attested as entry 13"| prov
mir ==>|"LOOP 2: carries proofs about its own accumulator"| c
classDef src fill:#f1f3f5,stroke:#8a93a0,color:#1c2430
classDef sub fill:#e2f2e9,stroke:#1e7f4f,color:#1c2430
classDef mach fill:#eef0f7,stroke:#3b4d8f,color:#1c2430
classDef held fill:#2b3442,stroke:#2b3442,color:#e8ecf2
classDef pub fill:#efe9f5,stroke:#6d4a8f,color:#1c2430
classDef cons fill:#fdf0da,stroke:#a86a10,color:#1c2430
class s1,s2,s3,s4,s5 src
class d,a,r,b,p,c sub
class prov,sig,lib,wal,pap,crs mach
class key,ops,sd held
class mir,site,fj,book pub
class cl,wr,ag,sw,rev cons
```
## The two loops (read these first)
**Loop 1 — the dogfood signer.** The log's tree heads are signed by
`verified-dalek-serial`, a binary built from `dalek-ed25519-verified`
whose own attestation is leaf 8 *inside the log it signs*. Before
signing, the provider re-checks inclusion of the signer's leaf. The
signature vouches for the tree; the tree contains the proofs of the
signer's source. (Execution provenance is reported, not proven — the
paper says so explicitly.)
**Loop 2 — the self-attestation.** `ltl-accumulator-verified` is a Lean
corpus proving soundness of the log's own accumulator *model*
(extractors, consistency binding, per-step pin safety). It was attested
into the log as **entry 13** — the log carries kernel-checked proofs
about its own machinery, scoped honestly (recursive model, not the
deployed verifier; see the corpus KNOWN-GAPS ledger).
## Repository inventory
| Repository | Lane | Role | Mutability |
|---|---|---|---|
| `curve25519-dalek-source`, `anza-cryptography-source`, `risc0-…-source`, `betrusted-…-source`, `pasta_curves-source` (+ `xous-core`, `litex-boards` context) | upstream | pinned inputs to extraction | **frozen — never modified** |
| `dalek-` / `anza-` / `risc0-` / `betrusted-ed25519-verified` | subject | Rust source + Lean proofs; 16 certs each; attested (leaves 811, generations at 07) | 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 (+ 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 |
## Services, infra, operator-held
| Entity | What it is |
|---|---|
| **ltl.zkdefi.org** | droplet (caddy → docker `cloud-ltl-1`): homepage rendered from live leaves, `/v1` API, `/paper` (+`/v0.2`, `/v0.1`), key endpoint. Read-only; no key material on the server. Deployment configuration is maintained privately. |
| **Forgejo** (`cloud-forgejo-1`) | nightly (03:00) mirror of the entire saymrwulf GitHub account — disaster-recovery copy. |
| **Signing key** | offline, operator-only; fingerprint `874c8a00…`; never on the server; public half published in two independent locations. |
| **Operational log state** | `provider/state/transparency-log-main` — the true accumulator. Appends happen here; the mirror is its projection. |
| **Evidence archive (offline)** | review kits and stamped artifacts (`_timestamp_hash8` convention); never in git. |
## Edge glossary
| Edge | Meaning |
|---|---|
| extract | pinned source → Lean model (Aeneas/Charon) |
| attest | subject at pinned commit → provider check → signed leaf |
| append / publish | leaf → operational state → generated mirror |
| templates (CI-pinned) | pacta `published_assets` → mirror's `verify.py`/selftest/README; guarded by `tests/test_published_assets.py` since 2026-07-19 |
| serve | pacta app + mirror copy + paper → droplet → site |
| consume | mirror/site → cloners, warden, agents (receipts recomputed, never trusted) |
## Maintenance
Update this file when: a leaf is appended or a head signed (snapshot
line), the paper version changes, a repo/service/consumer is added or
retired, or a loop-relevant mechanism changes. Rules that keep the map
honest: **generated artifacts are fixed at their source** (mirror files
→ pacta templates); subject repos move only for docs; the three
operator-held entities are never expanded into detail here, and the
private infrastructure layer is deliberately unnamed — this map lists
only entities whose existence is already public or must be public for
trust.

View file

@ -1,5 +1,8 @@
# proof-aware-crypto-tooling-agent
> **Lost in the estate?** [ESTATE.md](ESTATE.md) is the one-page map of
> every repo, service, mirror, and loop, with the edge semantics.
`pacta` is a local CLI prototype for interpreting formal-verification evidence in cryptographic tooling. It is not a trading bot, does not move funds, and does not make financial decisions.
The immediate corpus is the `saymrwulf/*-verified` family of repositories. The shipped Lean files are treated as the verification artifact. This project intentionally does not run Charon, Aeneas, extraction, or Rust-to-Lean regeneration.

View file

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

View file

@ -10,15 +10,17 @@
## Start here
- [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 R0R5 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.
## Live evidence
- Transparency log (RFC 9162): https://ltl.zkdefi.org — signed replay attestations of the Lean proofs, one leaf per fork.
- The paper: https://ltl.zkdefi.org/paper — "The Lean Transparency Log: Distributing Kernel-Checked Correctness Evidence for Deployed Ed25519 Implementations" (revised, with security proofs, 19 pages; prior version at /paper/v0.1).
- Transparency log (RFC 9162): https://ltl.zkdefi.org — signed replay attestations of the Lean proofs. Thirteen leaves: three replay generations over four Ed25519 forks, plus entry 13 attesting the Lean mechanization of the log's own accumulator model. The mirror ships a fail-closed offline verifier (verify.py --all covers every leaf, signed head, and receipt) with an adversarial self-test.
- The paper: https://ltl.zkdefi.org/paper — "Accountable Distribution of Machine-Checked Correctness Evidence: A Transparency Model and the Lean Transparency Log" (23 pages: trust decomposition, scheme-level accountability games with explicit reductions, live deployment, and the measured model/deployment divergence reported as a result). Earlier versions archived at /paper/v0.2 (19 pages) and /paper/v0.1 (4 pages).
## For agents

View file

@ -231,8 +231,8 @@
"After these cells pass, the agent knows: *the provider whose key I\n",
"pinned states that the Lean proofs of repository X at commit Y check\n",
"out with exactly the documented assumptions, and that statement is\n",
"irrevocably part of the log every other agent sees.* The agent then\n",
"clones commit Y (the git hash IS the content hash) and builds it -\n",
"committed to the log's signed view, which any agent can compare.* The agent then\n",
"clones commit Y (the commit pins the exact source tree) and builds it -\n",
"compiler and build remain declared trusted base until R5. Where a\n",
"claim lives (this notebook) and why it is true (the provider's Lean\n",
"replay, lecture 6a) never blur.\n",

View file

@ -1,22 +1,45 @@
"""Static assets dropped into the git-published log repository: a
standalone stdlib-only verifier and the customer README. Kept as string
constants so the published repo is fully self-contained."""
"""Static assets dropped into the git-published log repository: the
standalone fail-closed verifier, its adversarial self-test, and the
customer README. Kept as string constants so the published repo is fully
self-contained.
VERIFY_PY = '''#!/usr/bin/env python3
SYNC RULE: these constants MUST stay byte-identical to the canonical
files in the published mirror (lean-transparency-log: verify.py,
verify_selftest.py, README.md). A publish overwrites the mirror copies
from here, so drift REGRESSES shipped fixes (found 2026-07-19: this file
still carried the pre-hardening fail-open verify.py and the pre-Tier-2
README). tests/test_published_assets.py pins the security-critical
markers; regenerate from the canonical files rather than hand-editing.
"""
VERIFY_PY = r'''#!/usr/bin/env python3
"""Standalone verifier for the published Lean Transparency Log.
Python 3 standard library ONLY - no pacta, no pip. Verifies, from the
Pure Python 3 standard library for hashing and structure; Ed25519
signature checking shells out to the `openssl` binary. Verifies, from the
files in this repository alone:
1. every entry's leaf hash,
2. every historical Signed Tree Head against the recomputed prefix root
(this is the witness check: a split view or tampered entry fails here),
3. every STH Ed25519 signature (via the openssl binary, if available),
4. any receipt's inclusion proof (--receipt FILE).
(a split view or tampered entry fails here),
3. every STH Ed25519 signature,
4. every published receipt under receipts/ (with --all), and any receipt
supplied via --receipt FILE, as a FULL transparency receipt: type tag,
STH signature, REQUIRED key fingerprint, log id, presence of its STH
in the published history, REQUIRED leaf hash matching the named entry,
tree-size agreement, and the inclusion proof. Binding fields are
required, never compare-if-present.
FAIL-CLOSED: if signature checking is unavailable (no `openssl`, or the
public key is missing), the run FAILS signatures are load-bearing and a
"couldn't check" is not a pass. Use --structural-only to explicitly ask
for hashes/structure without signatures (it prints, and exits, as a
reduced check, never as full verification).
Usage:
python3 verify.py --all
python3 verify.py --receipt receipts/dalek-ed25519-verified.receipt.json
python3 verify.py --all --structural-only # explicit reduced check
"""
import argparse
import base64
@ -32,11 +55,11 @@ HERE = Path(__file__).resolve().parent
def leaf_hash(data: bytes) -> bytes:
return hashlib.sha256(b"\\x00" + data).digest()
return hashlib.sha256(b"\x00" + data).digest()
def node_hash(left: bytes, right: bytes) -> bytes:
return hashlib.sha256(b"\\x01" + left + right).digest()
return hashlib.sha256(b"\x01" + left + right).digest()
def merkle_root(leaves):
@ -86,11 +109,21 @@ def load_leaves():
return leaves, problems
def signatures_available() -> bool:
return bool(shutil.which("openssl")) and (HERE / "provider.ed25519.pub").exists()
def key_fingerprint() -> str:
return hashlib.sha256((HERE / "provider.ed25519.pub").read_bytes()).hexdigest()
def check_sth_signature(head) -> str:
"""VALID / INVALID / UNAVAILABLE. UNAVAILABLE is a FAILURE at the
caller unless the run is explicitly --structural-only."""
openssl = shutil.which("openssl")
key = HERE / "provider.ed25519.pub"
if not openssl or not key.exists():
return "skipped (openssl or provider.ed25519.pub missing)"
return "UNAVAILABLE"
signatures = head.get("signatures") or {}
ed = signatures.get("ed25519") or {}
payload = canonical_json({k: v for k, v in head.items() if k != "signatures"})
@ -107,62 +140,271 @@ def check_sth_signature(head) -> str:
return "VALID" if result.returncode == 0 else "INVALID"
RECEIPT_TYPE = "pacta.transparency.receipt.v1"
def verify_receipt(receipt, heads, structural_only: bool, label: str):
"""Full binding checks for one receipt. Every binding field is REQUIRED;
a missing field is a failure, never a skip. Returns failure strings."""
problems = []
if receipt.get("type") != RECEIPT_TYPE:
problems.append(f"{label}: type is {receipt.get('type')!r}, expected {RECEIPT_TYPE!r}")
sth = receipt.get("sth") or {}
if sth.get("hash_algorithm") != "RFC9162_SHA256":
problems.append(f"{label}: STH hash_algorithm is not RFC9162_SHA256")
if receipt.get("hash_algorithm") != "RFC9162_SHA256":
problems.append(f"{label}: receipt hash_algorithm is not RFC9162_SHA256")
if receipt.get("log_id") != sth.get("log_id"):
problems.append(f"{label}: receipt log_id != its STH log_id")
try:
index = int(receipt.get("leaf_index"))
except (TypeError, ValueError):
index = -1
entry_path = HERE / "entries" / f"{index:06d}.json" if index >= 0 else None
if entry_path is None or not entry_path.exists():
problems.append(f"{label}: leaf_index {receipt.get('leaf_index')!r} names no published entry")
return problems
entry = json.loads(entry_path.read_text())
leaf_bytes = canonical_json(entry["leaf"])
# (a) the receipt's STH must be signed by THIS log's key ...
rsig = check_sth_signature(sth)
if rsig == "INVALID" or (rsig == "UNAVAILABLE" and not structural_only):
problems.append(f"{label}: STH signature {rsig}")
# (b) ... the named key fingerprint is REQUIRED and must be this key ...
fp = (sth.get("signatures", {}).get("ed25519", {}) or {}).get("public_key_fingerprint_sha256")
if not fp:
problems.append(f"{label}: STH lacks public_key_fingerprint_sha256 (required)")
elif (HERE / "provider.ed25519.pub").exists() and fp != key_fingerprint():
problems.append(f"{label}: STH signed by a different key than provider.ed25519.pub")
# (c) ... its log_id must be present and match the log ...
meta_path = HERE / "log-metadata.json"
if meta_path.exists():
meta_log_id = json.loads(meta_path.read_text()).get("log_id")
if meta_log_id and sth.get("log_id") != meta_log_id:
problems.append(f"{label}: STH log_id missing or not this log's")
# (d) ... the receipt's STH must appear in the published history ...
if heads and canonical_json(sth) not in {canonical_json(h) for h in heads}:
problems.append(f"{label}: STH not present in sth-history.jsonl")
# (e) ... the leaf_hash is REQUIRED and must match the named entry ...
if not receipt.get("leaf_hash"):
problems.append(f"{label}: leaf_hash missing (required)")
elif receipt["leaf_hash"] != leaf_hash(leaf_bytes).hex():
problems.append(f"{label}: leaf_hash does not match the named entry")
# (f) ... tree_size agreement ...
try:
size_agree = int(receipt.get("tree_size")) == int(sth.get("tree_size"))
except (TypeError, ValueError):
size_agree = False
if not size_agree:
problems.append(f"{label}: tree_size != its STH tree_size")
# (g) ... and finally the inclusion proof itself.
try:
ok = verify_inclusion(
leaf_bytes, index, int(receipt.get("tree_size") or 0),
[bytes.fromhex(h) for h in receipt.get("inclusion_proof") or []],
bytes.fromhex(sth.get("root_hash") or ""),
)
except (TypeError, ValueError):
ok = False
if not ok:
problems.append(f"{label}: inclusion proof INVALID")
print(f"receipt {label}: leaf {index} of {receipt.get('tree_size')} "
f"STH-sig:{rsig} bindings+inclusion:{'OK' if not problems else 'FAIL'}")
return problems
def main() -> int:
parser = argparse.ArgumentParser()
parser.add_argument("--all", action="store_true")
parser.add_argument("--receipt")
parser.add_argument("--structural-only", action="store_true",
help="skip Ed25519 signature checks explicitly; the run reports a "
"REDUCED check and can never print full verification.")
args = parser.parse_args()
sigs_ok = signatures_available()
if not args.structural_only and not sigs_ok:
# Fail closed: a verifier that cannot check signatures must not
# imply it did. Do not silently continue.
print("FATAL: signature checking unavailable (need the `openssl` binary and "
"provider.ed25519.pub). Install openssl / fetch the key, or pass "
"--structural-only to run an explicit hashes-and-structure check.")
return 2
leaves, problems = load_leaves()
print(f"entries: {len(leaves)}")
failures = list(problems)
for problem in problems:
print("PROBLEM:", problem)
history_path = HERE / "sth-history.jsonl"
heads = [json.loads(line) for line in history_path.read_text().splitlines() if line.strip()] if history_path.exists() else []
if args.all or not args.receipt:
history_path = HERE / "sth-history.jsonl"
heads = [json.loads(line) for line in history_path.read_text().splitlines() if line.strip()] if history_path.exists() else []
# log-wide checks (GPT §4.3): history internally consistent AND the
# published latest-sth.json is exactly the final history head.
previous = -1
log_id = None
for position, head in enumerate(heads):
size = int(head["tree_size"])
if size > len(leaves):
failures.append(f"STH #{position} claims size {size} > {len(leaves)} leaves")
expected = merkle_root(leaves[:size]).hex()
structural = "OK" if head["root_hash"] == expected and size >= previous else "MISMATCH"
if structural != "OK":
failures.append(f"STH #{position}")
failures.append(f"STH #{position} prefix-root/monotonicity")
if log_id is None:
log_id = head.get("log_id")
elif head.get("log_id") != log_id:
failures.append(f"STH #{position} log_id changed mid-history")
signature = check_sth_signature(head)
if signature == "INVALID":
failures.append(f"STH #{position} signature")
if signature == "INVALID" or (signature == "UNAVAILABLE" and not args.structural_only):
failures.append(f"STH #{position} signature {signature}")
print(f"STH #{position} size={size} root={head['root_hash'][:16]}… prefix-root:{structural} signature:{signature}")
previous = max(previous, size)
latest_path = HERE / "latest-sth.json"
if latest_path.exists() and heads:
latest = json.loads(latest_path.read_text())
if canonical_json(latest) != canonical_json(heads[-1]):
failures.append("latest-sth.json is not the final sth-history head")
elif int(latest["tree_size"]) != len(leaves):
failures.append(f"latest-sth tree_size {latest['tree_size']} != {len(leaves)} leaves")
else:
print(f"latest-sth: size {latest['tree_size']} == leaf count, and == final history head OK")
for receipt_path in sorted((HERE / "receipts").glob("*.receipt.json")):
failures += verify_receipt(json.loads(receipt_path.read_text()),
heads, args.structural_only, receipt_path.name)
if args.receipt:
receipt = json.loads(Path(args.receipt).read_text())
index = int(receipt["leaf_index"])
entry = json.loads((HERE / "entries" / f"{index:06d}.json").read_text())
ok = verify_inclusion(
canonical_json(entry["leaf"]),
index,
int(receipt["tree_size"]),
[bytes.fromhex(h) for h in receipt["inclusion_proof"]],
bytes.fromhex(receipt["sth"]["root_hash"]),
)
print(f"receipt leaf {index} of {receipt['tree_size']}: inclusion {'VALID' if ok else 'INVALID'}")
if not ok:
failures.append("receipt inclusion")
failures += verify_receipt(json.loads(Path(args.receipt).read_text()),
heads, args.structural_only, Path(args.receipt).name)
print("RESULT:", "OK - the log is internally consistent" if not failures else f"FAILED ({len(failures)} problems)")
return 0 if not failures else 1
mode = "REDUCED (structural only, signatures NOT checked)" if args.structural_only else "full"
if failures:
print(f"RESULT: FAILED ({len(failures)} problems) [{mode}]")
return 1
print(f"RESULT: OK [{mode}]"
+ ("" if not args.structural_only else " — signatures were NOT verified; this is not full verification"))
return 0
if __name__ == "__main__":
sys.exit(main())
'''
README_MD = """# Lean Transparency Log — published mirror
VERIFY_SELFTEST_PY = r'''#!/usr/bin/env python3
"""Adversarial self-test for verify.py — proves the fail-closed paths fail.
Each case mutates a real published receipt (or the environment) and asserts
the verifier REJECTS it; plus the honest controls. Exit 0 only if every case
behaves. Run from a clone: python3 verify_selftest.py
"""
import copy
import json
import os
import subprocess
import sys
import tempfile
from pathlib import Path
HERE = Path(__file__).resolve().parent
def run(*args, env=None):
result = subprocess.run(
[sys.executable, str(HERE / "verify.py"), *args],
capture_output=True, text=True, env=env,
)
return result.returncode, result.stdout
def base_receipt():
path = sorted((HERE / "receipts").glob("*.receipt.json"))[0]
return json.loads(path.read_text())
def mutated(**changes):
receipt = copy.deepcopy(base_receipt())
for dotted, value in changes.items():
target, keys = receipt, dotted.split(".")
for key in keys[:-1]:
target = target[key]
if value is None:
target.pop(keys[-1], None)
else:
target[keys[-1]] = value
return receipt
def check_receipt(receipt) -> int:
with tempfile.NamedTemporaryFile("w", suffix=".json", delete=False) as handle:
json.dump(receipt, handle)
path = handle.name
try:
code, _ = run("--receipt", path)
return code
finally:
os.unlink(path)
def main() -> int:
cases = []
code, out = run("--all")
cases.append(("honest --all passes (full)", code == 0 and "RESULT: OK [full]" in out))
cases.append(("--all covers every published receipt",
out.count("receipt ") == len(list((HERE / "receipts").glob("*.receipt.json")))))
cases.append(("honest receipt passes", check_receipt(base_receipt()) == 0))
cases.append(("missing key fingerprint REJECTED",
check_receipt(mutated(**{"sth.signatures.ed25519.public_key_fingerprint_sha256": None})) == 1))
cases.append(("missing leaf_hash REJECTED", check_receipt(mutated(leaf_hash=None)) == 1))
cases.append(("wrong receipt type REJECTED", check_receipt(mutated(type="forged.v0")) == 1))
cases.append(("forged (unsigned) root REJECTED",
check_receipt(mutated(**{"sth.root_hash": "ff" * 32})) == 1))
cases.append(("tree_size mismatch REJECTED",
check_receipt(mutated(tree_size=int(base_receipt()["tree_size"]) + 1)) == 1))
cases.append(("wrong log_id REJECTED",
check_receipt(mutated(**{"sth.log_id": "00" * 32})) == 1))
code, out = run("--all", "--structural-only")
cases.append(("--structural-only is explicit, never claims full",
code == 0 and "REDUCED" in out and "[full]" not in out))
with tempfile.TemporaryDirectory() as tmp:
os.symlink(sys.executable, Path(tmp) / Path(sys.executable).name)
code, out = run("--all", env={"PATH": tmp})
cases.append(("no openssl -> FAIL CLOSED (exit 2)", code == 2))
width = max(len(name) for name, _ in cases)
for name, ok in cases:
print(f"{'PASS' if ok else 'FAIL'} {name:<{width}}")
if all(ok for _, ok in cases):
print(f"SELFTEST GREEN ({len(cases)} cases)")
return 0
print("SELFTEST RED")
return 1
if __name__ == "__main__":
sys.exit(main())
'''
README_MD = r'''# Lean Transparency Log — published mirror
This repository is the **git-published face** of a transparency log of
formal-verification attestations: signed statements that the Lean 4 proofs
of specific cryptographic Rust libraries, at specific git commits,
re-check with exactly their documented assumptions.
of specific software, at specific git commits, re-check with exactly their
documented assumptions. Its first twelve leaves attest four cryptographic
Rust libraries (Ed25519 implementations); as of **entry 13 (2026-07-16)**
the log also attests **its own accumulator machinery** a kernel-checked
mechanization of the log's security analysis, so the log carries
kernel-checked proofs *about the accumulator model* underlying its own
inclusion and consistency reasoning, as one of its own entries (subject
[`ltl-accumulator-verified`](https://github.com/saymrwulf/ltl-accumulator-verified);
scoped to the mechanized model it does not prove operator honesty,
signing, or execution provenance). Current head: tree size 13, root
`3488a2d0`.
Layout:
@ -173,8 +415,9 @@ Layout:
| `receipts/<component>.receipt.json` | inclusion proof binding that attestation to the latest signed head |
| `sth-history.jsonl` | **every** Signed Tree Head ever issued the witness channel: all cloners see the same heads |
| `latest-sth.json` | the current head |
| `provider.ed25519.pub` | the provider's public key (the sole trust anchor) |
| `verify.py` | standalone verifier, Python standard library only |
| `provider.ed25519.pub` | the provider's public key — the sole cryptographic identity anchor; each statement's truth additionally rests on the assumptions stated in its leaf |
| `verify.py` | standalone verifier (Python stdlib + the `openssl` binary; fails closed without them; `--all` covers every published receipt) |
| `verify_selftest.py` | adversarial self-test: proves the verifier's fail-closed paths reject mutated receipts |
Verify everything locally, no installation:
@ -190,10 +433,11 @@ The provider tooling, agent tooling, and course materials:
**https://github.com/saymrwulf/proof-aware-crypto-tooling-agent**
Honesty notes, always in force: attestations cover Rust **source** at a
pinned commit (clone it the git hash is the content hash and build it
yourself; compilers are declared trusted base). The log deliberately
pinned commit (clone it the commit identifies the committed git tree,
not dependencies or toolchains and build it yourself; compilers are
declared trusted base). The log deliberately
retains early leaves recording a **failed** audit run: an append-only
trust ledger keeps its history. Tree heads are signed by the merkleized,
proof-attested Ed25519 library itself, and each signature embeds the
provider's own Merkle self-check of that library's leaf.
"""
'''

View file

@ -295,9 +295,10 @@ class TransparencyLog:
(out / "entries" / f"{component}.attestation.json").write_text(
json.dumps(entry.leaf.get("attestation"), indent=2, sort_keys=True) + "\n", encoding="utf-8"
)
from .published_assets import README_MD, VERIFY_PY
from .published_assets import README_MD, VERIFY_PY, VERIFY_SELFTEST_PY
(out / "verify.py").write_text(VERIFY_PY, encoding="utf-8")
(out / "verify_selftest.py").write_text(VERIFY_SELFTEST_PY, encoding="utf-8")
(out / "README.md").write_text(README_MD, encoding="utf-8")
if public_key_path is not None:
(out / "provider.ed25519.pub").write_bytes(Path(public_key_path).read_bytes())

View file

@ -1560,8 +1560,8 @@ COURSE = {
After these cells pass, the agent knows: *the provider whose key I
pinned states that the Lean proofs of repository X at commit Y check
out with exactly the documented assumptions, and that statement is
irrevocably part of the log every other agent sees.* The agent then
clones commit Y (the git hash IS the content hash) and builds it -
committed to the log's signed view, which any agent can compare.* The agent then
clones commit Y (the commit pins the exact source tree) and builds it -
compiler and build remain declared trusted base until R5. Where a
claim lives (this notebook) and why it is true (the provider's Lean
replay, lecture 6a) never blur.

View file

@ -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
View 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 &amp; 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

View file

@ -1,11 +1,12 @@
"""The paper (ltl.tex, §5.3, App. B) claims the *recursive* inclusion and
consistency verifiers it defines and proves about are equivalent to the
deployed iterative RFC 9162 verifiers, and cites exhaustive
differential-testing counts. This test IS that verification, so the paper's
numbers cannot silently rot: it reproduces the exact recursive forms
written in the paper and asserts full agreement with the deployed code over
all sizes up to 256, honest inputs plus adversarial mutations, and pins the
cited case counts (164,479 inclusion; 164,224 consistency).
"""Historical regression pin from the archived v0.2 system report
(hosted at /paper/v0.2), which cited these exact differential-testing
counts (164,479 inclusion; 164,224 consistency) for its recursive forms
against the deployed iterative RFC 9162 verifiers over these families.
The CURRENT paper makes no extensional-equality claim: it cites the
accumulator corpus's fidelity harness instead (230,271 / 230,016 honest
families, 73,573 lied-size cases with 3,867 divergences, every one
accepted only by the deployed verifier). This test remains as a pinned
regression boundary for the pacta-internal recursive forms.
"""
import hashlib

View file

@ -0,0 +1,35 @@
"""The publish step overwrites the mirror's verify.py / verify_selftest.py /
README.md from the frozen constants in published_assets. A hardening
regression here silently fail-opens the PUBLIC verifier on the next publish
(exactly what the 2026-07-19 doc audit found: the constants still carried
the pre-hardening fail-open verify.py). These tests pin the
security-critical markers so that drift fails CI instead of shipping."""
from pacta_provider.published_assets import README_MD, VERIFY_PY, VERIFY_SELFTEST_PY
def test_verify_py_compiles_and_is_the_hardened_verifier():
compile(VERIFY_PY, "verify.py", "exec")
for marker in (
"FATAL: signature checking unavailable", # fail-closed exit 2
"RECEIPT_TYPE", # required type binding
"def verify_receipt", # full binding checks
"--structural-only", # explicit reduced mode
'glob("*.receipt.json")', # --all covers every receipt
"public_key_fingerprint_sha256", # required fingerprint
):
assert marker in VERIFY_PY, f"hardening marker missing: {marker}"
def test_selftest_compiles_and_covers_fail_closed():
compile(VERIFY_SELFTEST_PY, "verify_selftest.py", "exec")
for marker in ("FAIL CLOSED", "missing key fingerprint REJECTED",
"forged (unsigned) root REJECTED"):
assert marker in VERIFY_SELFTEST_PY, marker
def test_readme_is_the_corrected_template():
assert "identity anchor" in README_MD
assert "verify_selftest.py" in README_MD
assert "the git hash is the content hash" not in README_MD
assert "sole trust anchor" not in README_MD

197
tests/test_walletui.py Normal file
View 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")

View file

@ -132,6 +132,7 @@ def test_publish_and_witness_audit_catches_tampering(tmp_path):
report = log.publish(published, public_key_path=tmp_path / "k.pub")
assert report["entries"] == 3
assert (published / "verify.py").exists() and (published / "README.md").exists()
assert (published / "verify_selftest.py").exists()
clean = audit_published_log(published, tmp_path / "k.pub")
assert clean.ok and clean.heads_checked == 3
@ -156,4 +157,5 @@ def test_standalone_verify_py_runs(tmp_path):
log.publish(published, public_key_path=tmp_path / "k.pub")
result = subprocess.run([sys.executable, "verify.py", "--all"], cwd=published, capture_output=True, text=True)
assert result.returncode == 0, result.stdout + result.stderr
assert "OK - the log is internally consistent" in result.stdout
# hardened verifier: full mode (signatures verified) must report exactly this
assert "RESULT: OK [full]" in result.stdout