#!/usr/bin/env python3 """Clean-room verifier for the CAIN-42 E38 (portable proof-carrying machine agency) evidence bundle. Imports nothing from CAIN; only `cryptography` and the standard library. It does NOT reuse CAIN's authorization, decision, policy, execution or proof-generation code. From the published proof material, public keys, revocation notices and these rules alone it independently computes, for every proof case, one of VALID / INVALID / INCOMPLETE / STALE / REVOKED / UNKNOWN and requires it to equal the verdict CAIN recorded. It also re-verifies the receipt chain, every transparency-log inclusion proof against the signed checkpoint (RFC 9162, re-implemented here), translation honesty (declared fates, no scope broadening, no status upgrade), selective disclosure, the handshake/manifest/negotiation objects, file hashes and the master signature, and it must REJECT every object in MALICIOUS.json. Malformed material fails closed. python3 verify_e38.py """ from __future__ import annotations import base64 import hashlib import json import re import sys from pathlib import Path from cryptography.exceptions import InvalidSignature from cryptography.hazmat.primitives.asymmetric import ed25519 D = lambda k: "CAIN42/E38-" + k + "/v1" # noqa: E731 UNKNOWN = "UNKNOWN" REQUIRED = ("IDENTITY", "CAPABILITY", "DELEGATION", "POLICY", "RISK", "EVIDENCE", "AUTHORIZATION", "COMMIT", "EXECUTION", "OUTCOME", "ENVIRONMENT", "ENFORCEMENT") ACTION_FIELDS = ("agent_identity", "execution_identity", "runtime_identity", "environment_identity", "capability_set", "authority_grant", "delegation_chain", "policy_version", "policy_digest", "risk_state", "context_digest", "intent_digest", "decision_digest", "evidence_digest", "trajectory_digest", "authorization_token", "e8_commit", "execution_boundary", "execution_receipt", "outcome", "post_execution_evidence", "revocation_state", "timestamp", "governance_epoch", "proof_version", "verification_status") SIM = ("SIMULATED", "PREDICTED", "COUNTERFACTUAL") 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("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 obj_ok(dk, obj, pub) -> bool: try: return digest(D(dk), obj["body"]) == obj["hash"] and obj.get("pub") == pub and \ sig_ok(pub, obj["signature_b64"], D(dk), obj["body"]) except (KeyError, TypeError, Bad): return False # ---------------------------------------------------------------- RFC 9162 (re-implemented) def leaf(data: bytes) -> bytes: return hashlib.sha256(b"\x00" + data).digest() def node(a: bytes, b: bytes) -> bytes: return hashlib.sha256(b"\x01" + a + b).digest() def incl_ok(leaf_hash, index, size, path, root) -> bool: if index >= size: return False fn, sn, r = index, size - 1, leaf_hash for p in path: if sn == 0: return False if fn & 1 or fn == sn: r = node(p, r) if not fn & 1: while fn and not fn & 1: fn >>= 1 sn >>= 1 else: r = node(r, p) fn >>= 1 sn >>= 1 return sn == 0 and r == root # ---------------------------------------------------------------- independent verdict def verdict(case) -> str: """case = {"proofs": {pid: {"kind", "proof"}}, "issuer_pub", "target", "revoked": [...targets], "stale": [...]}""" try: P = case["proofs"] pub = case["issuer_pub"] pid = case["target"] if pid not in P: return "UNKNOWN" revoked = set(case.get("revoked", [])) stale = set(case.get("stale", [])) def valid(x): e = P[x] dk = "ACTION-PROOF" if e["kind"] == "ACTION" else "LAYER-PROOF" p = e["proof"] if not obj_ok(dk, p, pub) or p["body"].get("grants_authority") is not False: return False if dk == "LAYER-PROOF" and p["body"].get("kind") not in REQUIRED + ( "REVOCATION", "CONTINUITY", "MIGRATION", "RECOVERY", "FEDERATION", "CONFORMANCE", "GOVERNABILITY"): return False if dk == "ACTION-PROOF" and any(f not in p["body"] for f in ACTION_FIELDS): return False return True ap = P[pid]["proof"] b = ap["body"] if not valid(pid): return "INVALID" seen, stack, clo = set(), [pid], [] while stack: n = stack.pop() for d in (P[n]["proof"]["body"].get("depends_on", []) if n in P else []): if d not in seen: seen.add(d) clo.append(d) stack.append(d) present = [d for d in clo if d in P] missing = [d for d in clo if d not in P] if any(not valid(d) for d in present): return "INVALID" targets = lambda x: {x, str(P[x]["proof"]["body"].get("subject", "")), str(P[x]["proof"]["body"].get( # noqa "agent_identity", "")), str(P[x]["proof"]["body"].get("execution_identity", "")), str(P[x]["proof"]["body"].get("authorization_token", ""))} if any(targets(x) & revoked for x in present + [pid]): return "REVOKED" color = {} def cyc(n): color[n] = 1 for d in P[n]["proof"]["body"].get("depends_on", []): if d in P and (color.get(d) == 1 or (color.get(d) is None and cyc(d))): return True color[n] = 2 return False if cyc(pid): return "INVALID" byk = {} for d in present: lb = P[d]["proof"]["body"] k = (lb.get("kind"), lb.get("subject"), lb.get("governance_epoch")) if k[0] and k in byk and byk[k] != lb.get("claims_digest"): return "INVALID" byk.setdefault(k, lb.get("claims_digest")) if pid in stale or any(d in stale for d in present): return "STALE" layers = b.get("layer_proofs", {}) for k, v in layers.items(): if v in P: lb = P[v]["proof"]["body"] if lb.get("kind") != k or lb.get("subject") != b.get("agent_identity") or \ lb.get("governance_epoch") != b.get("governance_epoch") or \ lb.get("trust_domain") != b.get("trust_domain"): return "INVALID" bases = {k: P[v]["proof"]["body"]["basis"] for k, v in layers.items() if v in P} if any(x in SIM for x in bases.values()): return "INCOMPLETE" if [k for k in REQUIRED if k not in layers] or missing or \ b.get("e8_commit") in (None, "", UNKNOWN, "NO_E8_COMMIT"): return "INCOMPLETE" refs = set(re.findall(r"\b(lp_[a-z]+_[0-9a-f]{20}|ap_[0-9a-f]{24}|cp_[0-9a-f]{24})\b", json.dumps({k: v for k, v in b.items() if k not in ("depends_on", "layer_proofs")}))) if refs - set(b.get("depends_on", [])) - {pid}: return "INCOMPLETE" # hidden dependency if any(x == "UNVERIFIED" and k != "ENFORCEMENT" for k, x in bases.items()): return "INCOMPLETE" return "VALID" except (KeyError, TypeError, AttributeError): return "INVALID" # malformed material fails closed class Checks: 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}") def load(d, rel): return json.loads((d / rel).read_text()) def main() -> int: d = Path(sys.argv[1] if len(sys.argv) > 1 else ".") c = Checks() pe = load(d, "E38_PROOF_EXAMPLES/PROOF_CASES.json") for case in pe["cases"]: got = verdict(case) c.check(f"case.{case['name']}", got == case["expected"], f"independent={got} expected={case['expected']}") c.check(f"case.{case['name']}.matches_cain", got == case["cain_verdict"], f"independent={got} cain={case['cain_verdict']}") c.check("cases.real_valid", sum(1 for x in pe["cases"] if x["name"].startswith("real-") and verdict(x) == "VALID") >= 4, "real proofs") for pid, e in pe["real"]["proofs"].items(): c.check(f"real.{pid}.sig", obj_ok("ACTION-PROOF" if e["kind"] == "ACTION" else "LAYER-PROOF", e["proof"], pe["real"]["issuer_pub"]), "sig") c.check(f"real.{pid}.no_secret_keys", not ({"api_key", "private_key", "password", "prompt", "chain_of_thought"} & set(json.dumps(e).lower().split('"'))), "secret") for pid in pe["real"]["action_ids"]: b = pe["real"]["proofs"][pid]["proof"]["body"] c.check(f"real.{pid}.e8_bound", b["e8_commit"] not in (UNKNOWN, "", None) and pe["real"]["proofs"][b["layer_proofs"]["COMMIT"]]["proof"]["body"]["claims"]["e8_commit"] == b["e8_commit"], "E8 binding") c.check(f"real.{pid}.not_safety_claim", "not safety" in b["claim"], b["claim"]) rc = load(d, "E38_RECEIPTS/RECEIPT_CHAIN.json") prev = "0" * 64 for i, r in enumerate(rc["chain"]): c.check(f"receipt.{i}.sig", obj_ok("RECEIPT", r, rc["pub"]), "sig") c.check(f"receipt.{i}.chain", r["body"]["seq"] == i and r["body"]["prev"] == prev, "chain") c.check(f"receipt.{i}.no_retro", r["body"]["retroactive_authorization"] is False, "retro") prev = r["hash"] lg = load(d, "E38_TRANSPARENCY_LOG.json") cp = lg["checkpoint"] c.check("log.checkpoint.sig", obj_ok("LOG-HEAD", cp, lg["pub"]), "checkpoint") root = bytes.fromhex(cp["body"]["root"]) for inc in lg["inclusions"]: lf = leaf(cj(inc["entry"])) c.check(f"log.inclusion.{inc['index']}", inc["root"] == cp["body"]["root"] and incl_ok(lf, inc["index"], inc["size"], [bytes.fromhex(x) for x in inc["path"]], root), "inclusion") c.check(f"log.public_only.{inc['index']}", set(inc["entry"]) - {"index"} <= set(lg["public_fields"]), "fields") rv = load(d, "E38_REVOCATION/REVOCATIONS.json") for i, n in enumerate(rv["notices"]): c.check(f"revocation.{i}.sig", obj_ok("REVOCATION", n, rv["pub"]), "sig") m = load(d, "E38_REVOCATION/MEASUREMENT.json") c.check("revocation.not_instantaneous_claim", m["instantaneous_global_revocation"] is False, "claim") c.check("revocation.after_delivery_zero", m["delayed_domain_acceptances_after_delivery"] == 0, "after delivery") c.check("revocation.window_counted", m["delayed_domain_acceptances_before_delivery"] >= 0 and m["post_revocation_acceptances"] == m["delayed_domain_acceptances_before_delivery"], "window") tr = load(d, "E38_TRANSLATION/TRANSLATIONS.json") for t, obj in sorted(tr["translations"].items()): b = obj["body"] c.check(f"translation.{t}.sig", obj_ok("TRANSLATION", obj, tr["pub"]), "sig") lossy = any(v in ("DROPPED", "UNKNOWN") for v in b["fates"].values()) c.check(f"translation.{t}.lossy_declared", b["lossy"] == lossy and b["status"] == ("LOSSY" if lossy else "LOSSLESS"), "lossy") for f, fate in b["fates"].items(): present = b["how"][f] in b["translated"] c.check(f"translation.{t}.{f}.fate", present == (fate in ("PRESERVED", "TRANSFORMED")), fate) src = tr["source"]["scope"] sc = b["translated"].get(b["how"]["scope"], {}) if b["fates"]["scope"] not in ("DROPPED", "UNKNOWN") else {} c.check(f"translation.{t}.no_broadening", all(k in src and all(any(p.startswith(q) for q in src[k]) for p in v) for k, v in sc.items()), "scope") fe = load(d, "E38_FEDERATION/FEDERATION.json") c.check("federation.flow_ok", all(s["ok"] for s in fe["flow_steps"]), [s for s in fe["flow_steps"] if not s["ok"]]) hs = fe["handshake"] c.check("handshake.sig", obj_ok("HANDSHAKE", hs, hs["pub"]), "sig") c.check("handshake.no_authority", hs["body"]["creates_authority"] is False, "authority") for side in ("a", "b"): mf = fe["manifests"][side] c.check(f"manifest.{side}.sig", obj_ok("MANIFEST", mf, mf["pub"]), "sig") c.check(f"manifest.{side}.honest", mf["body"]["trust_score"] is None and mf["body"]["safety_certificate"] is False and mf["body"]["unknowns"], "honesty") c.check("federation.no_merge", fe["recognition"]["authority_merged"] is False, "merged") ds = load(d, "E38_DISCLOSURE.json") pkg = ds["package"] cm = pkg["commitment"] c.check("disclosure.root_sig", sig_ok(ds["pub"], cm["signature_b64"], "CAIN42/E34-SELECTIVE-DISCLOSURE-ROOT/v1", cm["body"]), "root") for x in pkg["disclosed"]: lf = leaf(cj({"field": x["field"], "salt": x["salt"], "value": x["value"]})) c.check(f"disclosure.{x['field']}", incl_ok(lf, x["index"], cm["body"]["n"], [bytes.fromhex(p) for p in x["path"]], bytes.fromhex(cm["body"]["root"])), "inclusion") c.check("disclosure.hides_policy", all(x["field"] != "policy_digest" for x in pkg["disclosed"]), "policy shown") c.check("disclosure.zk_honest", ds["zk"] == "NOT_IMPLEMENTED", ds["zk"]) cf = load(d, "E38_CONFORMANCE.json") c.check("conformance.defects_detected", cf["defects_detected_exactly"] is True, "defects") c.check("conformance.not_real_vendors", "mock" in cf["note"], cf["note"]) iv = load(d, "E38_INVARIANTS.json") c.check("invariants.min", iv["total"] >= 750 and iv["held"] == iv["total"], (iv["held"], iv["total"])) at = load(d, "E38_ATTACKS.json") c.check("attacks.min", at["total"] >= 2000 and at["held"] == at["total"], (at["held"], at["total"])) cg = load(d, "E38_CLAIM_GRAPH.json") for r in cg["claims"]: c.check(f"claimgraph.{r['claim_id']}", (r["status"] == "FULLY_LINKED") == (not r["missing"]), r["missing"]) cl = load(d, "E38_CLAIMS.json") blob = json.dumps(cl).lower() for w in ("controls the internet", "guarantees safety", "universal enforcement", "official internet standard", "certified by", "hardware attestation verified"): c.check(f"claims.no:{w}", w not in blob, w) names = {str(p.relative_to(d)) for p in d.rglob("*") if p.is_file()} for x in cl["claims"]: c.check(f"claim.{x['id']}.evidence", all(e in names for e in x["evidence"]), x["evidence"]) mal = load(d, "MALICIOUS.json")["objects"] for mo in mal: c.check(f"malicious.{mo['class']}", rejected(mo, pe), mo["mutation"]) 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") sg = load(d, "E38_SIGNATURES.json") c.check("signature.files", sg["files"] == sums, "files") c.check("signature.master_digest", sg["master"]["hashes_digest"] == hh(sg["files"]), "digest") c.check("signature.valid", sig_ok(sg["signer_public_key_b64"], sg["signature_b64"], D("MASTER"), sg["master"]), "sig") out = {"verifier": "verify_e38.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(mo, pe) -> bool: k, o = mo["class"], mo["object"] try: if k == "case": return verdict(o) != "VALID" if k == "receipt": return not obj_ok("RECEIPT", o["receipt"], o["pub"]) if k == "translation_upgrade": b = o["body"] return b["translated"].get(b["how"]["verification_status"]) is True and o["source_status"] != "VALID" if k == "log_inclusion": lf = leaf(cj(o["entry"])) return not incl_ok(lf, o["index"], o["size"], [bytes.fromhex(x) for x in o["path"]], bytes.fromhex(o["root"])) if k == "manifest_overclaim": return o["body"].get("trust_score") is not None or o["body"].get("safety_certificate") is not False if k == "handshake_authority": return o["body"].get("creates_authority") is not False if k == "revocation_instant": return o.get("instantaneous_global_revocation") is True except (KeyError, TypeError, ValueError, Bad): return True return True if __name__ == "__main__": sys.exit(main())