#!/usr/bin/env python3 """Verify a CAIN-45 ZoD live-run bundle without trusting any CAIN website or importing CAIN code. Needs Python 3.8+, `cryptography`, and verify_pbft_qc_bundle.py (published next to this file). The PBFT membership is pinned from membership.json; the hypervisor and approver keys from ZOD_RUN.json. python3 verify_cain45_zod.py Checks (each PASS/FAIL): CHAIN every evidence entry re-hashes, links to the previous one, and is Ed25519-signed by the hypervisor LIFECYCLE every recorded state transition is legal (own copy of the ZoD transition table); no ZoD ever goes CREATED -> RUNNING; nothing executes or calls a tool after a ZoD is TERMINATED CERTIFICATE each ZoD authorization certificate re-hashes to its digest and is signed by the hypervisor CONSENSUS each certificate digest was ordered by the cluster: the operation recomputed HERE hashes to the proposal digest of a commit QC with >= 3 of the 4 pinned signers (verify_pbft_qc_bundle.verify_qc) DECISION the CAIN decision the ZoDs rest on was itself quorum-certified with the pre-verdict it states APPROVAL the pre-verdict was REQUIRE_APPROVAL, so each certificate must carry a human approval that verifies against the approver key for exactly this ZoD, decision and spec ATTACKS every recorded attack was refused, and the refusal is in the signed evidence chain """ from __future__ import annotations import base64 import hashlib import json import sys import urllib.request from pathlib import Path sys.path.insert(0, str(Path(__file__).resolve().parent)) from cryptography.exceptions import InvalidSignature # noqa: E402 from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PublicKey # noqa: E402 D_EVIDENCE, D_ZOD_AUTH, D_APPROVAL, D_ZOD_ORDER = ("CAIN45/EVIDENCE/v1", "CAIN45/ZOD-AUTHORIZATION-CERTIFICATE/v1", "CAIN45/HUMAN-APPROVAL/v1", "CAIN45/ZOD-AUTHORIZATION-ORDER/v1") LEGAL = {"CREATED": {"ATTESTING", "REVOKED", "TERMINATING"}, "ATTESTING": {"VERIFIED", "REVOKED", "TERMINATING"}, "VERIFIED": {"AUTHORIZING", "REVOKED", "TERMINATING"}, "AUTHORIZING": {"AUTHORIZED", "VERIFIED", "REVOKED", "TERMINATING"}, "AUTHORIZED": {"RUNNING", "SUSPENDED", "REVOKED", "TERMINATING"}, "RUNNING": {"SUSPENDED", "QUARANTINED", "TERMINATING", "REVOKED"}, "SUSPENDED": {"REVALIDATING", "QUARANTINED", "TERMINATING", "REVOKED"}, "REVALIDATING": {"RUNNING", "AUTHORIZED", "QUARANTINED", "TERMINATING", "REVOKED"}, "QUARANTINED": {"RECOVERING", "TERMINATING", "REVOKED"}, "RECOVERING": {"REVALIDATING", "TERMINATING", "REVOKED"}, "TERMINATING": {"TERMINATED"}, "TERMINATED": set(), "REVOKED": {"TERMINATING"}} def canon(o): return json.dumps(o, sort_keys=True, separators=(",", ":"), ensure_ascii=True).encode() def digest(domain, fields): return hashlib.sha256(canon({"domain": domain, **fields})).hexdigest() def sig_ok(pub, sig, domain, fields): try: Ed25519PublicKey.from_public_bytes(base64.b64decode(pub)).verify(base64.b64decode(sig), digest(domain, fields).encode()) return True except (InvalidSignature, ValueError, TypeError): return False def load(base, name): if base.startswith("http"): with urllib.request.urlopen(base.rstrip("/") + "/" + name, timeout=30) as r: return json.loads(r.read()) return json.loads((Path(base) / name).read_text()) def verify(base): import verify_pbft_qc_bundle as Q run, chain, decision = load(base, "ZOD_RUN.json"), load(base, "EVIDENCE_CHAIN.json"), load(base, "decision.json") qcs, membership = load(base, "qcs.json"), load(base, "membership.json") hv, approver = run["hypervisor_public_key_b64"], run["approver"]["public_key_b64"] mb = Q.Membership(membership["membership"]) out = [] prev, bad = "0" * 64, [] for i, e in enumerate(chain, 1): f = {k: e[k] for k in ("seq", "prev_hash", "at", "zod_id", "event", "data")} if e["seq"] != i or e["prev_hash"] != prev or digest(D_EVIDENCE, f) != e["entry_hash"] \ or not sig_ok(hv, e["signature_b64"], D_EVIDENCE, f): bad.append(i) prev = e["entry_hash"] out.append(("CHAIN", not bad, f"{len(chain)} entries" + (f"; broken at {bad[:5]}" if bad else ", all linked and signed"))) state, dead, lbad = {}, set(), [] for e in chain: z = e["zod_id"] if e["event"] == "ZOD_CREATED": state[z] = "CREATED" elif e["event"] == "TRANSITION": if e["data"]["from"] != state.get(z) or e["data"]["to"] not in LEGAL.get(state.get(z), set()): lbad.append(f"{z}: {e['data']['from']} -> {e['data']['to']}") state[z] = e["data"]["to"] if state[z] == "TERMINATED": dead.add(z) elif e["event"] in ("EXECUTION_STARTED", "EXECUTION_RESULT", "TOOL_RESULT", "CAPABILITY_GRANTED"): if z in dead or state.get(z) not in ("RUNNING", "AUTHORIZED"): lbad.append(f"{z}: {e['event']} while {state.get(z)}") out.append(("LIFECYCLE", not lbad, "; ".join(lbad[:3]) or f"{len(state)} ZoDs, every transition legal, nothing acted after TERMINATED")) dq = qcs["cain_decision"] c = decision["consensus"] dop = {"action": "fabric_decision", "resource": f"fabric/{decision['decision_id']}", "data": {"decision_id": decision["decision_id"], "commitment_sha256": c["commitment_sha256"], "pre_verdict": c["pre_verdict"]}} want = Q.proposal_request_digest({"operation": dop, "request_id": decision["decision_id"], "client_id": "cain-hosted-fabric"}) ok, why = Q.verify_qc(dq, mb) out.append(("DECISION", ok and dq["proposal_digest"] == want and dq["sequence"] == c["sequence"], f"decision {decision['decision_id']} pre-verdict {c['pre_verdict']}, QC seq {dq['sequence']}: {why or 'quorum ok'}")) approvals = {a["zod_id"]: a for a in load(base, "approval.json")} for n, zd in enumerate(run["zods"], 1): cert = zd["consensus_certificate"] body = {k: v for k, v in cert.items() if k not in ("consensus", "certificate_digest", "signature_b64", "issuer_public_key_b64")} good = digest(D_ZOD_AUTH, body) == cert["certificate_digest"] and sig_ok(hv, cert["signature_b64"], D_ZOD_AUTH, body) \ and cert["zod_id"] == zd["zod_id"] and cert["decision"]["decision_id"] == decision["decision_id"] out.append((f"CERTIFICATE {n}", good, f"{zd['zod_id']} instance {cert['instance_id']}")) con = cert["consensus"] op = {"action": "zod_authorization", "resource": f"zod/{zd['zod_id']}", "data": {"zod_id": zd["zod_id"], "certificate_digest": cert["certificate_digest"], "domain": D_ZOD_ORDER}} want = Q.proposal_request_digest({"operation": op, "request_id": con["request_id"], "client_id": con["client_id"]}) qc = qcs[f"zod_{n}_authorization"] ok, why = Q.verify_qc(qc, mb) out.append((f"CONSENSUS {n}", ok and qc["proposal_digest"] == want and qc["sequence"] == con["sequence"], f"QC seq {qc['sequence']} signers {', '.join(qc.get('signer_set') or [])}: {why or 'quorum ok, digest matches'}")) ap = approvals.get(zd["zod_id"]) spec_hash = hashlib.sha256(canon(zd["spec"])).hexdigest() af = ap and {"human_id": ap["human_id"], "zod_id": zd["zod_id"], "decision_id": decision["decision_id"], "spec_hash": spec_hash, "expires_at": ap["expires_at"]} out.append((f"APPROVAL {n}", bool(ap) and c["pre_verdict"] == "REQUIRE_APPROVAL" and sig_ok(approver, ap["signature_b64"], D_APPROVAL, af) and cert["human_approval"]["signature_b64"] == ap["signature_b64"], f"approver {run['approver']['human_id']}")) denied = [e for e in chain if e["event"] == "DENIED"] att_ok = all(a["result"] == "PASS" for a in run["attacks"]) and len(denied) >= len(run["attacks"]) out.append(("ATTACKS", att_ok, f"{sum(a['result'] == 'PASS' for a in run['attacks'])}/{len(run['attacks'])} refused; " f"{len(denied)} signed DENIED entries")) return out def main(argv): if len(argv) != 1: print(__doc__) return 2 res = verify(argv[0]) for name, ok, detail in res: print(f"[{'PASS' if ok else 'FAIL'}] {name:14s} {detail}") good = all(ok for _, ok, _ in res) print("\nVERIFIED" if good else "\nNOT VERIFIED") return 0 if good else 1 if __name__ == "__main__": sys.exit(main(sys.argv[1:]))