The log goes public: git-published mirror, online service, witnesses

Three synchronized faces of one log - transport orthogonal to trust:

- PUBLISHED GIT MIRROR: log-publish exports the public face (one file
  per leaf so git history mirrors log history; the FULL STH history as
  the witness channel; per-component attestations + receipts; the
  provider public key; a standalone stdlib-only verify.py and customer
  README). Live at github.com/saymrwulf/lean-transparency-log (genesis:
  8 leaves incl. the honest failed-run entries, dogfood-signed head).
- ONLINE SERVICE (pacta_provider serve): read-only, zero-dependency
  HTTP with CT-style endpoints under a base path for
  zkdefi.org/lean-transparency-log - /v1/sth, /v1/sth-history,
  /v1/sth-consistency?first=N, /v1/proof, /v1/attestation, /v1/entries,
  /v1/metadata, /healthz - plus self-contained customer documentation
  at /docs (current state, attested components, API, the verify-
  without-trusting-this-site path, and the means/does-NOT-mean
  boundary). The process never loads private keys: heads are signed
  offline; a compromised server can withhold or replay (pinning +
  freshness detect both) but never forge. STH history now recorded
  append-only by the provider (with a backfill head signed for the
  existing log).
- AGENT ONLINE CLIENT: pacta log-fetch (download evidence; explicitly
  UNVERIFIED until receipt-verify runs - transport is not trust) and
  pacta sth-refresh (fetch head, verify signature, advance the pin via
  an online consistency proof from the pinned size; fail closed).
- WITNESSES: pacta witness-audit over a clone of the published mirror
  recomputes every prefix root from the public leaves and checks every
  historical head + signature - no consistency proofs needed when the
  leaves are public. Tampering one published entry trips both the
  leaf-hash check and the prefix-root check (tested). verify.py gives
  customers the same audit with zero installation.
- DEPLOY.md: the complete server-session checklist for zkdefi.org -
  reconstruct the servable log FROM the published mirror (the server
  stays in witness trust-position), hardened systemd unit, nginx/Caddy
  path routing, Forgejo mirror setup, the provider->world update
  cycle, and remote smoke tests.

Validated end-to-end on the REAL log: all 10 endpoints, online-fetched
proof re-verified locally through the dogfood verifier with pinning,
online pin refresh, publish + witness audit green, tamper caught,
standalone verify.py green in the published clone. 54/54 tests.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
This commit is contained in:
mrwulf 2026-07-06 16:05:20 +02:00
parent b6382dd0ad
commit fbe40c3dfe
13 changed files with 1119 additions and 0 deletions

125
DEPLOY.md Normal file
View file

@ -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 <clone of lean-transparency-log> \
--public-key provider/state/local-provider/provider.ed25519.pub
cd <clone> && 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 <provider.ed25519.pub from the published repo> \
--sth-store ~/.pacta-pins.json
pacta sth-refresh --url https://zkdefi.org/lean-transparency-log \
--sth-store ~/.pacta-pins.json --log-public-key <pubkey>
git clone https://github.com/saymrwulf/lean-transparency-log && cd lean-transparency-log && python3 verify.py --all
```

View file

@ -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. run (two pacta bugs, fixed and documented); the ledger keeps its history.
See `evidence/README.md` to re-verify everything yourself. 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 <mirror clone> --public-key <pub>
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 <pub>
pacta witness-audit --published-dir <clone of lean-transparency-log> --log-public-key <pub>
```
## Split-View Defense (STH Pinning) ## Split-View Defense (STH Pinning)
Standalone receipt verification cannot detect a provider maintaining two log views. `pacta` keeps a local STH pin store: Standalone receipt verification cannot detect a provider maintaining two log views. `pacta` keeps a local STH pin store:

View file

@ -209,6 +209,23 @@
"cell_type": "markdown", "cell_type": "markdown",
"metadata": {}, "metadata": {},
"source": [ "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", "## Convinced - of what, exactly?\n",
"\n", "\n",
"After these cells pass, the agent knows: *the provider whose key I\n", "After these cells pass, the agent knows: *the provider whose key I\n",

View file

@ -64,6 +64,19 @@ def build_parser() -> argparse.ArgumentParser:
log_append.add_argument("--out", required=True) log_append.add_argument("--out", required=True)
log_append.set_defaults(func=cmd_log_append) 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 = 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("--log-dir", required=True)
log_consistency.add_argument("--from-size", type=int, 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 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: def cmd_log_consistency(args) -> int:
from pacta.yamlio import dump_data from pacta.yamlio import dump_data

View file

@ -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/<component>.attestation.json` | the newest attestation per library, for convenience |
| `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 |
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.
"""

View file

@ -41,6 +41,7 @@ class TransparencyLog:
self.metadata_path = self.log_dir / "metadata.json" self.metadata_path = self.log_dir / "metadata.json"
self.entries_path = self.log_dir / "entries.jsonl" self.entries_path = self.log_dir / "entries.jsonl"
self.sth_path = self.log_dir / "sth.yaml" 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]: def init(self, provider: str, public_key_path: str | Path) -> dict[str, Any]:
if self.metadata_path.exists() or self.entries_path.exists(): if self.metadata_path.exists() or self.entries_path.exists():
@ -109,6 +110,7 @@ class TransparencyLog:
signing_provenance=self.signing_provenance(entries), signing_provenance=self.signing_provenance(entries),
) )
dump_data(sth, self.sth_path) dump_data(sth, self.sth_path)
self._record_sth(sth)
return sth return sth
def append_attestation( def append_attestation(
@ -151,6 +153,7 @@ class TransparencyLog:
signing_provenance=self.signing_provenance(entries), signing_provenance=self.signing_provenance(entries),
) )
dump_data(sth, self.sth_path) dump_data(sth, self.sth_path)
self._record_sth(sth)
consistency = [] consistency = []
if appended and previous_size > 0: if appended and previous_size > 0:
consistency = proof_to_hex(consistency_proof(leaves, previous_size)) consistency = proof_to_hex(consistency_proof(leaves, previous_size))
@ -218,6 +221,88 @@ class TransparencyLog:
provenance["signing_library_certificates_proven"] = f"{clean}/{len(attested)}" provenance["signing_library_certificates_proven"] = f"{clean}/{len(attested)}"
return provenance 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]: def consistency_from(self, old_tree_size: int) -> dict[str, Any]:
"""Consistency proof from an arbitrary earlier tree size - what a """Consistency proof from an arbitrary earlier tree size - what a
pinning agent requests when its pin is older than the receipt's pinning agent requests when its pin is older than the receipt's

View file

@ -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=<old tree size>"})
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=<hex> or ?component=<name>"})
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=<name>"})
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)

View file

@ -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"<tr><td><code>{escape(component)}</code></td>"
f"<td><a href='{base}/v1/attestation?component={escape(component)}'>attestation</a></td>"
f"<td><a href='{base}/v1/proof?component={escape(component)}'>inclusion proof</a></td></tr>"
for component in components
)
return f"""<!doctype html><html><head><meta charset="utf-8">
<title>Lean Transparency Log</title><style>{_STYLE}</style></head><body>
<h1>Lean Transparency Log <span class="pill">read-only</span></h1>
<p>This service publishes <strong>signed attestations of formal (Lean&nbsp;4) proof
verification</strong> for cryptographic Rust libraries, bound into an append-only
RFC&nbsp;9162-style Merkle log. Customers verify a signature and a
&asymp;{max(1, (latest.get('tree_size') or 1).bit_length())}-hash inclusion proof in milliseconds
&mdash; the hours of Lean kernel re-checking happened once, on the provider's side,
under memory-capped guards.</p>
<h2>Current state</h2>
<table>
<tr><th>log id</th><td><code>{escape(str(metadata.get('log_id', ''))[:32])}&hellip;</code></td></tr>
<tr><th>tree size</th><td>{latest.get('tree_size', 0)} leaves</td></tr>
<tr><th>latest root</th><td><code>{escape(str(latest.get('root_hash', ''))[:32])}&hellip;</code></td></tr>
<tr><th>root signed by</th><td><code>{escape(str(((latest.get('signatures') or {}).get('ed25519') or {}).get('signing_backend', 'n/a')))}</code>
&mdash; 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</td></tr>
<tr><th>attested components</th><td>{len(components)}</td></tr>
</table>
<h2>Attested libraries</h2>
<table><tr><th>component</th><th>claim document</th><th>proof of inclusion</th></tr>{rows}</table>
<p class="muted">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 &mdash; an
append-only trust ledger keeps its history.</p>
<h2>API</h2>
<pre>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&amp;end=M
GET {base}/v1/metadata log identity
GET {base}/healthz</pre>
<h2>Verify without trusting this site</h2>
<p>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 <code>verify.py</code> (Python standard library only,
&asymp;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:</p>
<pre>git clone &lt;mirror-url&gt;/lean-transparency-log
python3 verify.py --all</pre>
<p>For agents: the <code>pacta</code> tool adds pinning (split-view defense),
freshness policy, and the option to verify signatures through the
proof-attested Ed25519 code path itself
(<code>pacta receipt-verify &hellip; --sth-store &hellip; --require-verified-verifier</code>).</p>
<h2>What a verified inclusion means &mdash; and what it does not</h2>
<p><strong>Means:</strong> 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.</p>
<p><strong>Does not mean:</strong> a verified binary. The proofs cover Rust source;
you clone the attested commit (the git hash <em>is</em> the content hash) and build it
yourself &mdash; 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.</p>
</body></html>"""

View file

@ -1538,6 +1538,23 @@ COURSE = {
), ),
md( 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? ## Convinced - of what, exactly?
After these cells pass, the agent knows: *the provider whose key I After these cells pass, the agent knows: *the provider whose key I

View file

@ -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.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) 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 = 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("--source", required=True, help="Local checkout of saymrwulf/curve25519-dalek-source (the pinned proven workspace).")
dogfood_build.add_argument("--timeout", type=int, default=900) 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 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: def cmd_dogfood_build(args: argparse.Namespace) -> int:
from .dogfood import build_verifier from .dogfood import build_verifier

93
src/pacta/logclient.py Normal file
View file

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

90
src/pacta/witness.py Normal file
View file

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

View file

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