#!/usr/bin/env python3 """CAIN-42 E13 -- CLEAN-ROOM BPF GOVERNANCE ARTIFACT VERIFIER. ZERO CAIN imports: standard library + `cryptography` only. Written from docs/evolution13/PROTOCOL.md. It checks an E13 artifact set without running or trusting CAIN code: 1. genesis replicas/threshold; every certified governance event (activation / revocation / quarantine) carries >= threshold distinct valid replica signatures, epochs advance by exactly one, and each event's prev_root equals the governance root recomputed from the replayed state; 2. every activated program_sha256 equals SHA-256 of the published program text; 3. every authorization: quorum valid, epoch/policy/program/governance root match the state the verifier itself replayed at that point, lease digest recomputes, scope stays inside the policy, resource inside the scope, authorization lifetime inside the lease; 4. receipts form a hash chain (seq, prev_receipt_digest); the replayed final governance root AND the receipt-chain head/count are signed by >= threshold replicas (checkpoint.json), so a dropped or reordered event is detected; 5. every receipt: executor signature valid, bound to an authorization in the set (digest), EXECUTED only for a valid authorization, enforced program = attenuated program text hash, loaded program = enforced program, observation Merkle root recomputes, DENY receipts did not spawn. Usage: python3 verify_e13.py exit 0 = INTACT, 1 = problems, 2 = usage """ from __future__ import annotations import base64 import copy import hashlib import json import sys from pathlib import Path from typing import Any, Dict, List, Mapping, Sequence from cryptography.exceptions import InvalidSignature from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PublicKey AUTHZ = "cain.govbpf.authorization.v2" ACT = "cain.govbpf.activation.v2" REV = "cain.govbpf.revocation.v2" OBS = "cain.govbpf.observation.v2" RCPT = "cain.govbpf.receipt.v2" GROOT = "cain.govbpf.governance_root.v2" LEASE = "cain.govproof.lease.v1" def canon(x: Any) -> bytes: return json.dumps(x, ensure_ascii=False, sort_keys=True, separators=(",", ":")).encode("utf-8") def h(x: Any) -> str: return hashlib.sha256(canon(x)).hexdigest() def dg(domain: str, obj: Any) -> str: return h({"d": domain, "o": obj}) def mroot(items: Sequence[Any]) -> str: leaves = [hashlib.sha256(b"\x00" + canon(i)).digest() for i in items] if not leaves: return hashlib.sha256(b"").hexdigest() def rec(xs): if len(xs) == 1: return xs[0] k = 1 while k * 2 < len(xs): k *= 2 return hashlib.sha256(b"\x01" + rec(xs[:k]) + rec(xs[k:])).digest() return rec(leaves).hex() def ed_ok(pub: str, msg: Any, sig: str) -> bool: try: Ed25519PublicKey.from_public_bytes(base64.b64decode(pub)).verify(base64.b64decode(sig), canon(msg)) return True except (InvalidSignature, ValueError, TypeError): return False def qc_ok(cert: Mapping[str, Any], schema: str, replicas: Mapping[str, str], t: int) -> bool: if not isinstance(cert, Mapping) or cert.get("schema") != schema or not isinstance(cert.get("body"), dict): return False msg = {"d": schema + ".qc", "o": cert["body"]} seen = set() for s in cert.get("sigs") or []: r = s.get("replica") if isinstance(s, Mapping) else None if r in replicas and r not in seen and ed_ok(replicas[r], msg, s.get("sig", "")): seen.add(r) return len(seen) >= t def groot(s: Mapping[str, Any]) -> str: a = s["active"] return mroot([["tenant", s["tenant"]], ["epoch", s["epoch"]], ["replicas", dg(GROOT + ".replicas", [s["replicas"], s["threshold"]])], ["active", dg(GROOT + ".active", None if a is None else [a["policy_hash"], a["program_sha256"]])], ["revoked", dg(GROOT + ".revoked", s["revoked"])], ["quarantined", dg(GROOT + ".quarantined", s["quarantined"])], ["history", dg(GROOT + ".history", [x["cert_digest"] for x in s["history"]])]]) def within(path: str, prefixes: Sequence[str]) -> bool: if not isinstance(path, str) or not path.startswith("/") or "/.." in path: return False p = path if path.endswith("/") else path + "/" return any(p.startswith(x) for x in prefixes) def verify_dir(d: Path) -> Dict[str, Any]: probs: List[str] = [] checks = 0 gen = json.loads((d / "genesis.json").read_text()) events = json.loads((d / "events.json").read_text()) # Public bundles publish program SHA-256 commitments only (the program text is CAIN implementation # detail). Without programs/, text-hash checks are reported as COMMITMENT_ONLY instead of verified. commitments_only = not (d / "programs").exists() programs = {} if commitments_only else {p.stem: p.read_text(encoding="utf-8") for p in (d / "programs").glob("*.bt")} def text_ok(sha: str) -> bool: if commitments_only: return isinstance(sha, str) and len(sha) == 64 t = programs.get(sha) return t is not None and hashlib.sha256(t.encode("utf-8")).hexdigest() == sha trust = json.loads((d / "trust.json").read_text()) s = copy.deepcopy(gen) authz_by_digest: Dict[str, Dict[str, Any]] = {} rhead, rcount = "0" * 64, 0 for i, ev in enumerate(events): checks += 1 k = ev["kind"] if k in ("ACTIVATION", "REVOCATION"): cert = ev["cert"] schema = ACT if k == "ACTIVATION" else REV b = cert.get("body", {}) if not qc_ok(cert, schema, s["replicas"], s["threshold"]): probs.append(f"event {i}: quorum certificate invalid") continue if b.get("prev_root") != groot(s) or b.get("tenant") != s["tenant"]: probs.append(f"event {i}: prev_root/tenant does not chain") continue if k == "ACTIVATION": if b.get("new_epoch") != s["epoch"] + 1: probs.append(f"event {i}: epoch does not advance by one") continue if not text_ok(b["program_sha256"]): probs.append(f"event {i}: program text missing or hash mismatch") continue if dg("cain.govbpf.policy.v2", ev["policy"]) != b["policy_hash"]: probs.append(f"event {i}: policy hash mismatch") continue s["active"] = {"policy_hash": b["policy_hash"], "program_sha256": b["program_sha256"], "policy": ev["policy"]} s["epoch"] = b["new_epoch"] else: if b.get("epoch") != s["epoch"]: probs.append(f"event {i}: revocation epoch stale") continue kind, subj = b["kind"], b["subject"] if kind == "QUARANTINE_AGENT": s["quarantined"] = sorted(s["quarantined"] + [subj]) elif kind == "RELEASE_AGENT": if subj in s["revoked"]["AGENT"] or subj not in s["quarantined"]: probs.append(f"event {i}: illegal release") continue s["quarantined"] = [x for x in s["quarantined"] if x != subj] else: s["revoked"][kind] = sorted(s["revoked"][kind] + [subj]) s["history"].append({"kind": k, "epoch": s["epoch"], "cert_digest": dg(schema, b)}) elif k == "AUTHORIZATION": a = ev["authorization"] b = a.get("body", {}) lease = b.get("lease", {}) act = s["active"] ok = (qc_ok(a, AUTHZ, s["replicas"], s["threshold"]) and act is not None and b.get("governance_epoch") == s["epoch"] == lease.get("governance_epoch") and b.get("policy_hash") == act["policy_hash"] and b.get("program_sha256") == act["program_sha256"] and b.get("governance_root") == groot(s) and b.get("lease_digest") == dg(LEASE, lease) and b.get("tenant") == s["tenant"] == lease.get("tenant") and all(within(p, act["policy"]["write_prefixes"]) for p in b.get("write_scope") or ["x"]) and within(lease.get("resource", ""), b.get("write_scope") or []) and b.get("expires_ms", 1 << 62) <= lease.get("expires_ms", -1) and b.get("code_sha256") == hashlib.sha256(ev["code"].encode("utf-8")).hexdigest()) authz_by_digest[dg(AUTHZ, b)] = {"valid": ok, "body": b} if not ok and ev.get("expect_valid", True): probs.append(f"event {i}: authorization {b.get('authz_id')} invalid") elif k == "RECEIPT": r = ev["receipt"] b = r.get("body", {}) if not ed_ok(trust["executor_pub"], {"d": RCPT + ".sig", "o": b}, r.get("sig", "")): probs.append(f"event {i}: receipt signature invalid") continue if b.get("receipt_digest") != dg(RCPT, {x: y for x, y in b.items() if x != "receipt_digest"}): probs.append(f"event {i}: receipt digest mismatch") if b.get("seq") != rcount + 1 or b.get("prev_receipt_digest") != rhead: probs.append(f"event {i}: receipt chain broken (a receipt was removed or reordered)") rhead, rcount = b.get("receipt_digest"), rcount + 1 obs = ev.get("observations", []) if b.get("observation_root") != mroot([dg(OBS, o) for o in obs]) or b.get("observations") != len(obs): probs.append(f"event {i}: observation root mismatch") if any(o.get("grants_authority") is not False for o in obs): probs.append(f"event {i}: an observation claims authority") if b.get("decision") != "EXECUTED" and b.get("governance_root") != groot(s): probs.append(f"event {i}: DENY receipt not bound to the replayed governance state") ref = authz_by_digest.get(b.get("authorization_digest")) if b.get("decision") == "EXECUTED": if ref is None or not ref["valid"]: probs.append(f"event {i}: EXECUTED without a valid authorization in the set") continue if not text_ok(b.get("enforced_program_sha256")): probs.append(f"event {i}: enforced program text missing") if b.get("loaded_program_sha256") != b.get("enforced_program_sha256"): probs.append(f"event {i}: loaded program != enforced program") if b.get("active_program_sha256") != ref["body"].get("program_sha256"): probs.append(f"event {i}: receipt not bound to the authorized program") if b.get("code_sha256") != ref["body"].get("code_sha256"): probs.append(f"event {i}: executed code is not the authorized code") elif b.get("spawned") is not False: probs.append(f"event {i}: DENY receipt claims a spawned process") else: probs.append(f"event {i}: unknown kind {k}") # anchor: >= threshold replicas signed the final (tenant, epoch, root) the verifier recomputed checks += 1 final = groot(s) signed = set() for rep in json.loads((d / "checkpoint.json").read_text()) if (d / "checkpoint.json").exists() else []: rb = rep.get("body", {}) rid = rb.get("replica") if (rid in s["replicas"] and rb.get("root") == final and rb.get("epoch") == s["epoch"] and rb.get("tenant") == s["tenant"] and rb.get("receipts") == {"head": rhead, "count": rcount} and ed_ok(s["replicas"][rid], {"d": "cain.govbpf.replica_report.v2", "o": rb}, rep.get("sig", ""))): signed.add(rid) if len(signed) < s["threshold"]: probs.append("final governance root + receipt chain head are not signed by a replica quorum " "(log truncated or altered?)") return {"result": "INTACT" if not probs else "PROBLEMS", "checks": checks, "passed": checks - len(probs), "problems": probs, "final_governance_root": groot(s), "program_text": "COMMITMENT_ONLY" if commitments_only else "VERIFIED", "final_epoch": s["epoch"]} def main(argv: Sequence[str]) -> int: if len(argv) != 2: print(__doc__) return 2 r = verify_dir(Path(argv[1])) print(json.dumps(r, indent=1)) return 0 if r["result"] == "INTACT" else 1 # ------------------------------------------------------------------ E13 bundle runner def _bundle_sha(p): return hashlib.sha256(p.read_bytes()).hexdigest() def run_bundle(bundle_dir): import pathlib b = pathlib.Path(bundle_dir) problems, checks = [], 0 r = verify_dir(b / "artifacts") checks += r["checks"] problems += r["problems"] for rel, want in sorted(json.loads((b / "bundle_hashes.json").read_text()).items()): checks += 1 p = b / rel if not p.exists() or _bundle_sha(p) != want: problems.append("hash mismatch: " + rel) return checks - len(problems), checks, problems if __name__ == "__main__": _p, _c, _pr = run_bundle(sys.argv[1] if len(sys.argv) > 1 else ".") print(json.dumps({"result": "INTACT" if not _pr else "BROKEN", "passed": _p, "checks": _c, "problems": _pr[:20]})) raise SystemExit(0 if not _pr else 1)