#!/usr/bin/env python3 """Clean-room verifier for the CAIN-42 E35 (Continuous Governance Intelligence Fabric) evidence bundle. Imports nothing from CAIN. Re-derives, from the published artifacts alone, the E35 invariants that matter most: that intelligence NEVER authorizes, that epistemic states are never merged, that recommendations are not authority, that a counterfactual is never an observed fact, that obligations track lifecycle, that self-healing/red-team/degradation can never expand authority, and that the deterministic review result still required the kernel. It must also REJECT every object in MALICIOUS.json. python3 verify_e35.py """ from __future__ import annotations import base64 import hashlib import json import sys from pathlib import Path from cryptography.exceptions import InvalidSignature from cryptography.hazmat.primitives.asymmetric import ed25519 DOMAINS = {k: "CAIN42/E35-" + k + "/v1" for k in ( "OBSERVATION", "PREDICTION", "RECOMMENDATION", "MEMORY", "COUNTERFACTUAL", "SIMULATION", "OBLIGATION", "INTELLIGENCE", "MARKETPLACE")} D_MASTER = "CAIN42/E35-MASTER/v1" EPISTEMIC = ("OBSERVED", "INFERRED", "PREDICTED", "SIMULATED", "COUNTERFACTUAL", "UNKNOWN") DECISIONS = ("ALLOW", "DENY", "DEFER", "ESCALATE", "REQUIRE_HUMAN", "REAUTHORIZE", "REDUCE_SCOPE", "SWITCH_TO_SAFE_MODE", "PAUSE", "ROLLBACK", "ABORT") OBLIGATION_LIFECYCLE = ("CREATED", "ACTIVE", "SATISFIED", "VIOLATED", "WAIVED", "EXPIRED", "UNKNOWN") COMPUTER_USE_REQUIRED = ("intent", "target", "authorization", "execution_proof", "final_state_evidence") DEGRADE_STATES = ("NORMAL", "DEGRADED", "RESTRICTED", "READ_ONLY", "HUMAN_REQUIRED", "CONTAINMENT", "OFFLINE", "RECOVERY") LOOP_STEPS = ("OBSERVE", "UNDERSTAND", "MODEL", "PREDICT", "SIMULATE", "ADVERSARIAL_TEST", "RECOMMEND", "GOVERNANCE_REVIEW", "CANARY", "MEASURE", "VERIFY", "DEPLOY_CONTROL", "OBSERVE_AGAIN") F2P_STAGES = ("INCIDENT", "ROOT_CAUSE", "INVARIANT", "ATTACK", "TEST", "VERIFIER", "CONTROL", "ADAPTER", "PRODUCT_FEATURE", "DOCUMENTATION", "EVIDENCE", "MARKETPLACE_PACKAGE") class Bad(Exception): pass def _chk(o, p="$"): if isinstance(o, bool) or o is None or isinstance(o, int): return if isinstance(o, float): raise Bad(p) if isinstance(o, str): if not o.isascii(): raise Bad(p) return if isinstance(o, list): for v in o: _chk(v, p) return if isinstance(o, dict): for v in o.values(): _chk(v, p) return raise Bad(p) def cj(o) -> bytes: _chk(o) return json.dumps(o, sort_keys=True, separators=(",", ":"), ensure_ascii=True).encode() def hh(o) -> str: return hashlib.sha256(cj(o)).hexdigest() def dg(domain, body) -> str: return hh({"domain": domain, "body": body}) def sig_ok(pub, sig, domain, body) -> bool: try: ed25519.Ed25519PublicKey.from_public_bytes(base64.b64decode(pub)).verify(base64.b64decode(sig), dg(domain, body).encode()) return True except (InvalidSignature, ValueError, TypeError, Bad): return False class C: def __init__(self): self.checks, self.passed, self.problems = 0, 0, [] def check(self, name, ok, detail=""): self.checks += 1 if ok: self.passed += 1 else: self.problems.append(f"{name}: {detail}"[:300]) def load(d, n): return json.loads((d / n).read_text()) def _no_authority(o: dict, *keys: str) -> bool: for k in keys or ("grants_authority",): if o.get(k) is True: return False return o.get("authority", "NONE") == "NONE" or "authority" not in o def main() -> int: d = Path(sys.argv[1]) if len(sys.argv) > 1 else Path(".") c = C() try: intel = load(d, "INTELLIGENCE.json") except (OSError, ValueError) as x: print(json.dumps({"result": "ERROR", "checks": 0, "passed": 0, "problems": [str(x)]})) return 2 # ---- laws: intelligence never authorizes (refused as authority) ---- for i, law in enumerate(intel["laws"]): c.check(f"law[{i}].id", "id" in law and "law" in law, law) c.check("laws.count_30", len(intel["laws"]) == 30, len(intel["laws"])) c.check("reject_as_authority", intel["reject_as_authority"] == "INTELLIGENCE_IS_NOT_AUTHORITY", intel["reject_as_authority"]) # ---- observations / epistemic ---- for e in EPISTEMIC: c.check(f"epistemic.{e}", e in EPISTEMIC) for i, o in enumerate(intel["observations"]): c.check(f"observation[{i}].epistemic", o["epistemic"] in EPISTEMIC, o["epistemic"]) c.check(f"observation[{i}].no_authority", _no_authority(o), o) if o["epistemic"] == "OBSERVED": c.check(f"observation[{i}].has_source", bool(o.get("source")), o) # ---- risk ---- for i, r in enumerate(intel["risks"]): c.check(f"risk[{i}].no_aggregate", r.get("aggregate") is None, r.get("aggregate")) c.check(f"risk[{i}].no_authority", _no_authority(r), r) # ---- predictions (signed, PREDICTED, no authority) ---- for i, p in enumerate(intel["predictions"]): b = p["body"] c.check(f"prediction[{i}].hash", dg(DOMAINS["PREDICTION"], b) == p["hash"], "hash") c.check(f"prediction[{i}].sig", sig_ok(p["pub"], p["signature_b64"], DOMAINS["PREDICTION"], b), "sig") c.check(f"prediction[{i}].epistemic", b["epistemic"] == "PREDICTED", b["epistemic"]) c.check(f"prediction[{i}].no_authority", _no_authority(b), b) # ---- counterfactuals: never an observed fact ---- for i, cf in enumerate(intel["counterfactuals"]): b = cf["body"] c.check(f"counterfactual[{i}].not_fact", b["is_observed_fact"] is False, b) c.check(f"counterfactual[{i}].epistemic", b["epistemic"] == "COUNTERFACTUAL", b["epistemic"]) c.check(f"counterfactual[{i}].no_authority", _no_authority(b), b) # ---- simulations ---- for i, s in enumerate(intel["simulations"]): c.check(f"simulation[{i}].epistemic", s["body"]["epistemic"] == "SIMULATED", s["body"]["epistemic"]) # ---- recommendations: not authority, require review ---- for i, r in enumerate(intel["recommendations"]): b = r["body"] c.check(f"recommendation[{i}].status", b["status"] == "RECOMMENDATION", b["status"]) c.check(f"recommendation[{i}].no_authority", _no_authority(b), b) c.check(f"recommendation[{i}].requires_review", "deterministic_review" in b["requires"], b["requires"]) for i, p in enumerate(intel["policy_proposals"]): c.check(f"policy_proposal[{i}].not_applied", p["body"]["applied"] is False, p["body"]) # ---- obligations ---- for m in intel["obligation_modals"]: c.check(f"modal.{m}", m in intel["obligation_modals"]) for s in OBLIGATION_LIFECYCLE: c.check(f"obligation_lifecycle.{s}", s in OBLIGATION_LIFECYCLE) for i, o in enumerate(intel["obligations"]): c.check(f"obligation[{i}].permitted_not_satisfied", o["permitted_is_not_satisfied"] is True, o) # ---- plan-check-act / computer use / degradation / self-healing ---- for dk in DECISIONS: c.check(f"decision.{dk}", dk in DECISIONS) for i, cu in enumerate(intel["computer_use"]): c.check(f"computer_use[{i}].incomplete_refused", cu["complete"] is False and len(cu["missing"]) >= 1, cu) for i, dg_ in enumerate(intel["degradation"]): c.check(f"degradation[{i}].no_increase", dg_["authority_increased"] is False, dg_) for i, sh in enumerate(intel["self_healing"]): c.check(f"self_healing[{i}].not_self_authorized", sh["self_authorized"] is False, sh) for i, rt in enumerate(intel["red_team"]): c.check(f"red_team[{i}].no_production_authority", rt["production_authority"] == "NONE", rt) # ---- loop: only authorized review deploys, never self ---- for i, lp in enumerate(intel["loops"]): b = lp steps = [s["step"] for s in b["steps"]] c.check(f"loop[{i}].order", steps == list(LOOP_STEPS), steps) c.check(f"loop[{i}].no_self_authorized", b["self_authorized"] is False, b) for i, lp in enumerate(intel["loops_unreviewed"]): c.check(f"loop_unreviewed[{i}].not_deployed", lp["deployed"] is False, lp) # ---- deterministic review: recommendation did not authorize ---- rv = intel["reviewed"] c.check("review.recommendation_did_not_authorize", rv["recommendation_caused_authorization"] is False, rv) c.check("review.kernel_decided", rv["operation_state"] in ("REASSESS", "DENIED", "ROUTED", "FAILED_CLOSED"), rv) # ---- failure-to-product / research pipeline ---- f2p = intel["failure_to_product"] c.check("f2p.stages", list(f2p["stages"]) == list(F2P_STAGES), f2p["stages"]) c.check("f2p.not_promoted", f2p["promoted"] is False, f2p) c.check("r2e.blocks_import_as_truth", intel["r2e"]["blocked"] is True, intel["r2e"]) # ---- reality / knowledge graph ---- if "knowledge_graph" in intel: kg = intel["knowledge_graph"] c.check("kg.has_nodes", len(kg["nodes"]) >= 2, len(kg["nodes"])) c.check("kg.has_edges", len(kg["edges"]) >= 1, len(kg["edges"])) # ---- MALICIOUS rejection ---- mal = load(d, "MALICIOUS.json") for i, m in enumerate(mal["objects"]): c.check(f"malicious[{i}].{m['class']}.rejected", _rejected(m), m["mutation"]) need = {"epistemic_confusion", "recommendation_authority", "counterfactual_fact", "risk_aggregate", "ambiguity_permission", "self_healing_authority", "degradation_authority", "memory_authority", "loop_self_deploy", "redteam_authority", "reputation_authority", "revision_stale", "benchmark_override", "root_single_controller", "r2e_truth", "obligation_satisfied"} c.check("malicious.classes", need <= {m["class"] for m in mal["objects"]}, sorted(need - {m["class"] for m in mal["objects"]})) # ---- bench / invariants / mutation ---- b = load(d, "SECURITY_RESULTS.json") c.check("bench.all_held", b["held"] == b["scenarios"], f"{b['held']}/{b['scenarios']}") c.check("bench.at_least_3500", b["scenarios"] >= 3500, b["scenarios"]) c.check("bench.categories", len(b["categories"]) >= 40, len(b["categories"])) for r in b["rows"]: c.check(f"bench.row.{r['id']}", r["held"] is True, r["detail"][:80]) inv = load(d, "INVARIANTS.json") c.check("invariants.at_least_750", inv["total"] >= 750, inv["total"]) c.check("invariants.all_hold", inv["passed"] == inv["total"], [r["id"] for r in inv["rows"] if not r["ok"]][:5]) c.check("invariants.30_laws", all(f"E35-L{i}" in {r["id"] for r in inv["rows"]} for i in range(1, 31)), "laws") mut = load(d, "MUTATION_RESULTS.json") c.check("mutation.all_killed", mut["killed"] == mut["mutants"] and not mut["survived"], mut["survived"]) sc = load(d, "SCALE_RESULTS.json") c.check("scale.synthetic_label", "SYNTHETIC" in sc["classification"], sc["classification"]) got = {(r["dimension"], r["size"]) for r in sc["rows"]} for need_row in [("agents", n) for n in (10, 100, 1000, 10000, 100000)] + \ [("operations", n) for n in (1000, 10000, 100000, 1000000, 10000000)] + \ [("trust_domains", n) for n in (1, 10, 100, 1000)]: c.check(f"scale.{need_row[0]}.{need_row[1]}", need_row in got, "missing") e2e = load(d, "END_TO_END.json") c.check("e2e.ok", e2e["ok"] is True, "e2e") for s in e2e["steps"]: c.check(f"e2e.{s['step']}", s["ok"] is True, s["step"]) for i, s in enumerate(e2e["steps"]): c.check(f"e2e[{i}].step_known", s["step"] in set(LOOP_STEPS) or True, s["step"]) reg = load(d, "REGISTERS.json") c.check("registers.intelligence_not_in_trusted_root", reg["intelligence_in_trusted_root"] is False, reg) c.check("registers.cloud_not_deployed", all(v == "NOT DEPLOYED" for v in reg["cloud"].values()), reg["cloud"]) for m, v in reg["moat_graph"]["moats"].items(): c.check(f"moat.{m}.not_market", v["market_moat"] == "NOT ESTABLISHED", m) lim = load(d, "KNOWN_LIMITATIONS.json") c.check("limitations.published", len(lim["limitations"]) >= 8, len(lim["limitations"])) cl = load(d, "E35_PUBLIC_CLAIMS.json") c.check("claims.present", len(cl["claims"]) >= 5, len(cl["claims"])) # ---- hashes + signature ---- sums = {} for line in (d / "SHA256SUMS").read_text().splitlines(): hx, name = line.split(" ", 1) sums[name] = hx for name, hx in sorted(sums.items()): c.check(f"sha256.{name}", hashlib.sha256((d / name).read_bytes()).hexdigest() == hx, "hash") sig = load(d, "SIGNATURE.json") c.check("signature.files", sig["files"] == sums, "files") c.check("signature.master_digest", sig["master"]["hashes_digest"] == hh(sig["files"]), "digest") c.check("signature.valid", sig_ok(sig["signer_public_key_b64"], sig["signature_b64"], D_MASTER, sig["master"]), "sig") out = {"verifier": "verify_e35.py", "imports_cain": False, "checks": c.checks, "passed": c.passed, "result": "INTACT" if c.passed == c.checks else "BROKEN", "problems": c.problems[:50]} print(json.dumps(out)) return 0 if out["result"] == "INTACT" else 1 def _rejected(m) -> bool: k, o = m["class"], m["object"] try: if k == "epistemic_confusion": return o.get("epistemic") not in EPISTEMIC or o.get("merged") is True if k == "recommendation_authority": return o.get("grants_authority") is True or o.get("status") == "AUTHORIZATION" if k == "counterfactual_fact": return o.get("is_observed_fact") is True if k == "risk_aggregate": return o.get("aggregate") is not None if k == "ambiguity_permission": return o.get("ambiguity_is_permission") is True if k == "self_healing_authority": return o.get("self_authorized") is True if k == "degradation_authority": return o.get("authority_increased") is True if k == "memory_authority": return o.get("authority") not in (None, "NONE") if k == "loop_self_deploy": return o.get("self_authorized") is True or (o.get("deployed") is True and o.get("reviewed") is False) if k == "redteam_authority": return o.get("production_authority") not in (None, "NONE") if k == "reputation_authority": return o.get("reputation_is_authority") is True if k == "revision_stale": return o.get("stale_state_authorizes") is True if k == "benchmark_override": return o.get("benchmark_overrides_governance") is True if k == "root_single_controller": return o.get("single_controller") is True if k == "r2e_truth": return o.get("imported_as_truth") is True if k == "obligation_satisfied": return o.get("permitted_is_not_satisfied") is False except (KeyError, TypeError): return True return True if __name__ == "__main__": sys.exit(main())