#!/usr/bin/env python3 """Clean-room verifier for a CAIN governance package registry export (cain.govreg.export/1). Zero CAIN imports. From the exported, quorum-certified event log alone it re-derives and checks: * the membership digest and q = ceil((n + f + 1) / 2), f = floor((n - 1) / 3) * every event's canonical hash, its prev_hash chain and contiguous sequence numbers * every event's certificate: >= q valid Ed25519 endorsements from DISTINCT members over that event hash * every event is a legal transition (lifecycle table, sticky revocation, anti-rollback) * the state of each package at --at-ms (expiry applied), and which package was active usage: govreg_verify_offline.py EXPORT.json [--at-ms N] [--package DIGEST] exit 0 = VERIFIED, 1 = UNVERIFIED """ import argparse import base64 import hashlib import json import math import sys import unicodedata US = b"\x1f" T = {None: {"PUBLISHED"}, "PUBLISHED": {"ACTIVE", "REVOKED", "EXPIRED"}, "ACTIVE": {"SUSPENDED", "SUPERSEDED", "REVOKED", "EXPIRED"}, "SUSPENDED": {"ACTIVE", "SUPERSEDED", "REVOKED", "EXPIRED"}, "SUPERSEDED": set(), "REVOKED": set(), "EXPIRED": set()} KIND_TO = {"PUBLISH": "PUBLISHED", "ACTIVATE": "ACTIVE", "SUSPEND": "SUSPENDED", "REVOKE": "REVOKED"} TERMINAL = {"SUPERSEDED", "REVOKED", "EXPIRED"} class Bad(Exception): pass def _walk(v): if v is None or isinstance(v, bool): return if isinstance(v, int): if not -(2 ** 53) < v < 2 ** 53: raise Bad("integer out of range") return if isinstance(v, str): if unicodedata.normalize("NFC", v) != v or "\x00" in v: raise Bad("string not NFC") return if isinstance(v, list): for x in v: _walk(x) return if isinstance(v, dict): for k, x in v.items(): if not (isinstance(k, str) and 1 <= len(k) <= 64 and all(c in "abcdefghijklmnopqrstuvwxyz0123456789_" for c in k)): raise Bad(f"bad key {k!r}") _walk(x) return raise Bad(f"type {type(v).__name__}") def canon(v) -> bytes: _walk(v) return json.dumps(v, sort_keys=True, separators=(",", ":"), ensure_ascii=False, allow_nan=False).encode() def sha(b: bytes) -> str: return hashlib.sha256(b).hexdigest() def sig_ok(pub_b64, msg, sig_b64): try: from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PublicKey Ed25519PublicKey.from_public_bytes(base64.b64decode(pub_b64, validate=True)).verify( base64.b64decode(sig_b64, validate=True), msg) return True except Exception: # noqa: BLE001 return False def _expired_view(st, meta, active, now): """State as seen at time `now`: non-terminal packages past valid_until read EXPIRED (not stored).""" v = dict(st) for x, s in st.items(): if s not in TERMINAL and now >= meta[x]["valid_until_ms"]: v[x] = "EXPIRED" return v, (None if active and v.get(active) == "EXPIRED" else active) def replay(events, at_ms=None): st, meta, revoked, active, max_act = {}, {}, set(), None, 0 for e in events: d, k = e["package_digest"], e["kind"] view, _ = _expired_view(st, meta, active, e["at_ms"]) # legality is judged at the event's time cur = view.get(d) if d in revoked and k != "REVOKE": raise Bad(f"seq {e['seq']}: event on revoked package") if k == "REVOKE" and cur == "REVOKED": raise Bad(f"seq {e['seq']}: double revoke") to = KIND_TO.get(k) if to is None or to not in T.get(cur, set()): raise Bad(f"seq {e['seq']}: illegal {cur} -> {to}") if k == "ACTIVATE" and meta[d]["published_seq"] < max_act: raise Bad(f"seq {e['seq']}: rollback to an older publication") # stored state (no expiry applied), exactly as the runtime's derive() does if k == "PUBLISH": meta[d] = {"valid_until_ms": e["valid_until_ms"], "source_digest": e["source_digest"], "published_seq": e["seq"], "signer": e.get("signer", "")} elif k == "ACTIVATE": if active and active != d and st.get(active) == "ACTIVE": st[active] = "SUPERSEDED" active, max_act = d, max(max_act, meta[d]["published_seq"]) elif k in ("SUSPEND", "REVOKE") and active == d: active = None if k == "REVOKE": revoked.add(d) st[d] = to if at_ms is not None: st, active = _expired_view(st, meta, active, at_ms) return st, active, sorted(revoked) def verify(export, at_ms=None, package=None): checks = [] def mark(name, ok, why=""): checks.append({"check": name, "ok": bool(ok), "why": why}) out = {} try: mem = export["membership"]["replicas"] n = len(mem) f = (n - 1) // 3 q = math.ceil((n + f + 1) / 2) mdig = sha(b"CAIN/contract/v1/membership" + US + canon(dict(sorted(mem.items())))) mark("membership", export["membership"].get("digest") == mdig and export["membership"].get("q") == q, f"n={n} f={f} q={q}") prev, evs = "", [] for i, raw in enumerate(export["events"], 1): cert = raw["cert"] e = {k: v for k, v in raw.items() if k != "cert"} h = sha(b"CAIN/govreg/v1/event" + US + canon(e)) good = {rid for rid, s in (cert.get("signatures") or {}).items() if rid in mem and sig_ok(mem[rid], b"CAIN/govreg/v1/endorse" + US + h.encode(), s)} mark(f"seq{i}:chain", e["seq"] == i and e["prev_hash"] == prev and e["tenant"] == export["tenant"] and e["namespace"] == export["namespace"]) mark(f"seq{i}:cert", cert.get("event_hash") == h and cert.get("membership_digest") == mdig and len(good) >= q, f"{len(good)}/{q} endorsements") prev = h evs.append(e) st, active, revoked = replay(evs, at_ms) mark("lifecycle", True, "every event is a legal transition") out = {"states": st, "active": active, "revoked": revoked} if package: out["package_state"] = st.get(package, "NOT_IN_LOG") except (Bad, KeyError, TypeError, ValueError, AttributeError) as exc: mark("structure_or_lifecycle", False, f"{type(exc).__name__}: {exc}") ok = all(c["ok"] for c in checks) return {"verifier": "govreg_verify_offline/1", "status": "VERIFIED" if ok else "UNVERIFIED", "ok": ok, "checks": checks, **out} def main(argv=None): ap = argparse.ArgumentParser() ap.add_argument("export") ap.add_argument("--at-ms", type=int) ap.add_argument("--package") a = ap.parse_args(argv) with open(a.export, encoding="utf-8") as fh: r = verify(json.load(fh), a.at_ms, a.package) print(json.dumps(r, indent=1)) return 0 if r["ok"] else 1 if __name__ == "__main__": sys.exit(main())