#!/usr/bin/env python3 """Clean-room verifier for the CAIN-42 E37 (Autonomous Execution Mesh) evidence bundle. Imports nothing from CAIN; only `cryptography` and the standard library. From the published artifacts alone it: * re-verifies every signed E37 object (receipts, execution proofs, events, epochs, identities, tokens, passports, plans, BOMs, spawn/compute/mobility records...) with its own canonical-JSON + Ed25519 implementation; * re-walks the receipt chain and the governance event chain and checks each receipt's E8 commit equals the E34 envelope it points to (and that the envelope's own signature holds); * re-derives the weakest-link coverage rule, the migration carried-state rule, the authority-attenuation rule, the most-restrictive conflict rule, the certificate-state rule and the quorum 2f+1 rule from recorded vectors; * checks the host environment passport lists exactly its UNKNOWN fields; * recomputes every file hash and the master signature; * and must REJECT every object in MALICIOUS.json. python3 verify_e37.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 P = "CAIN42/E37-" D = lambda k: P + k + "/v1" # noqa: E731 D_MASTER = D("MASTER") UNKNOWN = "UNKNOWN" COVERAGE = ("ENFORCED", "VERIFIED", "MONITORED", "OBSERVED", "BYPASSABLE", "UNKNOWN") DECISIONS = ("DENY", "REQUIRE_REVIEW", "REQUIRE_APPROVAL", "UNKNOWN", "ALLOW") CARRIED = ("risk", "budget_spent", "revocations", "transaction_limits", "evidence", "reputation", "incident_state") CERT_STATES = ("VERIFIED_GOVERNED", "PARTIALLY_VERIFIED", "SIMULATED", "DEMO_ONLY", "UNVERIFIED", "UNKNOWN", "REVOKED", "SUPERSEDED") class Bad(Exception): pass def _chk(o): if isinstance(o, bool) or o is None or isinstance(o, int): return if isinstance(o, float): raise Bad("float") if isinstance(o, str): if not o.isascii(): raise Bad("non-ascii") return if isinstance(o, list): for v in o: _chk(v) return if isinstance(o, dict): for k, v in o.items(): if not isinstance(k, str) or not k.isascii(): raise Bad("key") _chk(v) return raise Bad("type") 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 digest(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), digest(domain, body).encode()) return True except (InvalidSignature, ValueError, TypeError, Bad): return False def signed_ok(dk, obj, pub=None) -> bool: try: b = obj["body"] if digest(D(dk), b) != obj.get("hash"): return False if pub is not None and obj.get("pub") != pub: return False return sig_ok(pub or obj.get("pub", ""), obj.get("signature_b64", ""), D(dk), b) except (KeyError, TypeError, Bad): return False def subset(child, parent) -> bool: return all(c in parent and all(any(x.startswith(p) for p in parent[c]) for x in v) for c, v in child.items()) def classify(hop) -> str: if hop.get("bypass_known") is True: return "BYPASSABLE" if hop.get("enforcement_probe") is True or hop.get("sealed_e8") is True: return "ENFORCED" if hop.get("signature_verified") is True: return "VERIFIED" if hop.get("monitored") is True: return "MONITORED" if hop.get("observed") is True: return "OBSERVED" return "UNKNOWN" def weakest(path) -> str: rel = [classify(h) for h in path if h.get("relevant", True)] return max(rel, key=COVERAGE.index) if rel else "UNKNOWN" def carried_ok(src, dst) -> bool: for f in CARRIED: if f not in dst: return False a, b = src.get(f), dst[f] if f in ("risk", "budget_spent"): if not isinstance(b, int) or b < a: return False elif f in ("revocations", "evidence"): if not set(a) <= set(b): return False elif f == "transaction_limits": if any(b.get(k, 10 ** 18) > v for k, v in a.items()): return False elif a != b: return False return True def resolve(inputs) -> str: if not inputs: return "UNKNOWN" return min((i["decision"] if i["decision"] in DECISIONS else "UNKNOWN" for i in inputs), key=DECISIONS.index) def cert_state(v) -> str: if v["revoked"]: return "REVOKED" if v["superseded"]: return "SUPERSEDED" if v["simulated"]: return "SIMULATED" if v["demo"]: return "DEMO_ONLY" if not v["verified"] or v["verifier_problems"]: return "UNVERIFIED" return "VERIFIED_GOVERNED" if v["coverage_overall"] in ("ENFORCED", "VERIFIED") else "PARTIALLY_VERIFIED" def quorum_ok(cert, pubs, f) -> bool: good = set() for v in cert.get("votes", []): i = v.get("replica") if not isinstance(i, int) or not 0 <= i < len(pubs): continue if v.get("body") != cert["object"]: continue if sig_ok(pubs[i], v.get("signature_b64", ""), D("QUORUM-VOTE"), v["body"]): good.add(i) return len(good) >= 2 * f + 1 class Checks: def __init__(self): self.checks = 0 self.passed = 0 self.problems = [] def check(self, name, ok, detail=""): self.checks += 1 if ok: self.passed += 1 else: self.problems.append(f"{name}: {detail}") def load(d, n): return json.loads((d / n).read_text()) def verify_receipts(c, ex, prefix="exec"): pub, e34 = ex["mesh_pub"], ex["e34_pub"] envs = {e["hash"]: e for e in ex["envelopes"]} proofs = {p["hash"]: p for p in ex["proofs"]} prev = "0" * 64 executed = 0 for i, r in enumerate(ex["receipts"]): b = r["body"] c.check(f"{prefix}.receipt.{i}.sig", signed_ok("RECEIPT", r, pub), "signature") c.check(f"{prefix}.receipt.{i}.chain", b.get("prev") == prev and b.get("seq") == i, "chain") c.check(f"{prefix}.receipt.{i}.noauth", b.get("grants_authority") is False, "authority") c.check(f"{prefix}.receipt.{i}.unknown_listed", isinstance(b.get("unknown"), list) and len(b["unknown"]) > 0, "unknown") prev = r["hash"] p = proofs.get(b.get("proof")) c.check(f"{prefix}.receipt.{i}.proof_present", p is not None, "proof missing") if p is None: continue pb = p["body"] c.check(f"{prefix}.proof.{i}.sig", signed_ok("EXECUTION-PROOF", p, pub), "signature") c.check(f"{prefix}.proof.{i}.not_safety", pb.get("proof_of_safety") is False and pb.get("grants_authority") is False, "claims") env = envs.get(pb.get("e34_envelope")) c.check(f"{prefix}.proof.{i}.envelope_present", env is not None, "envelope missing") if env is None: continue dom = ex["e34_domains"]["envelope"] if "proof_id" in env["body"] else ex["e34_domains"]["denial"] c.check(f"{prefix}.envelope.{i}.sig", sig_ok(e34, env.get("signature_b64", ""), dom, env["body"]), "e34 sig") if pb.get("executed"): executed += 1 c.check(f"{prefix}.receipt.{i}.e8_bound", env["body"].get("e8_commit_id") == b.get("e8_commit") and b.get("e8_commit") not in ("NO_E8_COMMIT", UNKNOWN, None, ""), "E8 binding") c.check(f"{prefix}.receipt.{i}.executed_state", b.get("what_happened") == "EXECUTED", "state") else: c.check(f"{prefix}.receipt.{i}.denied_no_e8", b.get("e8_commit") == "NO_E8_COMMIT" and str(b.get("what_happened", "")).startswith("REFUSED"), "denial") prev, seen, last = "0" * 64, set(), -1 for i, ev in enumerate(ex["events"]): b = ev["body"] c.check(f"{prefix}.event.{i}.sig", signed_ok("EVENT", ev, pub), "signature") c.check(f"{prefix}.event.{i}.chain", b.get("prev") == prev and b.get("seq") == i and b.get("event_id") not in seen and b.get("timestamp", -1) >= last, "chain") prev, last = ev["hash"], b.get("timestamp", -1) seen.add(b.get("event_id")) return executed def main() -> int: d = Path(sys.argv[1] if len(sys.argv) > 1 else ".") c = Checks() # ---- execution receipts (real governed executions through kernel + E8 + E34) ---- ex = load(d, "E37_EXECUTION_RECEIPTS.json") executed = verify_receipts(c, ex) c.check("exec.some_executed", executed >= 5, executed) c.check("exec.some_denied", any(not p["body"]["executed"] for p in ex["proofs"]), "no denial recorded") # ---- signed samples of every object type ---- tv = load(d, "E37_TEST_VECTORS.json") for name, s in sorted(tv["signed_samples"].items()): c.check(f"sample.{name}.sig", signed_ok(s["domain_key"], s["object"], s["pub"]), "signature") c.check(f"sample.{name}.noauth", s["object"]["body"].get("grants_authority") is False, "authority") for i, v in enumerate(tv["canonical_vectors"]): c.check(f"canonical.{i}", digest(v["domain"], v["body"]) == v["digest"], "digest") for i, v in enumerate(tv["attenuation"]): c.check(f"attenuation.{i}", subset(v["child"], v["parent"]) == v["subset"], v) for i, v in enumerate(tv["conflicts"]): c.check(f"conflict.{i}", resolve(v["inputs"]) == v["decision"], v["decision"]) for i, v in enumerate(tv["certificates"]): c.check(f"certificate.{i}", cert_state(v["inputs"]) == v["state"] and v["state"] in CERT_STATES, v["state"]) # ---- coverage ---- cov = load(d, "E37_COVERAGE.json") for i, m in enumerate(cov["maps"]): c.check(f"coverage.{i}", weakest(m["path"]) == m["overall"], m["overall"]) c.check("coverage.no_upgrade_rule", cov["upgrade_rule"] == "stronger class needs new evidence", "rule") # ---- migrations ---- mt = load(d, "E37_MIGRATION_TESTS.json") for i, m in enumerate(mt["vectors"]): want = carried_ok(m["source_state"], m["dest_state"]) and subset(m["requested_authority"], m["current_authority"]) and \ m["destination_ok"] c.check(f"migration.{i}.rule", want == m["authorized"], f"{m['case']}") c.check(f"migration.{i}.not_fresh", m["fresh_identity"] is False, "fresh identity") for i, m in enumerate(mt["world"]): c.check(f"migration.world.{i}", (m["authorized"] is True) == (m["expect"] == "AUTHORIZED"), m["case"]) for i, t in enumerate(mt["tokens"]): c.check(f"token.{i}.sig", signed_ok("CONTINUITY-TOKEN", t["token"], t["pub"]), "sig") c.check(f"token.{i}.scope_subset", subset(t["token"]["body"]["scope"], t["origin_authority"]) and t["token"]["body"]["origin_authority_digest"] == hh({k: sorted(v) for k, v in t["origin_authority"].items()}), "scope") # ---- quorum ---- q = load(d, "E37_TEST_VECTORS.json")["quorum"] for i, cert in enumerate(q["certificates"]): c.check(f"quorum.{i}", quorum_ok(cert, q["pubs"], q["f"]) == cert["committed"], cert["committed"]) # ---- runtime passports ---- rp = load(d, "E37_RUNTIME_PASSPORTS.json") hp = rp["host_passport"] c.check("passport.sig", signed_ok("ENV-PASSPORT", hp, rp["issuer_pub"]), "sig") b = hp["body"] want_unknown = sorted(f for f in rp["fields"] if b[f] == UNKNOWN or (isinstance(b[f], dict) and UNKNOWN in b[f].values())) c.check("passport.unknown_fields_exact", b["unknown_fields"] == want_unknown, b["unknown_fields"]) c.check("passport.attestation_unknown", b["attestation"] == UNKNOWN, "attestation must stay UNKNOWN") c.check("admission.not_self_claimed", rp["admission"]["self_claim_accepted"] is False, "self claim") c.check("admission.attestation_not_verified", rp["admission"]["verified"].get("attestation") != "VERIFIED", "attestation verified without TEE") # ---- invariants / attacks / chaos / self-test summaries ---- iv = load(d, "E37_INVARIANTS.json") c.check("invariants.sum", sum(g["total"] for g in iv["groups"].values()) == iv["total"], "sum") c.check("invariants.all_held", iv["held"] == iv["total"] and not iv["failures"], iv["failures"][:3]) c.check("invariants.min", iv["total"] >= 1000, iv["total"]) at = load(d, "E37_ATTACKS.json") c.check("attacks.sum", sum(g["total"] for g in at["categories"].values()) == at["total"], "sum") c.check("attacks.all_held", at["held"] == at["total"] and not at["failures"], at["failures"][:3]) ch = load(d, "E37_CHAOS.json") c.check("chaos.all_faults", len(ch["chaos"]["results"]) == 16, len(ch["chaos"]["results"])) c.check("chaos.no_false_allow", ch["chaos"]["aggregate"]["false_allow"] == 0 and ch["chaos"]["aggregate"]["authority_leakage"] == 0, ch["chaos"]["aggregate"]) c.check("selftest.all", ch["self_test"]["passed"] == ch["self_test"]["total"] == 14, ch["self_test"]["passed"]) ad = load(d, "E37_ADAPTERS.json") for p, rec in sorted(ad["adapters"].items()): if rec["status"] == "TESTED": c.check(f"adapter.{p}.tested_has_conformance", rec["conformance"] and all(rec["conformance"]["checks"].values()), p) c.check("adapters.clouds_honest", all(v["status"] != "LIVE" for v in ad["clouds"].values()), "cloud LIVE claim") # ---- claims ---- cl = load(d, "E37_CLAIMS.json") names = {p.name for p in d.iterdir()} for x in cl["claims"]: c.check(f"claim.{x['id']}.evidence", all(e in names for e in x["evidence"]), x["evidence"]) c.check(f"claim.{x['id']}.level", x["level"] in ("IMPLEMENTED", "TESTED", "VERIFIED", "REPRODUCIBLE", "SIMULATED", "ARCHITECTURE", "NOT_VERIFIED", "UNKNOWN"), x["level"]) banned = ("universal control", "perfect security", "guaranteed safety", "certified by", "production deployment", "hardware attestation verified") blob = json.dumps(cl).lower() c.check("claims.no_banned_language", not any(w in blob for w in banned), "banned wording") # ---- malicious objects: every one must be rejected ---- mal = load(d, "MALICIOUS.json")["objects"] for m in mal: c.check(f"malicious.{m['class']}", _rejected(m, ex), m["mutation"]) # ---- 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, "E37_SIGNATURES.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_e37.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, ex) -> bool: k, o = m["class"], m["object"] try: if k in ("tampered_receipt", "receipt_authority_flip"): return not signed_ok("RECEIPT", o, ex["mesh_pub"]) if k == "tampered_proof": return not signed_ok("EXECUTION-PROOF", o, ex["mesh_pub"]) if k == "tampered_event": return not signed_ok("EVENT", o, ex["mesh_pub"]) if k == "receipt_foreign_signer": return not signed_ok("RECEIPT", o, ex["mesh_pub"]) if k == "token_scope_expansion": return not subset(o["token"]["body"]["scope"], o["origin_authority"]) or \ o["token"]["body"]["origin_authority_digest"] != hh({kk: sorted(v) for kk, v in o["origin_authority"].items()}) if k == "token_forged": return not signed_ok("CONTINUITY-TOKEN", o["token"], o["pub"]) if k == "coverage_upgraded": return weakest(o["path"]) != o["overall"] if k == "migration_reset": return not carried_ok(o["source_state"], o["dest_state"]) and o["authorized"] is True if k == "executed_without_e8": return o["executed"] is True and o["e8_commit"] in ("NO_E8_COMMIT", UNKNOWN, None, "") if k == "chain_reordered": return any(r["body"]["seq"] != i for i, r in enumerate(o["receipts"])) or any( o["receipts"][i]["body"]["prev"] != o["receipts"][i - 1]["hash"] for i in range(1, len(o["receipts"]))) if k == "quorum_short": return not quorum_ok(o["certificate"], o["pubs"], o["f"]) if k == "certificate_collapsed": return cert_state(o["inputs"]) != o["state"] if k == "conflict_loosened": return resolve(o["inputs"]) != o["decision"] if k == "attestation_claimed": return o.get("attestation") != UNKNOWN and o.get("tee_present") is False if k == "passport_unknowns_hidden": return o["body"]["unknown_fields"] != sorted( f for f, v in o["body"].items() if v == UNKNOWN and f not in ("unknown_fields",)) except (KeyError, TypeError, IndexError): return True return True if __name__ == "__main__": sys.exit(main())