diff --git a/DEPLOY.md b/DEPLOY.md new file mode 100644 index 0000000..1c636bf --- /dev/null +++ b/DEPLOY.md @@ -0,0 +1,125 @@ +# Deploying the online log at zkdefi.org/lean-transparency-log + +Everything below is prepared to run on the DigitalOcean host (the one +running Forgejo). Nothing here needs to run on the development machine — +this file is the checklist for the server session. + +## What gets deployed + +One **read-only** Python process (standard library only, no pip installs) +serving the CT-style API + customer docs. It never touches private keys: +tree heads are signed offline by the provider CLI and only *stored, +already-signed* material is served. A compromised web process can withhold +or replay (agents detect both via pinning + freshness) but cannot forge. + +## 1. Get the code and the log data onto the server + +```bash +sudo useradd --system --home /srv/pacta --create-home pacta +sudo -u pacta git clone https://github.com/saymrwulf/proof-aware-crypto-tooling-agent /srv/pacta/app +# the log STATE (entries + signed heads, no keys) comes from the published mirror: +sudo -u pacta git clone https://github.com/saymrwulf/lean-transparency-log /srv/pacta/published +# reconstruct a servable log dir from the published mirror: +sudo -u pacta mkdir -p /srv/pacta/log +sudo -u pacta python3 - <<'EOF' +import json, pathlib +pub = pathlib.Path("/srv/pacta/published"); log = pathlib.Path("/srv/pacta/log") +(log / "metadata.json").write_text((pub / "log-metadata.json").read_text()) +with (log / "entries.jsonl").open("w") as out: + for p in sorted((pub / "entries").glob("[0-9]*.json")): + r = json.loads(p.read_text()) + out.write(json.dumps({"index": r["index"], "leaf_hash": r["leaf_hash"], "leaf": r["leaf"]}, + sort_keys=True, separators=(",", ":")) + "\n") +(log / "sth-history.jsonl").write_text((pub / "sth-history.jsonl").read_text()) +import shutil; shutil.copy(pub / "latest-sth.json", log / "sth.yaml") +print("log dir reconstructed") +EOF +``` + +(Alternative: rsync `provider/state/transparency-log-main/` from the +provider machine. The published mirror is preferred — it keeps the server +in the same trust position as any other witness.) + +## 2. Systemd unit + +`/etc/systemd/system/pacta-log.service`: + +```ini +[Unit] +Description=Lean Transparency Log (read-only) +After=network.target + +[Service] +User=pacta +WorkingDirectory=/srv/pacta/app +Environment=PYTHONPATH=/srv/pacta/app/src:/srv/pacta/app/provider/src +ExecStart=/usr/bin/python3 -m pacta_provider serve --log-dir /srv/pacta/log --base-path lean-transparency-log --host 127.0.0.1 --port 8461 +Restart=on-failure +# hardening: read-only service, no key material anywhere near it +ProtectSystem=strict +ReadOnlyPaths=/srv/pacta +PrivateTmp=true +NoNewPrivileges=true + +[Install] +WantedBy=multi-user.target +``` + +```bash +sudo systemctl daemon-reload && sudo systemctl enable --now pacta-log +curl -s http://127.0.0.1:8461/lean-transparency-log/healthz +``` + +## 3. Reverse proxy on zkdefi.org + +nginx (add inside the existing zkdefi.org server block, alongside Forgejo): + +```nginx +location /lean-transparency-log/ { + proxy_pass http://127.0.0.1:8461/lean-transparency-log/; + proxy_set_header Host $host; +} +location = /lean-transparency-log { + return 301 /lean-transparency-log/docs; +} +``` + +(Caddy equivalent: `handle_path` not needed — `reverse_proxy 127.0.0.1:8461` +under `route /lean-transparency-log*`.) + +Check: `https://zkdefi.org/lean-transparency-log/docs` renders the customer +documentation; `/v1/sth` returns the dogfood-signed head. + +## 4. Forgejo mirror + +In Forgejo: create migration/mirror of +`https://github.com/saymrwulf/lean-transparency-log` (and optionally the +pacta repo) with periodic sync. The published repo is the witness channel; +having it on BOTH GitHub and Forgejo means witnesses on two independent +hosts — exactly the point. + +## 5. Update cycle (provider machine → world) + +After each new proof-check run on the provider machine: + +```bash +pacta_provider log-append ... # signs new head (offline, dogfood) +pacta_provider log-publish --log-dir ... --git-dir \ + --public-key provider/state/local-provider/provider.ed25519.pub +cd && git add -A && git commit -m "log update" && git push # GitHub + Forgejo sync +# on the server: cd /srv/pacta/published && git pull && re-run step 1's reconstruction +sudo systemctl restart pacta-log +``` + +## 6. Smoke tests from anywhere + +```bash +pacta log-fetch --url https://zkdefi.org/lean-transparency-log --component dalek-ed25519-verified --out-dir /tmp/e +pacta receipt-verify --attestation /tmp/e/dalek-ed25519-verified.attestation.json \ + --receipt /tmp/e/dalek-ed25519-verified.receipt.json \ + --log-public-key \ + --sth-store ~/.pacta-pins.json +pacta sth-refresh --url https://zkdefi.org/lean-transparency-log \ + --sth-store ~/.pacta-pins.json --log-public-key +git clone https://github.com/saymrwulf/lean-transparency-log && cd lean-transparency-log && python3 verify.py --all +``` diff --git a/README.md b/README.md index 19a3b0f..92b36dc 100644 --- a/README.md +++ b/README.md @@ -229,6 +229,29 @@ provider model). The log's first four leaves honestly record a failed audit run (two pacta bugs, fixed and documented); the ledger keeps its history. See `evidence/README.md` to re-verify everything yourself. +## The Online Log and the Published Mirror + +The log has three synchronized faces, transport being orthogonal to trust: + +1. **Files** (`evidence/`): self-contained receipts, verifiable offline. +2. **Git mirror** ([saymrwulf/lean-transparency-log](https://github.com/saymrwulf/lean-transparency-log), mirrored on Forgejo): + every leaf, every signed tree head (the WITNESS CHANNEL - all cloners + see the same heads), per-component receipts, the provider public key, + and a standalone stdlib-only `verify.py`. Anyone: `python3 verify.py --all`. +3. **HTTP service** (deployed at `zkdefi.org/lean-transparency-log`, see + `DEPLOY.md`): read-only CT-style endpoints + customer docs. The web + process never touches private keys - heads are signed offline; a + compromised server can withhold or replay (pinning + freshness detect + both) but never forge. + +```bash +PYTHONPATH=src:provider/src python -m pacta_provider serve --log-dir ... --base-path lean-transparency-log +PYTHONPATH=src:provider/src python -m pacta_provider log-publish --log-dir ... --git-dir --public-key +pacta log-fetch --url https://zkdefi.org/lean-transparency-log --component dalek-ed25519-verified --out-dir fetched +pacta sth-refresh --url https://zkdefi.org/lean-transparency-log --sth-store pins.json --log-public-key +pacta witness-audit --published-dir --log-public-key +``` + ## Split-View Defense (STH Pinning) Standalone receipt verification cannot detect a provider maintaining two log views. `pacta` keeps a local STH pin store: diff --git a/notebooks/06b_agent_verify_inclusion.ipynb b/notebooks/06b_agent_verify_inclusion.ipynb index 7959cf8..be4cf70 100644 --- a/notebooks/06b_agent_verify_inclusion.ipynb +++ b/notebooks/06b_agent_verify_inclusion.ipynb @@ -209,6 +209,23 @@ "cell_type": "markdown", "metadata": {}, "source": [ + "## Transports: files, git, and the online service\n", + "\n", + "Everything you verified above arrived as FILES - and that is a\n", + "feature: receipts are self-contained, so transport never carries\n", + "trust. The same log has two more faces. The GIT MIRROR\n", + "(`lean-transparency-log` on GitHub and on the provider's Forgejo)\n", + "publishes every leaf and every signed head - all cloners see the\n", + "same heads, which makes every cloner a WITNESS\n", + "(`pacta witness-audit` recomputes every prefix root from the\n", + "published leaves; run `python3 verify.py --all` in a clone for the\n", + "no-install version). The ONLINE SERVICE\n", + "(`zkdefi.org/lean-transparency-log`) adds live endpoints: fetch\n", + "fresh evidence (`pacta log-fetch`), advance your pin with an\n", + "online consistency proof (`pacta sth-refresh`). The verification\n", + "you do afterwards is IDENTICAL in all three transports - this\n", + "notebook's ~25 lines never change.\n", + "\n", "## Convinced - of what, exactly?\n", "\n", "After these cells pass, the agent knows: *the provider whose key I\n", diff --git a/provider/src/pacta_provider/cli.py b/provider/src/pacta_provider/cli.py index 0853c24..df7474b 100644 --- a/provider/src/pacta_provider/cli.py +++ b/provider/src/pacta_provider/cli.py @@ -64,6 +64,19 @@ def build_parser() -> argparse.ArgumentParser: log_append.add_argument("--out", required=True) log_append.set_defaults(func=cmd_log_append) + log_publish = sub.add_parser("log-publish", help="Export the log's public face into a git-publishable directory (entries, STH history, receipts).") + log_publish.add_argument("--log-dir", required=True) + log_publish.add_argument("--git-dir", required=True) + log_publish.add_argument("--public-key", help="Provider public key to include in the published repo.") + log_publish.set_defaults(func=cmd_log_publish) + + serve = sub.add_parser("serve", help="Serve the log read-only over HTTP (CT-style endpoints + customer docs). Never touches private keys.") + serve.add_argument("--log-dir", required=True) + serve.add_argument("--base-path", default="lean-transparency-log") + serve.add_argument("--host", default="127.0.0.1") + serve.add_argument("--port", type=int, default=8461) + serve.set_defaults(func=cmd_serve) + log_consistency = sub.add_parser("log-consistency", help="Emit a consistency proof from an earlier tree size (for pinning agents).") log_consistency.add_argument("--log-dir", required=True) log_consistency.add_argument("--from-size", type=int, required=True) @@ -141,6 +154,26 @@ def cmd_log_append(args: argparse.Namespace) -> int: return 0 +def cmd_log_publish(args) -> int: + log = TransparencyLog(args.log_dir) + report = log.publish(args.git_dir, public_key_path=args.public_key) + print(f"published {report['entries']} entries, components: {', '.join(report['components'])}") + print(f"out: {report['out']}") + return 0 + + +def cmd_serve(args) -> int: + from .web import serve as make_server + + server = make_server(args.log_dir, base_path=args.base_path, host=args.host, port=args.port) + print(f"serving read-only log on http://{args.host}:{args.port}/{args.base_path.strip('/')}/docs") + try: + server.serve_forever() + except KeyboardInterrupt: + pass + return 0 + + def cmd_log_consistency(args) -> int: from pacta.yamlio import dump_data diff --git a/provider/src/pacta_provider/published_assets.py b/provider/src/pacta_provider/published_assets.py new file mode 100644 index 0000000..8f16aab --- /dev/null +++ b/provider/src/pacta_provider/published_assets.py @@ -0,0 +1,199 @@ +"""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.""" + +VERIFY_PY = '''#!/usr/bin/env python3 +"""Standalone verifier for the published Lean Transparency Log. + +Python 3 standard library ONLY - no pacta, no pip. 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). + +Usage: + python3 verify.py --all + python3 verify.py --receipt receipts/dalek-ed25519-verified.receipt.json +""" +import argparse +import base64 +import hashlib +import json +import shutil +import subprocess +import sys +import tempfile +from pathlib import Path + +HERE = Path(__file__).resolve().parent + + +def leaf_hash(data: bytes) -> bytes: + return hashlib.sha256(b"\\x00" + data).digest() + + +def node_hash(left: bytes, right: bytes) -> bytes: + return hashlib.sha256(b"\\x01" + left + right).digest() + + +def merkle_root(leaves): + if not leaves: + return hashlib.sha256(b"").digest() + if len(leaves) == 1: + return leaf_hash(leaves[0]) + split = 1 << ((len(leaves) - 1).bit_length() - 1) + return node_hash(merkle_root(leaves[:split]), merkle_root(leaves[split:])) + + +def verify_inclusion(leaf: bytes, index: int, size: int, proof, root: bytes) -> bool: + if index >= size: + return False + fn, sn = index, size - 1 + node = leaf_hash(leaf) + for sibling in proof: + if sn == 0: + return False + if fn % 2 == 1 or fn == sn: + node = node_hash(sibling, node) + if fn % 2 == 0: + while fn % 2 == 0 and fn != 0: + fn //= 2 + sn //= 2 + else: + node = node_hash(node, sibling) + fn //= 2 + sn //= 2 + return sn == 0 and node == root + + +def canonical_json(document) -> bytes: + return json.dumps(document, sort_keys=True, separators=(",", ":"), ensure_ascii=False).encode("utf-8") + + +def load_leaves(): + leaves, problems = [], [] + for position, path in enumerate(sorted((HERE / "entries").glob("[0-9]*.json"))): + record = json.loads(path.read_text()) + data = canonical_json(record["leaf"]) + if record.get("index") != position: + problems.append(f"{path.name}: index {record.get('index')} at position {position}") + if leaf_hash(data).hex() != record.get("leaf_hash"): + problems.append(f"{path.name}: leaf_hash mismatch (tampered entry)") + leaves.append(data) + return leaves, problems + + +def check_sth_signature(head) -> str: + openssl = shutil.which("openssl") + key = HERE / "provider.ed25519.pub" + if not openssl or not key.exists(): + return "skipped (openssl or provider.ed25519.pub missing)" + signatures = head.get("signatures") or {} + ed = signatures.get("ed25519") or {} + payload = canonical_json({k: v for k, v in head.items() if k != "signatures"}) + with tempfile.TemporaryDirectory() as tmp: + payload_path = Path(tmp) / "p" + signature_path = Path(tmp) / "s" + payload_path.write_bytes(payload) + signature_path.write_bytes(base64.b64decode(ed.get("signature_base64", ""))) + result = subprocess.run( + [openssl, "pkeyutl", "-verify", "-pubin", "-inkey", str(key), "-rawin", + "-in", str(payload_path), "-sigfile", str(signature_path)], + capture_output=True, + ) + return "VALID" if result.returncode == 0 else "INVALID" + + +def main() -> int: + parser = argparse.ArgumentParser() + parser.add_argument("--all", action="store_true") + parser.add_argument("--receipt") + args = parser.parse_args() + leaves, problems = load_leaves() + print(f"entries: {len(leaves)}") + failures = list(problems) + for problem in problems: + print("PROBLEM:", problem) + + 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 [] + previous = -1 + for position, head in enumerate(heads): + size = int(head["tree_size"]) + 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}") + signature = check_sth_signature(head) + if signature == "INVALID": + failures.append(f"STH #{position} signature") + print(f"STH #{position} size={size} root={head['root_hash'][:16]}… prefix-root:{structural} signature:{signature}") + previous = max(previous, size) + + 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") + + print("RESULT:", "OK - the log is internally consistent" if not failures else f"FAILED ({len(failures)} problems)") + return 0 if not failures else 1 + + +if __name__ == "__main__": + sys.exit(main()) +''' + +README_MD = """# 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. + +Layout: + +| Path | Content | +|---|---| +| `entries/NNNNNN.json` | one log leaf per file, append-only (git history mirrors log history) | +| `entries/.attestation.json` | the newest attestation per library, for convenience | +| `receipts/.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 | + +Verify everything locally, no installation: + +```bash +python3 verify.py --all +python3 verify.py --receipt receipts/dalek-ed25519-verified.receipt.json +``` + +The online service (same data, live endpoints + customer documentation): +**https://zkdefi.org/lean-transparency-log** + +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 +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. +""" diff --git a/provider/src/pacta_provider/transparency_log.py b/provider/src/pacta_provider/transparency_log.py index f9932ab..5c770b7 100644 --- a/provider/src/pacta_provider/transparency_log.py +++ b/provider/src/pacta_provider/transparency_log.py @@ -41,6 +41,7 @@ class TransparencyLog: self.metadata_path = self.log_dir / "metadata.json" self.entries_path = self.log_dir / "entries.jsonl" self.sth_path = self.log_dir / "sth.yaml" + self.sth_history_path = self.log_dir / "sth-history.jsonl" def init(self, provider: str, public_key_path: str | Path) -> dict[str, Any]: if self.metadata_path.exists() or self.entries_path.exists(): @@ -109,6 +110,7 @@ class TransparencyLog: signing_provenance=self.signing_provenance(entries), ) dump_data(sth, self.sth_path) + self._record_sth(sth) return sth def append_attestation( @@ -151,6 +153,7 @@ class TransparencyLog: signing_provenance=self.signing_provenance(entries), ) dump_data(sth, self.sth_path) + self._record_sth(sth) consistency = [] if appended and previous_size > 0: consistency = proof_to_hex(consistency_proof(leaves, previous_size)) @@ -218,6 +221,88 @@ class TransparencyLog: provenance["signing_library_certificates_proven"] = f"{clean}/{len(attested)}" return provenance + def _record_sth(self, sth: dict[str, Any]) -> None: + with self.sth_history_path.open("a", encoding="utf-8") as handle: + handle.write(json.dumps(sth, sort_keys=True, separators=(",", ":")) + "\n") + + def sth_history(self) -> list[dict[str, Any]]: + if not self.sth_history_path.exists(): + return [] + return [ + json.loads(line) + for line in self.sth_history_path.read_text(encoding="utf-8").splitlines() + if line.strip() + ] + + def proof_for_leaf_hash(self, leaf_hash_hex: str) -> dict[str, Any] | None: + """Read-only inclusion proof against the CURRENT tree for an existing + leaf - what the online service returns. Never signs anything.""" + entries = self.entries() + match = next((entry for entry in entries if entry.leaf_hash == leaf_hash_hex), None) + if match is None: + return None + leaves = [entry.leaf_bytes() for entry in entries] + stored_sth = load_data(self.sth_path) if self.sth_path.exists() else None + return { + "schema_version": 1, + "type": RECEIPT_TYPE, + "log_id": self.metadata()["log_id"], + "hash_algorithm": HASH_ALGORITHM, + "leaf_index": match.index, + "leaf_hash": match.leaf_hash, + "tree_size": len(leaves), + "inclusion_proof": proof_to_hex(inclusion_proof(leaves, match.index)), + "sth": stored_sth, + } + + def newest_entry_for_component(self, component: str) -> LogEntry | None: + matches = [ + entry + for entry in self.entries() + if ((entry.leaf.get("attestation") or {}).get("subject") or {}).get("component") == component + ] + return matches[-1] if matches else None + + def publish(self, git_dir: str | Path, public_key_path: str | Path | None = None) -> dict[str, Any]: + """Export the PUBLIC face of the log into a git-publishable directory: + metadata, one file per leaf (append-only in git history too), the + full STH history (the witness channel: everyone who clones sees the + same heads), the latest head, and per-component convenience + receipts. Private keys never appear here.""" + out = Path(git_dir) + (out / "entries").mkdir(parents=True, exist_ok=True) + (out / "receipts").mkdir(parents=True, exist_ok=True) + metadata = self.metadata() + (out / "log-metadata.json").write_text(json.dumps(metadata, indent=2, sort_keys=True) + "\n", encoding="utf-8") + entries = self.entries() + for entry in entries: + path = out / "entries" / f"{entry.index:06d}.json" + if not path.exists(): + path.write_text(json.dumps({"index": entry.index, "leaf_hash": entry.leaf_hash, "leaf": entry.leaf}, indent=2, sort_keys=True) + "\n", encoding="utf-8") + if self.sth_history_path.exists(): + (out / "sth-history.jsonl").write_text(self.sth_history_path.read_text(encoding="utf-8"), encoding="utf-8") + if self.sth_path.exists(): + latest = load_data(self.sth_path) + (out / "latest-sth.json").write_text(json.dumps(latest, indent=2, sort_keys=True) + "\n", encoding="utf-8") + components = {} + for entry in entries: + component = ((entry.leaf.get("attestation") or {}).get("subject") or {}).get("component") + if component: + components[component] = entry + for component, entry in components.items(): + receipt = self.proof_for_leaf_hash(entry.leaf_hash) + (out / "receipts" / f"{component}.receipt.json").write_text(json.dumps(receipt, indent=2, sort_keys=True) + "\n", encoding="utf-8") + (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 + + (out / "verify.py").write_text(VERIFY_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()) + return {"entries": len(entries), "components": sorted(components), "out": str(out)} + def consistency_from(self, old_tree_size: int) -> dict[str, Any]: """Consistency proof from an arbitrary earlier tree size - what a pinning agent requests when its pin is older than the receipt's diff --git a/provider/src/pacta_provider/web.py b/provider/src/pacta_provider/web.py new file mode 100644 index 0000000..5f5516a --- /dev/null +++ b/provider/src/pacta_provider/web.py @@ -0,0 +1,165 @@ +"""The online face of the transparency log: a READ-ONLY, zero-dependency +HTTP service exposing CT-style endpoints under a base path (deployed at +zkdefi.org/lean-transparency-log behind a reverse proxy). + +Security posture: this process never loads a private key. Tree heads are +signed OFFLINE by the provider CLI (log-append / log-sth); the service +serves stored, already-signed material. Compromise of the web process can +therefore withhold or replay data (which agents detect via pinning and +freshness policies) but can never forge a signature. +""" +from __future__ import annotations + +import json +from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer +from typing import Any +from urllib.parse import parse_qs, urlparse + +from .transparency_log import TransparencyLog + +API_VERSION = "v1" + + +def make_handler(log: TransparencyLog, base_path: str, docs_html: str): + base = "/" + base_path.strip("/") if base_path.strip("/") else "" + + class Handler(BaseHTTPRequestHandler): + server_version = "pacta-log/1" + + def do_GET(self) -> None: # noqa: N802 (stdlib API) + try: + self._route() + except Exception as exc: # noqa: BLE001 - the service must not die on a bad request + self._send(500, {"error": f"internal error: {type(exc).__name__}"}) + + def _route(self) -> None: + parsed = urlparse(self.path) + path = parsed.path.rstrip("/") + query = {key: values[0] for key, values in parse_qs(parsed.query).items()} + if not path.startswith(base): + self._send(404, {"error": "unknown path", "base_path": base or "/"}) + return + route = path[len(base):] or "/" + + if route in ("/", "/docs"): + self._send_html(docs_html) + elif route == "/healthz": + self._send(200, {"ok": True, "tree_size": len(log.entries())}) + elif route == f"/{API_VERSION}/metadata": + self._send(200, log.metadata()) + elif route == f"/{API_VERSION}/sth": + history = log.sth_history() + if history: + self._send(200, history[-1]) + else: + from pacta.yamlio import load_data + + if log.sth_path.exists(): + self._send(200, load_data(log.sth_path)) + else: + self._send(404, {"error": "no signed tree head yet"}) + elif route == f"/{API_VERSION}/sth-history": + self._send(200, {"sth_history": log.sth_history()}) + elif route == f"/{API_VERSION}/sth-consistency": + first = int(query.get("first", "-1")) + if first < 0: + self._send(400, {"error": "missing or invalid ?first="}) + return + try: + self._send(200, log.consistency_from(first)) + except ValueError as exc: + self._send(400, {"error": str(exc)}) + elif route == f"/{API_VERSION}/proof": + leaf_hash = query.get("leaf_hash") + component = query.get("component") + if component and not leaf_hash: + entry = log.newest_entry_for_component(component) + leaf_hash = entry.leaf_hash if entry else None + if not leaf_hash: + self._send(400, {"error": "supply ?leaf_hash= or ?component="}) + return + proof = log.proof_for_leaf_hash(leaf_hash) + if proof is None: + self._send(404, {"error": f"no leaf with hash {leaf_hash}"}) + return + self._send(200, proof) + elif route == f"/{API_VERSION}/attestation": + component = query.get("component") + if not component: + self._send(400, {"error": "supply ?component="}) + return + entry = log.newest_entry_for_component(component) + if entry is None: + self._send(404, {"error": f"no attestation for component {component}"}) + return + self._send(200, { + "leaf_index": entry.index, + "leaf_hash": entry.leaf_hash, + "attestation": entry.leaf.get("attestation"), + }) + elif route == f"/{API_VERSION}/entries": + start = int(query.get("start", "0")) + end = int(query.get("end", str(len(log.entries())))) + entries = log.entries()[start:end] + self._send(200, { + "entries": [ + {"index": entry.index, "leaf_hash": entry.leaf_hash, "leaf": entry.leaf} + for entry in entries + ] + }) + else: + self._send(404, { + "error": "unknown endpoint", + "endpoints": [ + f"{base}/docs", + f"{base}/healthz", + f"{base}/{API_VERSION}/metadata", + f"{base}/{API_VERSION}/sth", + f"{base}/{API_VERSION}/sth-history", + f"{base}/{API_VERSION}/sth-consistency?first=N", + f"{base}/{API_VERSION}/proof?component=NAME | ?leaf_hash=HEX", + f"{base}/{API_VERSION}/attestation?component=NAME", + f"{base}/{API_VERSION}/entries?start=N&end=M", + ], + }) + + def _send(self, code: int, payload: dict[str, Any], code_if_error: int | None = None) -> None: + if code_if_error and "error" in payload: + code = code_if_error + body = json.dumps(payload, indent=2, sort_keys=True).encode("utf-8") + self.send_response(code) + self.send_header("Content-Type", "application/json; charset=utf-8") + self.send_header("Content-Length", str(len(body))) + self.send_header("Cache-Control", "no-store") + self.end_headers() + self.wfile.write(body) + + def _send_html(self, html: str) -> None: + body = html.encode("utf-8") + self.send_response(200) + self.send_header("Content-Type", "text/html; charset=utf-8") + self.send_header("Content-Length", str(len(body))) + self.end_headers() + self.wfile.write(body) + + def log_message(self, fmt: str, *args: Any) -> None: # quiet by default + pass + + return Handler + + +def serve( + log_dir: str, + base_path: str = "lean-transparency-log", + host: str = "127.0.0.1", + port: int = 8461, + docs_html: str | None = None, +) -> ThreadingHTTPServer: + log = TransparencyLog(log_dir) + log.metadata() # fail fast if the log is not initialized + if docs_html is None: + from .webdocs import render_docs + + docs_html = render_docs(log, base_path) + handler = make_handler(log, base_path, docs_html) + return ThreadingHTTPServer((host, port), handler) diff --git a/provider/src/pacta_provider/webdocs.py b/provider/src/pacta_provider/webdocs.py new file mode 100644 index 0000000..4f3c516 --- /dev/null +++ b/provider/src/pacta_provider/webdocs.py @@ -0,0 +1,103 @@ +"""Customer documentation served at the log's base path — self-contained +HTML, no external assets (the service must work air-gapped behind any +reverse proxy).""" +from __future__ import annotations + +from html import escape + +from .transparency_log import TransparencyLog + +_STYLE = """ + body{font-family:system-ui,sans-serif;max-width:60rem;margin:2rem auto;padding:0 1rem; + color:#1c2430;line-height:1.55;background:#f8f9fa} + h1{font-size:1.6rem} h2{font-size:1.15rem;margin-top:2rem} + code,pre{font-family:ui-monospace,Menlo,Consolas,monospace;background:#eef0f3;border-radius:4px} + code{padding:.1rem .3rem} pre{padding:.8rem;overflow-x:auto} + table{border-collapse:collapse;width:100%;font-size:.92rem} + td,th{border-bottom:1px solid #dde2e9;padding:.4rem .6rem;text-align:left;vertical-align:top} + .pill{display:inline-block;background:#e2f2e9;color:#1e7f4f;border-radius:9px; + padding:.05rem .55rem;font-size:.8rem;font-weight:600} + .muted{color:#5a6675;font-size:.9rem} +""" + + +def render_docs(log: TransparencyLog, base_path: str) -> str: + base = "/" + base_path.strip("/") + metadata = log.metadata() + history = log.sth_history() + latest = history[-1] if history else {} + entries = log.entries() + components = sorted({ + component + for entry in entries + if (component := ((entry.leaf.get("attestation") or {}).get("subject") or {}).get("component")) + }) + provenance = ((latest.get("signatures") or {}).get("ed25519") or {}).get("signing_provenance") or {} + rows = "".join( + f"{escape(component)}" + f"attestation" + f"inclusion proof" + for component in components + ) + return f""" +Lean Transparency Log +

Lean Transparency Log read-only

+

This service publishes signed attestations of formal (Lean 4) proof +verification for cryptographic Rust libraries, bound into an append-only +RFC 9162-style Merkle log. Customers verify a signature and a +≈{max(1, (latest.get('tree_size') or 1).bit_length())}-hash inclusion proof in milliseconds +— the hours of Lean kernel re-checking happened once, on the provider's side, +under memory-capped guards.

+ +

Current state

+ + + + + + +
log id{escape(str(metadata.get('log_id', ''))[:32])}…
tree size{latest.get('tree_size', 0)} leaves
latest root{escape(str(latest.get('root_hash', ''))[:32])}…
root signed by{escape(str(((latest.get('signatures') or {}).get('ed25519') or {}).get('signing_backend', 'n/a')))} + — the merkleized, proof-attested Ed25519 library itself; before signing, the provider + Merkle-verified that library's own leaf (index {provenance.get('signing_library_leaf_index', '?')}, + certificates {escape(str(provenance.get('signing_library_certificates_proven', '?')))}) against this very tree
attested components{len(components)}
+ +

Attested libraries

+{rows}
componentclaim documentproof of inclusion
+

Each attestation names the exact git commit it covers, every certificate +with its observed axiom cone, and the machine-protection used during replay. The log +also retains earlier leaves that honestly record a failed audit run — an +append-only trust ledger keeps its history.

+ +

API

+
GET {base}/v1/sth                      latest Signed Tree Head
+GET {base}/v1/sth-history              every head ever signed (witness material)
+GET {base}/v1/sth-consistency?first=N  consistency proof from your pinned size
+GET {base}/v1/proof?component=NAME     inclusion proof for the newest attestation
+GET {base}/v1/attestation?component=NAME
+GET {base}/v1/entries?start=N&end=M
+GET {base}/v1/metadata                 log identity
+GET {base}/healthz
+ +

Verify without trusting this site

+

Everything is verifiable offline. Clone the mirror repository (published on GitHub +and on this Forgejo), which contains every leaf, every signed tree head, the provider +public key, and a standalone verify.py (Python standard library only, +≈100 lines). It recomputes the entire tree from the leaves, checks every historical +head against its prefix, verifies the signatures, and checks any inclusion proof:

+
git clone <mirror-url>/lean-transparency-log
+python3 verify.py --all
+

For agents: the pacta tool adds pinning (split-view defense), +freshness policy, and the option to verify signatures through the +proof-attested Ed25519 code path itself +(pacta receipt-verify … --sth-store … --require-verified-verifier).

+ +

What a verified inclusion means — and what it does not

+

Means: the provider whose key you hold attests that the Lean proofs +of the named repository at the named commit re-check with exactly the documented +assumptions, and this attestation is irrevocably part of the log everyone sees.

+

Does not mean: a verified binary. The proofs cover Rust source; +you clone the attested commit (the git hash is the content hash) and build it +yourself — compiler and build remain declared trusted base until the R5 +program (reproducible builds) lands. Every attestation carries the full residual-risk +list; honesty about the boundary is the product.

+""" diff --git a/scripts/build_curriculum_notebooks.py b/scripts/build_curriculum_notebooks.py index 82d1fae..39034b5 100644 --- a/scripts/build_curriculum_notebooks.py +++ b/scripts/build_curriculum_notebooks.py @@ -1538,6 +1538,23 @@ COURSE = { ), md( """ + ## Transports: files, git, and the online service + + Everything you verified above arrived as FILES - and that is a + feature: receipts are self-contained, so transport never carries + trust. The same log has two more faces. The GIT MIRROR + (`lean-transparency-log` on GitHub and on the provider's Forgejo) + publishes every leaf and every signed head - all cloners see the + same heads, which makes every cloner a WITNESS + (`pacta witness-audit` recomputes every prefix root from the + published leaves; run `python3 verify.py --all` in a clone for the + no-install version). The ONLINE SERVICE + (`zkdefi.org/lean-transparency-log`) adds live endpoints: fetch + fresh evidence (`pacta log-fetch`), advance your pin with an + online consistency proof (`pacta sth-refresh`). The verification + you do afterwards is IDENTICAL in all three transports - this + notebook's ~25 lines never change. + ## Convinced - of what, exactly? After these cells pass, the agent knows: *the provider whose key I diff --git a/src/pacta/cli.py b/src/pacta/cli.py index e427f77..cc5d1a6 100644 --- a/src/pacta/cli.py +++ b/src/pacta/cli.py @@ -142,6 +142,23 @@ def build_parser() -> argparse.ArgumentParser: receipt_verify.add_argument("--require-verified-verifier", action="store_true", help="Fail closed unless Ed25519 verification ran on the dogfood (certificate-covered) verifier.") receipt_verify.set_defaults(func=cmd_receipt_verify) + log_fetch = sub.add_parser("log-fetch", help="Fetch attestation + inclusion proof for a component from an ONLINE log; verify locally afterwards.") + log_fetch.add_argument("--url", required=True, help="Base URL, e.g. https://zkdefi.org/lean-transparency-log") + log_fetch.add_argument("--component", required=True) + log_fetch.add_argument("--out-dir", default="fetched-evidence") + log_fetch.set_defaults(func=cmd_log_fetch) + + sth_refresh = sub.add_parser("sth-refresh", help="Fetch the latest STH online, verify signature + consistency from the pinned size, advance the pin.") + sth_refresh.add_argument("--url", required=True) + sth_refresh.add_argument("--sth-store", required=True) + sth_refresh.add_argument("--log-public-key", required=True) + sth_refresh.set_defaults(func=cmd_sth_refresh) + + witness = sub.add_parser("witness-audit", help="Audit a CLONE of the published log repo: recompute every prefix root, check every historical STH + signature.") + witness.add_argument("--published-dir", required=True) + witness.add_argument("--log-public-key") + witness.set_defaults(func=cmd_witness_audit) + dogfood_build = sub.add_parser("dogfood-build", help="Build the dogfood Ed25519 verifier from the pinned proven source workspace.") dogfood_build.add_argument("--source", required=True, help="Local checkout of saymrwulf/curve25519-dalek-source (the pinned proven workspace).") dogfood_build.add_argument("--timeout", type=int, default=900) @@ -421,6 +438,46 @@ def cmd_receipt_verify(args: argparse.Namespace) -> int: return 0 if result.accepted else 1 +def cmd_log_fetch(args: argparse.Namespace) -> int: + from .logclient import LogClientError, fetch_evidence + + try: + paths = fetch_evidence(args.url, args.component, args.out_dir) + except LogClientError as exc: + print(f"error: {exc}") + return 1 + for kind, path in paths.items(): + print(f"{kind}: {path}") + print("fetched material is UNVERIFIED until you run receipt-verify on it (transport is not trust).") + return 0 + + +def cmd_sth_refresh(args: argparse.Namespace) -> int: + from .logclient import LogClientError, refresh_pin + + try: + ok, diagnostics = refresh_pin(args.url, args.sth_store, args.log_public_key) + except LogClientError as exc: + print(f"error: {exc}") + return 1 + for diagnostic in diagnostics: + print(("" if ok else "REFUSED: ") + diagnostic) + return 0 if ok else 1 + + +def cmd_witness_audit(args: argparse.Namespace) -> int: + from .witness import audit_published_log + + report = audit_published_log(args.published_dir, args.log_public_key) + print(f"entries: {report.tree_size} | heads checked: {report.heads_checked}") + for note in report.notes: + print(f"note: {note}") + for problem in report.problems: + print(f"PROBLEM: {problem}") + print(f"ok: {str(report.ok).lower()}") + return 0 if report.ok else 1 + + def cmd_dogfood_build(args: argparse.Namespace) -> int: from .dogfood import build_verifier diff --git a/src/pacta/logclient.py b/src/pacta/logclient.py new file mode 100644 index 0000000..268a3c3 --- /dev/null +++ b/src/pacta/logclient.py @@ -0,0 +1,93 @@ +"""Agent-side client for the online transparency log (zero dependencies). + +Everything fetched here is verified LOCALLY afterwards - the client never +extends trust to the transport: signatures, inclusion proofs, pin-store +consistency, and freshness are all checked by the same code paths used for +file-based evidence. HTTPS certificate handling is the standard library's; +the security of the system does not rest on it (a hostile server can only +withhold or replay, which pinning + freshness detect). +""" +from __future__ import annotations + +import json +import urllib.error +import urllib.parse +import urllib.request +from pathlib import Path +from typing import Any + +DEFAULT_TIMEOUT = 20 + + +class LogClientError(RuntimeError): + pass + + +def _get(base_url: str, path: str, timeout: int = DEFAULT_TIMEOUT) -> dict[str, Any]: + url = base_url.rstrip("/") + path + try: + with urllib.request.urlopen(url, timeout=timeout) as response: + return json.loads(response.read().decode("utf-8")) + except urllib.error.HTTPError as exc: + try: + detail = json.loads(exc.read().decode("utf-8")).get("error", "") + except Exception: # noqa: BLE001 + detail = "" + raise LogClientError(f"{url}: HTTP {exc.code} {detail}".strip()) from exc + except (urllib.error.URLError, TimeoutError, json.JSONDecodeError) as exc: + raise LogClientError(f"{url}: {exc}") from exc + + +def fetch_sth(base_url: str) -> dict[str, Any]: + return _get(base_url, "/v1/sth") + + +def fetch_consistency(base_url: str, first: int) -> dict[str, Any]: + return _get(base_url, f"/v1/sth-consistency?first={int(first)}") + + +def fetch_attestation(base_url: str, component: str) -> dict[str, Any]: + return _get(base_url, f"/v1/attestation?component={urllib.parse.quote(component)}") + + +def fetch_proof(base_url: str, component: str | None = None, leaf_hash: str | None = None) -> dict[str, Any]: + if leaf_hash: + return _get(base_url, f"/v1/proof?leaf_hash={urllib.parse.quote(leaf_hash)}") + if component: + return _get(base_url, f"/v1/proof?component={urllib.parse.quote(component)}") + raise LogClientError("fetch_proof needs component or leaf_hash") + + +def fetch_evidence(base_url: str, component: str, out_dir: str | Path) -> dict[str, Path]: + """Download attestation + inclusion proof for a component into out_dir + (JSON files compatible with every offline pacta flow).""" + out = Path(out_dir) + out.mkdir(parents=True, exist_ok=True) + attestation = fetch_attestation(base_url, component)["attestation"] + proof = fetch_proof(base_url, component=component) + attestation_path = out / f"{component}.attestation.json" + receipt_path = out / f"{component}.receipt.json" + attestation_path.write_text(json.dumps(attestation, indent=2, sort_keys=True) + "\n", encoding="utf-8") + receipt_path.write_text(json.dumps(proof, indent=2, sort_keys=True) + "\n", encoding="utf-8") + return {"attestation": attestation_path, "receipt": receipt_path} + + +def refresh_pin(base_url: str, sth_store_path: str | Path, log_public_key_path: str | Path) -> tuple[bool, list[str]]: + """Fetch the latest STH; verify its signature; advance the local pin + using an online consistency proof from the pinned size. Fail closed on + any mismatch - a hostile or broken log cannot move the pin.""" + from .sthstore import check_sth_against_store, load_store + from .transparency import verify_signed_tree_head + + sth = fetch_sth(base_url) + ok, diagnostics, _statuses = verify_signed_tree_head(sth, log_public_key_path) + if not ok: + return False, ["Fetched STH failed signature verification:"] + diagnostics + store = load_store(sth_store_path) + pinned = store["logs"].get(str(sth.get("log_id") or "")) + proof_hex = None + if pinned is not None and int(sth.get("tree_size", -1)) > int(pinned["tree_size"]): + consistency = fetch_consistency(base_url, int(pinned["tree_size"])) + proof_hex = [str(item) for item in consistency.get("proof") or []] + result = check_sth_against_store(sth, sth_store_path, consistency_proof_hex=proof_hex) + return result.ok, result.diagnostics diff --git a/src/pacta/witness.py b/src/pacta/witness.py new file mode 100644 index 0000000..01449f8 --- /dev/null +++ b/src/pacta/witness.py @@ -0,0 +1,90 @@ +"""Witness audit over the PUBLISHED log repository. + +The git-published log (entries/ + sth-history.jsonl + provider key) turns +every cloner into a witness: since all leaves are public, a witness does +not need consistency proofs at all - it recomputes the root of every +prefix directly and checks every historical Signed Tree Head against its +prefix root and its signature. Two different clones agreeing on the git +history and both passing this audit have PROOF the provider never +equivocated within it. +""" +from __future__ import annotations + +import json +from dataclasses import dataclass, field +from pathlib import Path +from typing import Any + +from .signing import canonical_json +from .transparency import leaf_hash, merkle_root, verify_signed_tree_head + + +@dataclass(slots=True) +class WitnessReport: + ok: bool + tree_size: int + heads_checked: int + problems: list[str] = field(default_factory=list) + notes: list[str] = field(default_factory=list) + + +def audit_published_log( + published_dir: str | Path, + log_public_key_path: str | Path | None = None, +) -> WitnessReport: + root_dir = Path(published_dir) + problems: list[str] = [] + notes: list[str] = [] + + entry_files = sorted((root_dir / "entries").glob("[0-9]*.json")) + leaves: list[bytes] = [] + for position, path in enumerate(entry_files): + record = json.loads(path.read_text(encoding="utf-8")) + if int(record.get("index", -1)) != position: + problems.append(f"{path.name}: index {record.get('index')} at position {position} (gap or reorder).") + leaf_bytes = canonical_json(record["leaf"]) + if leaf_hash(leaf_bytes).hex() != record.get("leaf_hash"): + problems.append(f"{path.name}: leaf_hash does not match the leaf content.") + leaves.append(leaf_bytes) + + history_path = root_dir / "sth-history.jsonl" + heads = [ + json.loads(line) + for line in history_path.read_text(encoding="utf-8").splitlines() + if line.strip() + ] if history_path.exists() else [] + if not heads: + notes.append("No sth-history.jsonl; only the latest head can be checked.") + latest_path = root_dir / "latest-sth.json" + if latest_path.exists(): + heads = [json.loads(latest_path.read_text(encoding="utf-8"))] + + previous_size = -1 + for position, head in enumerate(heads): + size = int(head.get("tree_size", -1)) + if size < previous_size: + problems.append(f"STH #{position}: tree_size {size} SHRANK from {previous_size} (rollback in history).") + previous_size = max(previous_size, size) + if size < 0 or size > len(leaves): + problems.append(f"STH #{position}: tree_size {size} outside published entries ({len(leaves)}).") + continue + expected_root = merkle_root(leaves[:size]).hex() + if head.get("root_hash") != expected_root: + problems.append( + f"STH #{position} (size {size}): signed root {str(head.get('root_hash'))[:16]}… does not match " + f"the recomputed prefix root {expected_root[:16]}… - EQUIVOCATION or tampered entries." + ) + if log_public_key_path is not None: + ok, diagnostics, _statuses = verify_signed_tree_head(head, log_public_key_path) + if not ok: + problems.append(f"STH #{position} (size {size}): signature check failed: {'; '.join(diagnostics)}") + + if log_public_key_path is None: + notes.append("No public key supplied; structural audit only (signatures unchecked).") + return WitnessReport( + ok=not problems, + tree_size=len(leaves), + heads_checked=len(heads), + problems=problems, + notes=notes, + ) diff --git a/tests/test_web_and_witness.py b/tests/test_web_and_witness.py new file mode 100644 index 0000000..9774519 --- /dev/null +++ b/tests/test_web_and_witness.py @@ -0,0 +1,112 @@ +import json +import threading +import urllib.request +from pathlib import Path + +from pacta.signing import generate_ed25519_keypair +from pacta.transparency import leaf_bytes_for_attestation, verify_inclusion +from pacta.witness import audit_published_log +from pacta_provider.transparency_log import TransparencyLog +from pacta_provider.web import serve + + +def _make_log(tmp_path, n=3): + generate_ed25519_keypair(tmp_path / "k.key", tmp_path / "k.pub") + log = TransparencyLog(tmp_path / "log") + log.init("test-provider", tmp_path / "k.pub") + from pacta.yamlio import dump_data + + for i in range(n): + att = { + "schema_version": 1, + "provider": "test-provider", + "issued_at": "2026-07-07T00:00:00Z", + "subject": {"component": f"component-{i}", "repo_commit": f"commit-{i}"}, + "certificates": [{"name": "T.cert", "status": "proven", "axiom_status": "clean"}], + } + dump_data(att, tmp_path / f"a{i}.yaml") + log.append_attestation(tmp_path / f"a{i}.yaml", tmp_path / "k.key", tmp_path / "k.pub") + return log + + +def test_web_endpoints_and_online_proof_roundtrip(tmp_path): + _make_log(tmp_path) + server = serve(str(tmp_path / "log"), port=0) + port = server.server_address[1] + threading.Thread(target=server.serve_forever, daemon=True).start() + base = f"http://127.0.0.1:{port}/lean-transparency-log" + try: + def get(path): + with urllib.request.urlopen(base + path, timeout=10) as r: + return json.loads(r.read()) + + assert get("/healthz")["tree_size"] == 3 + sth = get("/v1/sth") + assert sth["tree_size"] == 3 + att = get("/v1/attestation?component=component-1")["attestation"] + proof = get("/v1/proof?component=component-1") + ok = verify_inclusion( + leaf_bytes_for_attestation(att), proof["leaf_index"], proof["tree_size"], + [bytes.fromhex(h) for h in proof["inclusion_proof"]], + bytes.fromhex(proof["sth"]["root_hash"]), + ) + assert ok + consistency = get("/v1/sth-consistency?first=2") + assert consistency["from_tree_size"] == 2 and consistency["proof"] + history = get("/v1/sth-history")["sth_history"] + assert len(history) == 3 # one head per append + finally: + server.shutdown() + + +def test_logclient_fetch_and_refresh_pin(tmp_path): + _make_log(tmp_path) + server = serve(str(tmp_path / "log"), port=0) + port = server.server_address[1] + threading.Thread(target=server.serve_forever, daemon=True).start() + base = f"http://127.0.0.1:{port}/lean-transparency-log" + try: + from pacta.logclient import fetch_evidence, refresh_pin + + paths = fetch_evidence(base, "component-2", tmp_path / "fetched") + assert paths["attestation"].exists() and paths["receipt"].exists() + ok, diagnostics = refresh_pin(base, tmp_path / "pins.json", tmp_path / "k.pub") + assert ok, diagnostics + # second refresh: matched, still ok + ok, _ = refresh_pin(base, tmp_path / "pins.json", tmp_path / "k.pub") + assert ok + finally: + server.shutdown() + + +def test_publish_and_witness_audit_catches_tampering(tmp_path): + log = _make_log(tmp_path) + published = tmp_path / "published" + 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() + + clean = audit_published_log(published, tmp_path / "k.pub") + assert clean.ok and clean.heads_checked == 3 + + # tamper one entry: structural audit must fail loudly + victim = published / "entries" / "000001.json" + record = json.loads(victim.read_text()) + record["leaf"]["attestation"]["subject"]["repo_commit"] = "EVIL" + victim.write_text(json.dumps(record)) + dirty = audit_published_log(published, tmp_path / "k.pub") + assert not dirty.ok + assert any("leaf_hash" in problem for problem in dirty.problems) + assert any("EQUIVOCATION or tampered" in problem for problem in dirty.problems) + + +def test_standalone_verify_py_runs(tmp_path): + import subprocess + import sys + + log = _make_log(tmp_path) + published = tmp_path / "published" + 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