#!/usr/bin/env python3 """Clean-room verifier for the CAIN-42 Evolution 24 evidence bundle (Governed Agentic Internet Fabric). Imports NOTHING from CAIN: Python standard library + `cryptography` only, with its OWN copies of every rule it re-derives (the canonical domain-separated hashing, the fourteen trust dimensions, the protocol map and minimum versions, the quarantine transition table and ceilings, the ten cross-domain authority factors, and the intersection that decides whether a capability is authorized). From the published JSON it re-hashes every file, recomputes every published digest, verifies every Ed25519 signature, re-derives the verdict for every published execution decision and every soundness row, and checks the invariant, bench, end-to-end, simulation, benchmark and mutation records for consistency. It does not re-run the fabric and proves nothing formally. python3 verify_e24.py -> JSON on stdout; exit 0 only if INTACT """ from __future__ import annotations import base64 import hashlib import json import math import re import sys from pathlib import Path from cryptography.exceptions import InvalidSignature from cryptography.hazmat.primitives.asymmetric import ed25519 D = {k: f"CAIN42/E24-{v}/v1" for k, v in { "organization": "ORGANIZATION", "domain": "TRUST-DOMAIN", "identity": "AGENT-IDENTITY", "claim": "CAPABILITY-CLAIM", "cap_evidence": "CAPABILITY-EVIDENCE", "advert": "DISCOVERY-ADVERTISEMENT", "bridge": "FEDERATION-BRIDGE", "offer": "NEGOTIATION-OFFER", "negotiation": "NEGOTIATION-RESULT", "contract": "AGENT-CONTRACT", "contract_sig": "CONTRACT-SIGNATURE", "delegation": "DELEGATION", "relationship": "RELATIONSHIP", "message": "MESSAGE", "exec": "EXECUTION-DECISION", "txn": "TRANSACTION", "escrow": "ESCROW", "reputation": "REPUTATION-EVENT", "dispute": "DISPUTE-RESOLUTION", "incident": "INCIDENT", "revocation": "REVOCATION", "quarantine": "QUARANTINE", "reattest": "REATTESTATION", "artifact": "ARTIFACT", "bom": "MACHINE-SYSTEM-BOM", "memory": "CROSS-AGENT-MEMORY", "evidence": "EVIDENCE-LOG", "screen": "SCREEN-STATE", "adversarial": "ADVERSARIAL", "trust": "TRUST-VECTOR"}.items()} D_MASTER = "CAIN42/E24-MASTER/v1" REQUIRED = ("CAIN42_E24_EVIDENCE_BUNDLE.json", "MANIFEST.json", "SHA256SUMS", "SIGNATURES.json", "SCHEMAS.json", "INVARIANTS.json", "ATTACK_MANIFEST.json", "ATTACK_RESULTS.json", "BENCHMARK_RESULTS.json", "CLEAN_ROOM_RESULTS.json", "AGENT_NETWORK_MANIFEST.json", "TRUST_DOMAIN_MANIFEST.json", "CONTRACT_VECTORS.json", "DELEGATION_VECTORS.json", "REVOCATION_RESULTS.json", "INCIDENT_PROPAGATION_RESULTS.json", "QUARANTINE_RESULTS.json", "SUPPLY_CHAIN_RESULTS.json", "ECONOMIC_RESULTS.json", "PROTOCOL_INTEROP_RESULTS.json", "LIMITATIONS.json", "LIMITATIONS.md", "CAIN42_EVOLUTION24_CLAIMS.json", "PROVENANCE.json", "TEST_RESULTS.json", "MUTATION_RESULTS.json", "README.md", "VERIFICATION.md", "REPRODUCE.txt", "verify_e24.py.txt") FORBIDDEN_MATERIAL = ("-----begin", "private_key", "secret_key", "password") FORBIDDEN_CLAIMS = (r"\bgi\b", r"\basi\b", r"conscious", r"human-equivalent", r"perfect alignment", r"universal safety", r"universal control", r"universal trust", r"solved intelligence", r"controls the internet", r"controls all ai", r"\bguarantee[sd]?\b", r"unhackable") CLAIM_STATES = ("IMPLEMENTED", "TESTED", "VERIFIED", "REPRODUCIBLE", "SIMULATED", "DEMO_ONLY", "UNVERIFIED", "UNKNOWN", "NOT_IMPLEMENTED", "NOT_PERFORMED", "PARTIALLY_VERIFIED", "NOT_VERIFIED") REQUIRED_CATEGORIES = ("agent_impersonation", "identity_cycling", "sybil_network", "trust_laundering", "capability_laundering", "delegation_laundering", "contract_laundering", "reputation_gaming", "economic_collusion", "agent_bribery", "fake_evidence", "protocol_confusion", "protocol_downgrade", "cross_domain_escalation", "revocation_propagation_failure", "quarantine_bypass", "incident_propagation_failure", "supply_chain_compromise", "artifact_substitution", "stale_authorization", "stale_contract", "stale_reputation", "network_partition", "identity_provider_outage", "revocation_provider_outage", "evidence_provider_outage", "capability_spoofing", "computer_use_escalation", "browser_authority_escalation", "economic_abuse", "cross_organization_attack", "memory_poisoning", "authority_laundering", "fail_open", "false_positive_punishment", "recursive_delegation") E2E_STEPS = ("DISCOVERY", "IDENTITY_VERIFICATION", "CAPABILITY_VERIFICATION", "TRUST_EVALUATION", "NEGOTIATION", "CONTRACT", "DELEGATION", "AUTHORIZATION_E8_A_TO_B", "AUTHORIZATION_E8_B_TO_C", "EVIDENCE", "B_COMPROMISED", "DETECTED", "QUARANTINED", "DELEGATIONS_FROZEN", "HIGH_RISK_RELATIONSHIPS_EVALUATED", "INCIDENT_PROPAGATION", "A_AND_C_CONTINUE_WHERE_SAFE", "EVIDENCE_PRESERVED", "REVOCATION", "NETWORK_DID_NOT_COLLAPSE") # ---- independent copies of the E24 rules ------------------------------------------------------------------------ TRUST_DIMS = ("identity_trust", "provenance_trust", "capability_trust", "behavioral_trust", "execution_trust", "evidence_trust", "economic_trust", "communication_trust", "delegation_trust", "temporal_trust", "organizational_trust", "cross_domain_trust", "security_trust", "policy_trust") TRUST_SCALES = ("numeric_0_10", "verified_unverified", "low_medium_high") PROTOCOLS = ("a2a", "mcp", "http", "rest", "grpc", "websocket", "event_bus", "message_queue", "cli", "browser", "computer_use", "local_ipc", "cloud_api") MIN_PROTOCOL_VERSION = {"a2a": (1, 0), "mcp": (2025, 6), "http": (1, 1), "rest": (1, 1), "grpc": (1, 0), "websocket": (13, 0), "event_bus": (1, 0), "message_queue": (1, 0), "cli": (1, 0), "browser": (1, 0), "computer_use": (1, 0), "local_ipc": (1, 0), "cloud_api": (1, 0)} PROTOCOL_MAPS = { "a2a": {"tasks/send": "delegate", "tasks/get": "read", "message/send": "notify"}, "mcp": {"tools/call:read_record": "read", "tools/call:dispatch_order": "dispatch", "tools/call:write_record": "write", "resources/read": "read"}, "http": {"GET": "read", "POST": "write", "PUT": "write", "DELETE": "write"}, "rest": {"GET": "read", "POST": "write"}, "grpc": {"Query": "read", "Mutate": "write"}, "websocket": {"subscribe": "read", "publish": "notify"}, "event_bus": {"emit": "notify"}, "message_queue": {"enqueue": "dispatch"}, "cli": {"cat": "read", "run": "execute"}, "browser": {"navigate": "read", "click:submit_payment": "pay", "click:submit_order": "submit", "type": "write"}, "computer_use": {"screenshot": "read", "click:submit_order": "submit", "click:submit_payment": "pay", "terminal:run": "execute", "file:write": "write"}, "local_ipc": {"call:read": "read", "call:write": "write"}, "cloud_api": {"Describe": "read", "Update": "configure", "Terminate": "configure"}} QUARANTINE_STATES = ("ACTIVE", "RESTRICTED", "SUSPECTED", "QUARANTINED", "ISOLATED", "REVOKED", "RECOVERING", "RESTORED") _CONSEQ, _LOW = ("dispatch", "write", "pay", "delegate", "execute", "submit", "transfer", "configure", "actuate", "hire"), ("read", "notify", "query") QUARANTINE_CEILING = {"ACTIVE": None, "RESTORED": None, "RESTRICTED": frozenset({"read", "notify"}), "SUSPECTED": frozenset({"read"}), "QUARANTINED": frozenset(), "ISOLATED": frozenset(), "REVOKED": frozenset(), "RECOVERING": frozenset({"read"})} QUARANTINE_TRANSITIONS = {"ACTIVE": {"RESTRICTED", "SUSPECTED", "QUARANTINED", "ISOLATED", "REVOKED"}, "RESTORED": {"RESTRICTED", "SUSPECTED", "QUARANTINED", "ISOLATED", "REVOKED"}, "RESTRICTED": {"SUSPECTED", "QUARANTINED", "ISOLATED", "REVOKED", "ACTIVE"}, "SUSPECTED": {"QUARANTINED", "ISOLATED", "REVOKED", "RESTRICTED"}, "QUARANTINED": {"ISOLATED", "REVOKED", "RECOVERING"}, "ISOLATED": {"REVOKED", "RECOVERING"}, "REVOKED": set(), "RECOVERING": {"RESTORED", "QUARANTINED", "REVOKED"}} AUTHORITY_FACTORS = ("local", "remote", "delegated", "contract_scope", "policy", "risk", "capability", "resource", "time", "context") NON_AUTHORITY_INPUTS = ("discovered", "communicated", "negotiated", "contract_value", "coalition_size", "wallet_balance", "payment", "reputation", "trust_score", "federated", "advertised_capability", "routing_choice", "arbitration_outcome", "model_quality", "urgency", "self_assessment", "peer_endorsement") TRUST_POLICIES = { ("numeric_0_10", "low_medium_high"): lambda v: None if v is None else ("HIGH" if v >= 8 else "MEDIUM" if v >= 5 else "LOW"), ("low_medium_high", "numeric_0_10"): lambda v: {"HIGH": 8.0, "MEDIUM": 5.0, "LOW": 1.0}.get(v), ("verified_unverified", "low_medium_high"): lambda v: {"VERIFIED": "MEDIUM", "UNVERIFIED": "LOW"}.get(v), ("numeric_0_10", "unit"): lambda v: None if v is None else (v / 10.0 if 0 <= v <= 10 else None), ("low_medium_high", "unit"): lambda v: {"HIGH": 0.8, "MEDIUM": 0.5, "LOW": 0.1}.get(v), ("verified_unverified", "unit"): lambda v: {"VERIFIED": 0.5, "UNVERIFIED": 0.0}.get(v)} TRANSLATION_INPUTS = {"numeric_0_10": (0, 5, 8, 10), "low_medium_high": ("HIGH", "MEDIUM", "LOW"), "verified_unverified": ("VERIFIED", "UNVERIFIED")} PROBLEMS: list = [] CHECKS = [0, 0] def check(ok: bool, what: str) -> bool: CHECKS[0] += 1 if ok: CHECKS[1] += 1 else: PROBLEMS.append(what) return ok def h(o) -> str: return hashlib.sha256(json.dumps(o, sort_keys=True, separators=(",", ":"), ensure_ascii=True).encode()).hexdigest() def digest(domain: str, fields: dict) -> str: return h({"domain": domain, **fields}) def verify_sig(pub: str, sig: str, domain: str, fields: dict) -> bool: try: ed25519.Ed25519PublicKey.from_public_bytes(base64.b64decode(pub)).verify( base64.b64decode(sig), digest(domain, fields).encode()) return True except (InvalidSignature, ValueError, TypeError): return False def signed_body(obj: dict) -> dict: return {k: v for k, v in obj.items() if k not in ("signer", "signature_b64")} def verify_signed(obj, domain: str, pub: str = None) -> bool: if not isinstance(obj, dict) or "signature_b64" not in obj or "signer" not in obj: return False if pub is not None and obj["signer"] != pub: return False return verify_sig(obj["signer"], obj["signature_b64"], domain, signed_body(obj)) def load(d: Path, name: str): try: return json.loads((d / name).read_text()) except (OSError, ValueError): check(False, f"UNREADABLE:{name}") return {} # ---- independent bridge scope ----------------------------------------------------------------------------------- def bridge_scope(bridges, dom, a, b, now): out, found = set(), False for br in bridges: bd = br["body"] if sorted([a, b]) != bd["domains"] or br.get("revoked"): continue if not (bd["start"] <= now < bd["expires"]): continue d0, d1 = bd["domains"] if not (verify_sig(dom[d0]["pub"], br["sig_a"], D["bridge"], bd) and verify_sig(dom[d1]["pub"], br["sig_b"], D["bridge"], bd)): continue found = True out |= set(bd["capabilities"]) return frozenset(out) if found else frozenset() def derive_authorized(row, ident, dom, bridges, contracts, capev, now): """Re-derive the cross-domain authority intersection for one soundness row, assuming ACTIVE states and full trust (the reference fixture): effective = granted_a n granted_b n contract n policy n capability.""" a, b, cap, cid = row["caller"], row["target"], row["capability"], row["contract"] ia, ib = ident[a]["body"], ident[b]["body"] da, db = ia["domain_id"], ib["domain_id"] if cid and cid in contracts: contract = set(contracts[cid]["body"]["capabilities"]) elif da == db: contract = set(ia["granted"]) else: contract = set() policy = set(dom[da]["ceiling"]) & set(dom[db]["ceiling"]) if da != db: policy = policy & set(bridge_scope(bridges, dom, da, db, now)) verified = {c for (aid, c), ev in capev.items() if aid == b and float(ev["fresh_until"]) > now} capability = set(ib["granted"]) & verified full = set(_CONSEQ) | set(_LOW) eff = (set(ia["granted"]) & set(ib["granted"]) & contract & policy & capability & full) return cap in eff def main(argv) -> int: if len(argv) < 2: print(json.dumps({"result": "USAGE", "usage": "verify_e24.py "})) return 2 d = Path(argv[1]) # ---- files, manifest, sums, signature for f in REQUIRED: check((d / f).is_file(), f"MISSING:{f}") man = load(d, "MANIFEST.json") for name, hx in sorted(man.get("files", {}).items()): p = d / name check(p.is_file() and hashlib.sha256(p.read_bytes()).hexdigest() == hx, f"MANIFEST_HASH:{name}") listed = set(man.get("files", {})) | set(man.get("unhashed_presentation", [])) check(all(p.name in listed for p in d.iterdir() if p.is_file() and p.name != "MANIFEST.json"), "UNLISTED_FILE_IN_BUNDLE") sums = {} for line in (d / "SHA256SUMS").read_text().splitlines() if (d / "SHA256SUMS").is_file() else []: hx, _, name = line.partition(" ") sums[name] = hx for name, hx in sorted(sums.items()): check((d / name).is_file() and hashlib.sha256((d / name).read_bytes()).hexdigest() == hx, f"SHA256SUMS:{name}") sig = load(d, "SIGNATURES.json") check(sig.get("files") == sums, "SIGNATURES_FILES_NE_SHA256SUMS") check(sig.get("master", {}).get("hashes_digest") == h(sums), "MASTER_HASHES_DIGEST") check(verify_sig(sig.get("signer_public_key_b64", ""), sig.get("signature_b64", ""), D_MASTER, sig.get("master", {})), "MASTER_SIGNATURE") for p in d.iterdir(): if p.is_file() and p.name != "verify_e24.py.txt": # the verifier text only lists the denylist words low = p.read_text(errors="ignore").lower() check(not any(m in low for m in FORBIDDEN_MATERIAL), f"KEY_MATERIAL:{p.name}") for name in ("README.md", "VERIFICATION.md", "LIMITATIONS.md", "index.html"): low = (d / name).read_text(errors="ignore").lower() if (d / name).is_file() else "" hits = [pat for pat in FORBIDDEN_CLAIMS if re.search(pat, low)] check(not hits, f"FORBIDDEN_CLAIM:{name}:{hits}") claims = load(d, "CAIN42_EVOLUTION24_CLAIMS.json").get("claims", []) check(bool(claims) and all(c.get("state") in CLAIM_STATES for c in claims), "CLAIM_STATE_INVALID") check(any(c["state"] == "NOT_PERFORMED" and "third-party" in c["statement"] for c in claims), "THIRD_PARTY_NOT_DISCLOSED") check(any(c["state"] == "NOT_IMPLEMENTED" and "network service" in c["statement"] for c in claims), "REAL_NETWORK_NOT_DISCLOSED") check(any(c["state"] == "NOT_IMPLEMENTED" and "real money" in c["statement"] for c in claims), "REAL_MONEY_NOT_DISCLOSED") # ---- trust domains / bridges / translation tdm = load(d, "TRUST_DOMAIN_MANIFEST.json") dom = tdm.get("domains", {}) bridges = tdm.get("bridges", []) check(len(dom) >= 3, "TOO_FEW_DOMAINS") for did, body in sorted(dom.items()): check(body.get("domain_id") == did, f"DOMAIN_ID:{did}") check(body.get("trust_scale") in TRUST_SCALES, f"DOMAIN_SCALE:{did}") check(bool(body.get("ceiling")) and "*" not in body.get("ceiling", []), f"DOMAIN_CEILING:{did}") check(body.get("authority") == "NONE", f"DOMAIN_AUTHORITY:{did}") for br in bridges: bd = br["body"] check(bd.get("domains") == sorted(bd.get("domains", [])), "BRIDGE_DOMAINS_NOT_SORTED") check(set(bd.get("domains", [])) <= set(dom), f"BRIDGE_DOMAIN_UNKNOWN:{bd.get('bridge_id')}") check(bool(bd.get("purpose")) and bool(bd.get("capabilities")), f"BRIDGE_SCOPE:{bd.get('bridge_id')}") a, b = bd["domains"] check(verify_sig(dom[a]["pub"], br["sig_a"], D["bridge"], bd) and verify_sig(dom[b]["pub"], br["sig_b"], D["bridge"], bd), f"BRIDGE_SIGNATURES:{bd.get('bridge_id')}") for x in (a, b): check(set(bd["capabilities"]) <= set(dom[x]["ceiling"]), f"BRIDGE_EXCEEDS_DOMAIN:{bd.get('bridge_id')}") check(bd.get("authority") == "NONE", "BRIDGE_AUTHORITY") tp = tdm.get("translation_policies", {}) for (src, dst), fn in sorted(TRUST_POLICIES.items()): key = f"{src}->{dst}" if key not in tp: continue for v in TRANSLATION_INPUTS[src]: try: want = fn(v) except (TypeError, ValueError): want = "UNKNOWN" want = "UNKNOWN" if want is None else want check(tp[key].get(str(v)) == want, f"TRANSLATION:{key}:{v}") # ---- network: identities, evidence, decisions, chain, soundness anm = load(d, "AGENT_NETWORK_MANIFEST.json") ident = anm.get("identities", {}) orgs = anm.get("organizations", {}) for oid, o in sorted(orgs.items()): check(o.get("domain") in dom, f"ORG_DOMAIN:{oid}") check(set(o.get("authority", [])) <= set(dom.get(o.get("domain", ""), {}).get("ceiling", [])), f"ORG_EXCEEDS_DOMAIN:{oid}") capev = {} for ev in anm.get("capability_evidence", []): aid = ev.get("agent_id") if check(aid in ident, f"CAP_EVIDENCE_AGENT:{aid}"): body = ident[aid]["body"] att = dom.get(body["domain_id"], {}).get("attestors", {}).get(ev.get("attestor", "")) check(att is not None and verify_signed(ev, D["cap_evidence"], pub=att), f"CAP_EVIDENCE_SIG:{aid}:{ev.get('capability')}") check(ev.get("identity_digest") == ident[aid]["digest"], f"CAP_EVIDENCE_IDENTITY:{aid}") check(ev.get("capability") in body["granted"], f"CAP_EVIDENCE_NOT_GRANTED:{aid}") capev[(aid, ev.get("capability"))] = ev trust = anm.get("trust", {}) for aid, t in sorted(trust.items()): check(set(t.get("dims", {})) == set(TRUST_DIMS), f"TRUST_DIMS:{aid}") check(t.get("collapsed_to_single_score") is False and t.get("authority") == "NONE", f"TRUST_COLLAPSED:{aid}") for aid, rec in sorted(ident.items()): body = rec["body"] check(digest(D["identity"], body) == rec["digest"], f"IDENTITY_DIGEST:{aid}") check(body.get("authority") == "NONE", f"IDENTITY_AUTHORITY:{aid}") org = orgs.get(body.get("org_id"), {}) check(set(body.get("granted", [])) <= set(org.get("authority", [])), f"IDENTITY_EXCEEDS_ORG:{aid}") osig = rec.get("org_signature", {}) check(verify_sig(org.get("pub", ""), osig.get("signature_b64", ""), D["identity"], {"identity": rec["digest"]}), f"IDENTITY_ORG_SIGNATURE:{aid}") contracts = load(d, "CONTRACT_VECTORS.json").get("contracts", {}) decisions = anm.get("decisions", []) for dec in decisions: body = {k: v for k, v in dec.items() if k not in ("digest", "authority")} check(digest(D["exec"], body) == dec.get("digest"), f"DECISION_DIGEST:{dec.get('digest', '')[:12]}") if dec.get("decision") == "AUTHORIZED": check(dec.get("commit") == "E8" and dec.get("authority") == "E8" and bool(dec.get("e19_decision")), f"DECISION_E8:{dec.get('digest', '')[:12]}") else: check(dec.get("commit") == "NONE" and dec.get("authority") == "NONE", f"DECISION_AUTHORITY:{dec.get('digest', '')[:12]}") e19 = {x.get("digest") for x in anm.get("e19_decisions", [])} for dec in decisions: if dec.get("decision") == "AUTHORIZED": check(dec.get("e19_decision") in e19, f"DECISION_E19_MISSING:{dec.get('digest', '')[:12]}") chain = anm.get("evidence_chain", []) prev, ok = "genesis", bool(chain) for i, e in enumerate(chain): body = {k: e[k] for k in ("n", "kind", "data", "prev")} if e["n"] != i or e["prev"] != prev or digest(D["evidence"], body) != e["digest"]: ok = False break prev = e["digest"] check(ok, "EVIDENCE_CHAIN") # ---- negotiations / contracts / delegations cv = load(d, "CONTRACT_VECTORS.json") for n in cv.get("negotiations", []): body = {k: v for k, v in n.items() if k != "digest"} check(digest(D["negotiation"], body) == n.get("digest"), f"NEGOTIATION_DIGEST:{n.get('neg_id')}") check(n.get("result") == "AGREED" and n.get("execution_permission") is False and n.get("authority") == "NONE", f"NEGOTIATION_PERMISSION:{n.get('neg_id')}") check(len(n.get("parties", [])) == 2, f"NEGOTIATION_PARTIES:{n.get('neg_id')}") for cid, rec in sorted(contracts.items()): body = rec["body"] check(digest(D["contract"], body) == rec.get("digest"), f"CONTRACT_DIGEST:{cid}") for f in ("contract_id", "version", "parties", "party_identities", "terms", "capabilities", "obligations", "constraints", "budget", "valid_from", "expires_at", "negotiation"): check(f in body, f"CONTRACT_FIELD:{cid}:{f}") check(body.get("authority") == "NONE" and "*" not in body.get("capabilities", []), f"CONTRACT_SCOPE:{cid}") check(body["valid_from"] < body["expires_at"] <= body["valid_from"] + 86400, f"CONTRACT_TIME:{cid}") for p in body["parties"]: check(body["party_identities"].get(p) == ident.get(p, {}).get("digest"), f"CONTRACT_PARTY:{cid}:{p}") for p, s in rec.get("signatures", {}).items(): check(verify_sig(ident.get(p, {}).get("body", {}).get("pub", ""), s.get("signature_b64", ""), D["contract_sig"], {"contract": rec["digest"], "party": p}), f"CONTRACT_SIG:{cid}:{p}") check(set(rec.get("signatures", {})) == set(body["parties"]), f"CONTRACT_NOT_ALL_SIGNED:{cid}") for p in body["parties"]: check(set(body["capabilities"]) <= set(ident.get(p, {}).get("body", {}).get("granted", [])), f"CONTRACT_EXCEEDS_PARTY:{cid}:{p}") dels = load(d, "DELEGATION_VECTORS.json").get("delegations", []) by_id = {x["body"]["del_id"]: x for x in dels} for x in dels: b = x["body"] check(set(b.get("capabilities", [])) and "*" not in b.get("capabilities", []), f"DELEGATION_SCOPE:{b.get('del_id')}") check(bool(b.get("purpose")) and b.get("authority") == "NONE", f"DELEGATION_PURPOSE:{b.get('del_id')}") check(b["start"] < b["expires"], f"DELEGATION_TIME:{b.get('del_id')}") check(verify_sig(ident.get(b["delegator"], {}).get("body", {}).get("pub", ""), x.get("sig", ""), D["delegation"], b), f"DELEGATION_SIG:{b.get('del_id')}") if b.get("parent"): par = by_id.get(b["parent"], {}).get("body", {}) check(set(b["capabilities"]) <= set(par.get("capabilities", [])), f"DELEGATION_EXCEEDS_PARENT:{b.get('del_id')}") check(b["depth"] == par.get("depth", 0) + 1, f"DELEGATION_DEPTH:{b.get('del_id')}") check(b["expires"] <= par.get("expires", float("inf")), f"DELEGATION_OUTLIVES_PARENT:{b.get('del_id')}") # ---- revocation preserves unrelated contracts rev = load(d, "REVOCATION_RESULTS.json").get("revocation", {}) if rev: rb = {k: v for k, v in rev.items() if k != "digest"} check(digest(D["revocation"], rb) == rev.get("digest"), "REVOCATION_DIGEST") check(rev.get("authority") == "NONE", "REVOCATION_AUTHORITY") for cid, rec in contracts.items(): if cid in rev.get("contracts", []): check(rec.get("state") == "REVOKED", f"REVOKED_CONTRACT_ACTIVE:{cid}") else: check(rec.get("state") == "ACTIVE", f"UNRELATED_CONTRACT_REVOKED:{cid}") # ---- incident / quarantine inc = load(d, "INCIDENT_PROPAGATION_RESULTS.json") e2e = inc.get("end_to_end", {}) check(e2e.get("order") == list(E2E_STEPS) and e2e.get("all_ok") is True and all(s.get("ok") for s in e2e.get("steps", {}).values()), "END_TO_END") br = inc.get("blast_radius", {}) if br: bb = {k: v for k, v in br.items() if k != "digest"} check(digest(D["incident"], bb) == br.get("digest"), "BLAST_RADIUS_DIGEST") check(br.get("authority") == "NONE", "BLAST_RADIUS_AUTHORITY") step = e2e.get("steps", {}).get("INCIDENT_PROPAGATION", {}) check(br.get("compromised") == "agent-b", "BLAST_RADIUS_COMPROMISED") check(not (set(br.get("exposed", {})) & set(br.get("unrelated_untouched", []))), "BLAST_RADIUS_OVERLAP") check("agent-a2" in br.get("unrelated_untouched", []) and "agent-c2" in br.get("unrelated_untouched", []), f"BLAST_RADIUS_UNRELATED:{step.get('blast')}") check(inc.get("denied_after_quarantine", {}).get("decision") == "DENY", "QUARANTINE_DID_NOT_DENY") check(inc.get("unrelated_domain_continues", {}).get("decision") == "AUTHORIZED", "UNRELATED_DOMAIN_HALTED") qr = load(d, "QUARANTINE_RESULTS.json") for a, cur, to, _t, _e in qr.get("state_history", []): check(to in QUARANTINE_TRANSITIONS.get(cur, set()), f"QUARANTINE_TRANSITION:{cur}->{to}") for a, st in qr.get("states", {}).items(): check(st in QUARANTINE_STATES, f"QUARANTINE_STATE:{st}") check(QUARANTINE_CEILING[st] is None or set(QUARANTINE_CEILING[st]) <= set(_CONSEQ) | set(_LOW), f"QUARANTINE_CEILING:{st}") check({k: sorted(v) for k, v in (qr.get("transitions") or {}).items()} == {k: sorted(v) for k, v in QUARANTINE_TRANSITIONS.items()}, "QUARANTINE_TRANSITIONS_REDERIVED") check({k: (sorted(v) if v is not None else None) for k, v in (qr.get("ceilings") or {}).items()} == {k: (sorted(v) if v is not None else None) for k, v in QUARANTINE_CEILING.items()}, "QUARANTINE_CEILINGS_REDERIVED") # ---- economy / protocols / supply chain eco = load(d, "ECONOMIC_RESULTS.json") s = eco.get("settlement", {}) if s: ver = s.get("verifier", "") pub = None for _did, db in dom.items(): if db.get("attestors", {}).get(ver): pub = db["attestors"][ver] check(pub is not None and verify_sig(pub, s.get("signature_b64", ""), D["txn"], signed_body(s)), f"SETTLEMENT_SIGNATURE:{ver}") check(s.get("real_money_moved") is False and s.get("settlement") == "SIMULATED" and s.get("authority") == "NONE", "SETTLEMENT_NOT_SIMULATED") check(eco.get("result_kind") == "SIMULATED" and eco.get("real_money_moved") is False, "ECONOMY_LABEL") for name, a in eco.get("attacks", {}).items(): check(a.get("contained") is True, f"ECONOMY_ATTACK:{name}") pi = load(d, "PROTOCOL_INTEROP_RESULTS.json") check(set(pi.get("protocols", [])) == set(PROTOCOLS), "PROTOCOL_SET") for m in pi.get("mappings", []): p, op = m["protocol"], m["op"] check(p in PROTOCOLS, f"INTEROP_PROTOCOL:{p}") check(m.get("capability") == PROTOCOL_MAPS.get(p, {}).get(op, "UNKNOWN"), f"INTEROP_CAP:{p}:{op}") check(list(m.get("min_version", [])) == list(MIN_PROTOCOL_VERSION[p]), f"INTEROP_VERSION:{p}") check(m.get("downgrade_refused") is True, f"INTEROP_DOWNGRADE:{p}:{op}") sc = load(d, "SUPPLY_CHAIN_RESULTS.json") check(bool(sc.get("attacks")) and all(a.get("contained") is True for a in sc.get("attacks", {}).values()), "SUPPLY_ATTACKS") # ---- soundness: re-derive every authorized verdict now = 1002.0 for row in anm.get("soundness", []): dec = row.get("decision", {}) body = {k: v for k, v in dec.items() if k not in ("digest", "authority")} check(digest(D["exec"], body) == dec.get("digest"), f"SOUNDNESS_DIGEST:{row['caller']}->{row['target']}") want = derive_authorized(row, ident, dom, bridges, contracts, capev, now) got = dec.get("decision") == "AUTHORIZED" check(want == got, f"SOUNDNESS:{row['caller']}->{row['target']}:{row['capability']}:derived={want}:got={got}") if got: check(dec.get("commit") == "E8" and dec.get("authority") == "E8", "SOUNDNESS_E8") # ---- invariants + bench inv = load(d, "INVARIANTS.json") check(inv.get("checked") == len(inv.get("checks", [])) >= 300, "INVARIANTS_COUNT") check(inv.get("all_hold") is True and all(c["holds"] for c in inv.get("checks", [])) and not inv.get("failed"), "INVARIANTS_HOLD") ids = [c["id"] for c in inv.get("checks", [])] check(len(ids) == len(set(ids)), "INVARIANT_IDS_UNIQUE") check(all(f"NET-I{i:03d}" in ids for i in range(1, 11)), "SPEC_INVARIANTS_PRESENT") check(all(f"NET-L{i:02d}" in ids for i in range(1, 21)), "SPEC_LAWS_PRESENT") ar = load(d, "ATTACK_RESULTS.json") atk = ar.get("attacks", {}) check(ar.get("total") == len(atk) >= 750 and ar.get("contained") == sum(1 for a in atk.values() if a["contained"]), "BENCH_COUNTS") check(ar.get("all_contained") is True, "BENCH_ALL_CONTAINED") check(ar.get("distinct_families") == len({a["family"] for a in atk.values()}), "BENCH_FAMILIES") check(ar.get("real_world_attack_validation") == "NOT_PERFORMED", "BENCH_REAL_WORLD_DISCLOSURE") check(all(c in ar.get("categories", {}) for c in REQUIRED_CATEGORIES), "BENCH_CATEGORIES") bad = [k for k, a in atk.items() if digest(D["adversarial"], {x: y for x, y in a.items() if x != "digest"}) != a["digest"]] check(not bad, f"ATTACK_DIGESTS:{bad[:3]}") mf = load(d, "ATTACK_MANIFEST.json") check({a["attack"]: a["digest"] for a in mf.get("attacks", [])} == {k: a["digest"] for k, a in atk.items()}, "ATTACK_MANIFEST_MATCHES_RESULTS") inv_ids = set(ids) check(all(a["expected_invariant"] in inv_ids for a in atk.values()), "ATTACK_TARGETS_EXIST") # ---- benchmark, mutation, tests, summary bm = load(d, "BENCHMARK_RESULTS.json") sim = bm.get("extreme_simulation", {}) bench = bm.get("benchmark", {}) check(sim.get("population", {}).get("agents", 0) >= 10000, "SIM_POPULATION") check(sim.get("results", {}).get("authority_leakage") == 0, "SIM_LEAKAGE") check(str(sim.get("result_kind", "")).startswith("SIMULATION"), "SIM_LABEL") check(sim.get("authority") == "NONE", "SIM_AUTHORITY") for k in ("governed_execute_e19_e8", "revocation_propagation_per_identity_ms", "firebreak_containment_ms", "false_negative_rate_on_bench", "false_positive_rate_on_legitimate_set"): check(k in bench, f"BENCH_METRIC:{k}") check(bench.get("false_negative_rate_on_bench") == 0.0 and bench.get("false_positive_rate_on_legitimate_set") == 0.0, "BENCH_RATES") check(bm.get("environment", {}).get("python"), "BENCH_ENVIRONMENT") mu = load(d, "MUTATION_RESULTS.json") check(mu.get("all_mutants_caught") is True and len(mu.get("rows", [])) >= 6, "MUTATION_SELFTEST") rows = mu.get("rows", []) if rows: base = rows[0] check(all((r["bench_contained"] < base["bench_contained"] and r["invariants_holding"] < base["invariants_holding"]) == r["caught"] for r in rows[1:]), "MUTATION_ROWS_CONSISTENT") te = load(d, "TEST_RESULTS.json") check(te.get("failed") == 0 and te.get("errors") == 0 and te.get("passed", 0) > 0 and te.get("exit_code") == 0, "E24_TESTS") s2 = load(d, "CAIN42_E24_EVIDENCE_BUNDLE.json") check(s2.get("status") in ("COMPLETE", "INCOMPLETE"), "STATUS_VALUE") gates = s2.get("completion_gates", {}) check(s2.get("status") == ("COMPLETE" if gates and all(v is True for v in gates.values()) else "INCOMPLETE"), "STATUS_FOLLOWS_GATES") check(s2.get("invariants", {}).get("checked") == inv.get("checked") and s2.get("attacks", {}).get("total") == ar.get("total"), "SUMMARY_MATCHES") check(s2.get("authority") == "NONE", "SUMMARY_AUTHORITY") out = {"result": "INTACT" if not PROBLEMS else "BROKEN", "checks": CHECKS[0], "passed": CHECKS[1], "problems": PROBLEMS[:60], "imports_cain": False} print(json.dumps(out, indent=1)) return 0 if not PROBLEMS else 1 if __name__ == "__main__": sys.exit(main(sys.argv))