#!/usr/bin/env python3 """CAIN-42 E27 clean-room verifier. Imports NOTHING from CAIN. python3 verify_e27.py [bundle-dir] """ 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 D_REGISTRY = "CAIN42/E27-REGISTRY-ENTRY/v1" D_DESCRIPTOR = "CAIN42/E27-SERVICE-DESCRIPTOR/v1" D_PACKET = "CAIN42/E27-GOVERNANCE-PACKET/v1" D_AIP = "CAIN42/E27-AIP-MESSAGE/v1" D_POLICY_DIST = "CAIN42/E27-POLICY-DISTRIBUTION/v1" D_CONTRACT = "CAIN42/E27-MACHINE-CONTRACT/v1" def canonical(o) -> bytes: return json.dumps(o, sort_keys=True, separators=(",", ":"), ensure_ascii=True).encode() def h(o) -> str: return hashlib.sha256(canonical(o)).hexdigest() def digest(domain: str, body) -> str: return h({"domain": domain, "body": body}) def verify(pub: str, sig: str, domain: str, 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): return False class Checker: 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}"[:300]) def main() -> int: d = Path(sys.argv[1]) if len(sys.argv) > 1 else Path(".") c = Checker() must = ["CAIN42_E27_EVIDENCE_BUNDLE.json", "CONTROL_PLANE_RUN_MANIFEST.json", "INVARIANTS.json", "SECURITY_RESULTS.json", "ROUTING_RESULTS.json", "AUTHORIZATION_RESULTS.json", "DELEGATION_RESULTS.json", "REVOCATION_RESULTS.json", "SIMULATION_RESULTS.json", "RESEARCH_RESULTS.json", "EVOLUTION_RESULTS.json", "SCORECARD.json", "SHA256SUMS", "SIGNATURE.json", "MANIFEST.json", "SCHEMAS.json", "KNOWN_LIMITATIONS.json", "UNKNOWN_BOUNDARIES.json", "TEST_RESULTS.json"] for n in must: c.check(f"file_present:{n}", (d / n).exists(), "missing") def load(n): try: return json.loads((d / n).read_text()) except Exception as e: # noqa: BLE001 return {"_error": str(e)} sums = {} for line in (d / "SHA256SUMS").read_text().splitlines() if (d / "SHA256SUMS").exists() else []: parts = line.split(" ", 1) if len(parts) == 2: sums[parts[1]] = parts[0] for n, dig in sums.items(): c.check(f"sha256sums:{n}", (d / n).exists() and hashlib.sha256((d / n).read_bytes()).hexdigest() == dig, "hash mismatch") manifest = load("MANIFEST.json") for n, dig in (manifest.get("files") or {}).items(): c.check(f"manifest:{n}", (d / n).exists() and hashlib.sha256((d / n).read_bytes()).hexdigest() == dig, "manifest hash mismatch") sig = load("SIGNATURE.json") c.check("master:hashes_digest", sig.get("master", {}).get("hashes_digest") == h(sums), "hashes_digest mismatch") c.check("master:signature", verify(sig.get("signer_public_key_b64", ""), sig.get("signature_b64", ""), "CAIN42/E27-MASTER/v1", sig.get("master", {})), "master signature invalid") c.check("master:files_match_sums", sig.get("files") == sums, "files differ from SHA256SUMS") run = load("CONTROL_PLANE_RUN_MANIFEST.json") for i, e in enumerate(run.get("registry", [])): body = e["entry"] c.check(f"registry:{i}:signature", verify(e["issuer_pub"], e["signature_b64"], D_REGISTRY, body), "signature invalid") c.check(f"registry:{i}:no_authority", "authority" not in body, "authority in registry entry") for i, p in enumerate(run.get("packets", [])): body = p["packet"] c.check(f"packet:{i}:id", body.get("packet_id") == "gp_" + digest( D_PACKET, {k: v for k, v in body.items() if k != "packet_id"})[:32], "id mismatch") c.check(f"packet:{i}:signature", verify(p["issuer_pub"], p["signature_b64"], D_PACKET, body), "signature invalid") c.check(f"packet:{i}:bound_transaction", bool(body.get("transaction_id")), "no transaction") c.check(f"packet:{i}:bound_policy", bool(body.get("policy_digest")), "no policy digest") for i, rec in enumerate(run.get("policy_distribution", [])): c.check(f"poldist:{i}:signature", verify(run["key_pub"], rec["signature_b64"], D_POLICY_DIST, {k: v for k, v in rec.items() if k != "signature_b64"}), "signature invalid") c.check(f"poldist:{i}:epoch", rec["epoch"] == i + 1, "epoch not monotonic") contract = run.get("contract", {}) for party, s in (contract.get("signatures") or {}).items(): c.check(f"contract:{party}", verify(run["key_pub"], s, D_CONTRACT, contract.get("fields", {})), "contract signature invalid") for i, m in enumerate(run.get("aip", [])): c.check(f"aip:{i}:signature", verify(run["key_pub"], m["signature_b64"], D_AIP, m["message"]), "aip signature invalid") inv = load("INVARIANTS.json") c.check("invariants:all_hold", inv.get("all_hold") is True, "not all hold") c.check("invariants:count", inv.get("checked", 0) >= 750, "fewer than 750") for i, r in enumerate(inv.get("rows", [])): c.check(f"invariant:{i}:ok", r["ok"] is True, f"{r.get('id')} failed") bench = load("SECURITY_RESULTS.json") c.check("attacks:total", bench.get("total", 0) >= 1500, "fewer than 1500") c.check("attacks:all_contained", bench.get("all_contained") is True, "not all contained") for name, a in (bench.get("attacks") or {}).items(): c.check(f"attack:{name}:contained", a["contained"] is True, "not contained") bare = name changed = True while changed: changed = False for p in ("e27::", "e26::", "e25::"): if bare.startswith(p): bare = bare[len(p):] changed = True c.check(f"attack:{name}:digest", a["digest"] == h({"attack": bare, "contained": a["contained"], "evidence": a["evidence"]}), "digest mismatch") mut = load("MUTATION_RESULTS.json") c.check("mutation:all_caught", mut.get("all_mutants_caught") is True, "mutant not caught") c.check("mutation:controls", mut.get("controls_ok") is True, "control mutated") sc = load("SCORECARD.json") c.check("scorecard:no_aggregate", sc.get("aggregate_score") is None, "aggregate present") for dim, v in (sc.get("dimensions") or {}).items(): c.check(f"scorecard:{dim}", v.get("status") in ("VERIFIED", "PARTIAL", "SIMULATED", "OBSERVED", "NOT IMPLEMENTED", "UNKNOWN"), "bad status") bundle = load("CAIN42_E27_EVIDENCE_BUNDLE.json") c.check("bundle:invariants_match", bundle.get("invariants", {}).get("checked") == inv.get("checked"), "count differs") c.check("bundle:attacks_match", bundle.get("attacks", {}).get("total") == bench.get("total"), "count differs") c.check("bundle:kernel_gate", bundle.get("completion_gates", {}).get("kernel_enforcement") == "NOT IMPLEMENTED", "kernel gate not marked") c.check("bundle:federation_gate", bundle.get("completion_gates", {}).get("identity_federation") == "NOT IMPLEMENTED", "federation gate not marked") c.check("bundle:research_gate", bundle.get("completion_gates", {}).get("research_capabilities") == "NOT IMPLEMENTED", "research gate not marked") c.check("standards:no_compliance_claim", all(v not in ("COMPLIANT", "CERTIFIED") for v in (load("STANDARDS.json").get("standards") or {}).values()), "compliance claimed") c.check("limits:documented", bool(load("KNOWN_LIMITATIONS.json").get("limitations")), "no limitations") c.check("unknowns:published", bool(load("UNKNOWN_BOUNDARIES.json").get("unknowns")), "no unknowns") result = "INTACT" if not c.problems and c.checks >= 500 else "BROKEN" print(json.dumps({"schema": "cain42.e27.verify.v1", "verifier": "verify_e27.py", "imports_cain": False, "result": result, "checks": c.checks, "passed": c.passed, "problems": c.problems[:50]}, indent=1, sort_keys=True)) return 0 if result == "INTACT" else 1 if __name__ == "__main__": sys.exit(main())