#!/usr/bin/env python3 """CAIN-42 E42 verifier v2 — corrected per external review (2026-10-09). Supersedes scripts/cain45/verify_e42.py (kept as historical v1). Zero CAIN imports. Corrections over v1: A the tautological `for s in AUTH_PATH: s in AUTH_PATH` loop is removed; the externally supplied authority trace is validated against an independently declared EXPECTED_AUTH_PATH constant. B ABI versions are checked against an independently declared SUPPORTED_ABI_VERSIONS tuple; missing/unknown versions yield UNSUPPORTED_ABI, never a silent pass. C report assertions (held == scenarios, row["held"] is True, passed == total, killed == mutants, e2e["ok"] is True) are RECORD_CONSISTENCY checks and are never counted as behavioural verification. D unknown fixture classes yield UNSUPPORTED_FIXTURE; malformed harness data yields HARNESS_ERROR; neither counts as an exercised rejection. E coverage classifications are treated as a CLAIM unless linked to deployment configuration / credential placement / boundary / bypass / side-effect evidence; absent that, the enforcement result is NOT_ESTABLISHED. F receipts are not accepted on a non-empty authority string; a structured, signature-verified authority is required, else AUTHORITY_NOT_ESTABLISHED. Offline authority validity is reported separately from live revocation. Every check lands in exactly one category, each with an explicit statement of what passing does and does not establish. A "behavioural pass" is claimed only when a SEMANTIC_RECOMPUTATION / IMPLEMENTATION_REEXECUTION / LIVE_OBSERVATION check passed. python3 verify_e42_v2.py [--json] """ from __future__ import annotations import base64 import hashlib import json import sys from collections import OrderedDict from pathlib import Path from cryptography.exceptions import InvalidSignature from cryptography.hazmat.primitives.asymmetric import ed25519 D_KERNEL = "CAIN42/E42-KERNEL/v1" D_ABI = "CAIN42/E42-ABI/v1" D_RECEIPT = "CAIN42/E42-RECEIPT/v1" D_PUBLISHER = "CAIN42/PUBLISHER-ATTESTATION/v1" # Independently declared expectations (never derived from the input under test). EXPECTED_AUTH_PATH = ("IDENTITY", "CAPABILITY", "TRUST", "POLICY", "AUTHORITY", "RISK", "CONSEQUENCE", "TRAJECTORY", "DECISION", "E8_COMMIT", "ENFORCEMENT") SUPPORTED_ABI_VERSIONS = ("1.0",) ABI_OBJECTS = ("Identity", "Intent", "Action", "Capability", "Policy", "Authority", "Risk", "Evidence", "Authorization", "Execution", "Outcome", "Receipt", "Incident", "Evolution") DECISIONS = ("ALLOW", "DENY", "REQUIRE_APPROVAL", "UNKNOWN") COVERAGE = ("ENFORCED", "MONITORED", "OBSERVED", "BYPASSABLE", "UNKNOWN", "UNCONTROLLED") COVERAGE_STRENGTH = {"ENFORCED": 5, "MONITORED": 4, "OBSERVED": 3, "BYPASSABLE": 2, "UNKNOWN": 1, "UNCONTROLLED": 0} EVIDENCE_CLASSES = ("FORMAL_PROOF", "EMPIRICAL_EVIDENCE", "REPRODUCIBLE_TEST", "MACHINE_VERIFICATION", "INDEPENDENT_REPRODUCTION", "SIMULATION", "OBSERVATION", "CLAIM") RECEIPT_FIELDS = ("who", "what", "why", "policy", "authority", "capability", "environment", "at", "version", "risk", "evidence", "result") PINNED_PUBLISHER_KEY_B64 = "tOh40wrQGGYpF9XrO+Uk0cTj8od4HRyGtGZcEIZwTv8=" # Categories. Only the last three can establish observed BEHAVIOUR. CATEGORIES = OrderedDict([ ("PUBLISHER_AUTHENTICATION", "the bytes were signed by the pinned publisher key"), ("ARTIFACT_INTEGRITY", "the published files match their declared SHA-256"), ("SCHEMA", "fields, enums, hashes and signatures are well-formed"), ("REPORT_CONSISTENCY", "the report's own numbers agree with each other"), ("SEMANTIC_RECOMPUTATION", "a result was recomputed from underlying observations"), ("IMPLEMENTATION_REEXECUTION", "the implementation was run again and observed"), ("LIVE_OBSERVATION", "a live deployment was queried and observed"), ("UNSUPPORTED", "the input is not a supported version/class"), ("HARNESS_ERROR", "the input/harness is malformed"), ]) BEHAVIOURAL = ("SEMANTIC_RECOMPUTATION", "IMPLEMENTATION_REEXECUTION", "LIVE_OBSERVATION") ESTABLISHES = { "PUBLISHER_AUTHENTICATION": "that the holder of the pinned publisher key signed the attestation bytes", "ARTIFACT_INTEGRITY": "that served files are byte-identical to the declared hashes", "SCHEMA": "that fields, enums and signatures are well-formed", "REPORT_CONSISTENCY": "that the report's numbers are self-consistent", "SEMANTIC_RECOMPUTATION": "that a claim was re-derived from raw observations", "IMPLEMENTATION_REEXECUTION": "that the implementation was observed doing the claim", "LIVE_OBSERVATION": "that a live deployment exhibited the behaviour at the stated time", "UNSUPPORTED": "nothing; the input version/class is not supported", "HARNESS_ERROR": "nothing; the input is malformed", } DOES_NOT_ESTABLISH = { "PUBLISHER_AUTHENTICATION": "that the publisher is trustworthy or that the content is true", "ARTIFACT_INTEGRITY": "that the content is semantically correct", "SCHEMA": "that the values are true, only well-formed", "REPORT_CONSISTENCY": "any behaviour; a signed report can be internally consistent and false", "SEMANTIC_RECOMPUTATION": "anything about components not covered by the recomputation", "IMPLEMENTATION_REEXECUTION": "correctness beyond the executed test, or independence of the tester", "LIVE_OBSERVATION": "permanence, or that the observer is independent", "UNSUPPORTED": "nothing", "HARNESS_ERROR": "nothing", } class Bad(Exception): pass def _chk(o, p="$"): if isinstance(o, (bool, type(None), 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 Report: def __init__(self): self.by_cat = {k: {"checks": 0, "pass": 0, "fail": 0, "not_established": 0, "unsupported": 0, "harness_error": 0} for k in CATEGORIES} self.records = [] def add(self, category, name, outcome, detail=""): b = self.by_cat[category] b["checks"] += 1 key = {"PASS": "pass", "FAIL": "fail", "NOT_ESTABLISHED": "not_established", "UNSUPPORTED": "unsupported", "HARNESS_ERROR": "harness_error"}[outcome] b[key] += 1 if outcome != "PASS": self.records.append({"category": category, "name": name, "outcome": outcome, "detail": str(detail)[:200]}) def behavioural_pass(self) -> bool: for cat in BEHAVIOURAL: b = self.by_cat[cat] if b["fail"] or b["not_established"] or b["unsupported"] or b["harness_error"]: return False return any(self.by_cat[c]["pass"] for c in BEHAVIOURAL) def out(self): return { "verifier": "verify_e42_v2.py", "imports_cain": False, "categories": {k: {**v, "establishes": ESTABLISHES[k], "does_not_establish": DOES_NOT_ESTABLISH[k]} for k, v in self.by_cat.items()}, "behavioural_pass": self.behavioural_pass(), "result": "NO_FAILURES" if not any(v["fail"] or v["harness_error"] for v in self.by_cat.values()) else "FAILURES", "non_pass": self.records, } def run_bundle(files: dict) -> Report: r = Report() pf = files.get("PLATFORM.json") if not isinstance(pf, dict): r.add("HARNESS_ERROR", "PLATFORM.json", "HARNESS_ERROR", "missing or not an object") return r # --- PUBLISHER_AUTHENTICATION --- att = files.get("_attestation") if not isinstance(att, dict): r.add("PUBLISHER_AUTHENTICATION", "publisher_attestation", "NOT_ESTABLISHED", "no attestation file") else: body = {k: att.get(k) for k in ("schema", "bundle", "files", "attested_at", "publisher_public_key_b64")} body_digest = hashlib.sha256(cj(body)).hexdigest() r.add("PUBLISHER_AUTHENTICATION", "attestation.schema", "PASS" if att.get("schema") == "cain42.publisher-attestation.v1" else "FAIL", att.get("schema")) r.add("PUBLISHER_AUTHENTICATION", "attestation.pinned_key", "PASS" if att.get("publisher_public_key_b64") == PINNED_PUBLISHER_KEY_B64 else "FAIL", "pinned" if att.get("publisher_public_key_b64") == PINNED_PUBLISHER_KEY_B64 else "unpinned") pub_ok = False try: ed25519.Ed25519PublicKey.from_public_bytes(base64.b64decode(PINNED_PUBLISHER_KEY_B64)).verify( base64.b64decode(att.get("signature_b64", "")), f"{D_PUBLISHER}\x1f{body_digest}".encode()) pub_ok = True except Exception: # noqa: BLE001 pub_ok = False r.add("PUBLISHER_AUTHENTICATION", "attestation.signature", "PASS" if pub_ok else "FAIL", "publisher signature mismatch" if not pub_ok else "ok") # --- ARTIFACT_INTEGRITY --- sums = files.get("SHA256SUMS") if not isinstance(sums, dict): r.add("ARTIFACT_INTEGRITY", "sha256sums", "HARNESS_ERROR", "SHA256SUMS missing") else: for name, hx in sorted(sums.items()): got = hashlib.sha256(files.get("_raw", {}).get(name, b"")).hexdigest() r.add("ARTIFACT_INTEGRITY", f"sha256.{name}", "PASS" if got == hx else "FAIL", "hash") # --- SCHEMA: authority trace against the independently declared path (A) --- supplied_path = pf.get("auth_path") if supplied_path is None: r.add("SCHEMA", "auth_path.present", "FAIL", "no auth_path supplied") else: r.add("SCHEMA", "auth_path.matches_expected", "PASS" if tuple(supplied_path) == EXPECTED_AUTH_PATH else "FAIL", f"{supplied_path} != {list(EXPECTED_AUTH_PATH)}") for i, dec in enumerate(pf.get("kernel_decisions", [])): b = dec.get("body", {}) r.add("SCHEMA", f"decision[{i}].decision_enum", "PASS" if b.get("decision") in DECISIONS else "FAIL", b.get("decision")) r.add("SCHEMA", f"decision[{i}].path_expected", "PASS" if tuple(b.get("path", [])) == EXPECTED_AUTH_PATH else "FAIL", b.get("path")) r.add("SCHEMA", f"decision[{i}].signature", "PASS" if sig_ok(dec.get("pub"), dec.get("signature_b64"), D_KERNEL, b) else "FAIL", "sig") # semantic: recompute the decision from its own stage states r.add("SEMANTIC_RECOMPUTATION", f"decision[{i}].recomputed", "PASS" if _recompute_decision(b.get("states", {})) == b.get("decision") else "FAIL", b) r.add("SCHEMA", f"decision[{i}].grants_no_authority", "PASS" if b.get("grants_authority") is False and b.get("authority") == "NONE" else "FAIL", b.get("authority")) # --- SCHEMA: ABI version against the independently declared support set (B) --- for i, env in enumerate(pf.get("abi_envelopes", [])): b = env.get("body", {}) ver = b.get("abi", env.get("abi_version")) if ver is None or ver not in SUPPORTED_ABI_VERSIONS: r.add("UNSUPPORTED", f"abi[{i}].version", "UNSUPPORTED", f"abi={ver!r} not in {SUPPORTED_ABI_VERSIONS}") else: r.add("SCHEMA", f"abi[{i}].version", "PASS", ver) r.add("SCHEMA", f"abi[{i}].object", "PASS" if b.get("object") in ABI_OBJECTS else "FAIL", b.get("object")) r.add("SCHEMA", f"abi[{i}].hash", "PASS" if dg(D_ABI, b) == env.get("hash") else "FAIL", "hash") r.add("SCHEMA", f"abi[{i}].signature", "PASS" if sig_ok(env.get("pub"), env.get("signature_b64"), D_ABI, b) else "FAIL", "sig") # --- receipts: structured authority required (F) --- for i, rec in enumerate(pf.get("receipts", [])): b = rec.get("body", {}) auth = b.get("authority") r.add("SCHEMA", f"receipt[{i}].signature", "PASS" if sig_ok(rec.get("pub"), rec.get("signature_b64"), D_RECEIPT, b) else "FAIL", "sig") for f in RECEIPT_FIELDS: r.add("SCHEMA", f"receipt[{i}].field.{f}", "PASS" if f in b else "FAIL", f) # Offline authority validity: a non-empty string is NOT sufficient. if isinstance(auth, dict) and auth.get("issuer") and auth.get("signature_b64"): ok = sig_ok(auth.get("issuer_pub", ""), auth.get("signature_b64"), "CAIN42/AUTHORITY/v1", auth) r.add("SEMANTIC_RECOMPUTATION", f"receipt[{i}].authority_offline", "PASS" if ok else "FAIL", "structured authority signature") else: r.add("UNSUPPORTED", f"receipt[{i}].authority_offline", "NOT_ESTABLISHED", f"authority is {auth!r}: no structured, verifiable authority to check") r.add("LIVE_OBSERVATION", f"receipt[{i}].authority_revocation", "NOT_ESTABLISHED", "offline bundle cannot observe live revocation/stateful replay") # --- proofs --- for i, p in enumerate(pf.get("proofs", [])): b = p.get("body", {}) r.add("SCHEMA", f"proof[{i}].evidence_class", "PASS" if b.get("evidence_class") in EVIDENCE_CLASSES else "FAIL", b.get("evidence_class")) r.add("SCHEMA", f"proof[{i}].no_authority", "PASS" if b.get("authority") == "NONE" else "FAIL", b) # --- coverage: label consistency (E), not enforcement --- for i, cov in enumerate(pf.get("coverages", [])): for path, cls in cov.get("map", {}).items(): r.add("SCHEMA", f"coverage[{i}].{path}.class", "PASS" if cls in COVERAGE else "FAIL", cls) r.add("SCHEMA", f"coverage[{i}].no_authority", "PASS" if cov.get("authority") == "NONE" else "FAIL", cov) for path, claimed, actual, ok in pf.get("coverage_claims", []): consistent = (COVERAGE_STRENGTH.get(claimed, -1) <= COVERAGE_STRENGTH.get(actual, -1)) == ok r.add("REPORT_CONSISTENCY", f"coverage_claim.{path}", "PASS" if consistent else "FAIL", (claimed, actual)) # Enforcement is not established by a label in the same report. r.add("UNSUPPORTED", "coverage.enforcement_classification", "NOT_ESTABLISHED", "no linked deployment config / credential placement / boundary / bypass / side-effect evidence") # --- registry --- reg = pf.get("registry", {}) r.add("REPORT_CONSISTENCY", "registry.present", "PASS" if reg.get("present", 0) >= 30 else "FAIL", reg) r.add("REPORT_CONSISTENCY", "registry.total", "PASS" if reg.get("total", 0) >= 30 else "FAIL", reg) # --- empty-array detection: a bundle that omits its subjects cannot pass --- for key, minimum in (("kernel_decisions", 1), ("abi_envelopes", 1), ("receipts", 1), ("proofs", 1), ("coverages", 1), ("coverage_claims", 1)): n = len(pf.get(key, []) or []) r.add("REPORT_CONSISTENCY", f"count.{key}", "PASS" if n >= minimum else "FAIL", f"{n} < {minimum}") # --- malicious fixtures: unknown class / malformed -> not an exercised rejection (D) --- mal = files.get("MALICIOUS.json") if not isinstance(mal, dict): r.add("HARNESS_ERROR", "MALICIOUS.json", "HARNESS_ERROR", "missing") else: known = {"unknown_to_allow", "coverage_upgrade", "receipt_no_authority", "false_completion"} for i, m in enumerate(mal.get("objects", [])): cls = m.get("class") if cls not in known: r.add("UNSUPPORTED", f"malicious[{i}].class", "UNSUPPORTED", f"unknown fixture class {cls!r}") continue try: outcome = _classify_fixture(cls, m) except (KeyError, TypeError, ValueError) as e: # malformed harness data r.add("HARNESS_ERROR", f"malicious[{i}]", "HARNESS_ERROR", repr(e)) continue # The implementation does not "receive" the input here (data-only fixture), # so an expected rejection is a SEMANTIC_RECOMPUTATION of the fixture only. r.add("SEMANTIC_RECOMPUTATION", f"malicious[{i}].{cls}", outcome, m.get("mutation")) missing = {"unknown_to_allow", "coverage_upgrade", "receipt_no_authority", "false_completion"} - \ {m.get("class") for m in mal.get("objects", [])} r.add("REPORT_CONSISTENCY", "malicious.classes_covered", "PASS" if not missing else "FAIL", sorted(missing)) # --- report assertions: REPORT_CONSISTENCY, never behaviour (C) --- sec = files.get("SECURITY_RESULTS.json") if isinstance(sec, dict): r.add("REPORT_CONSISTENCY", "bench.held_eq_scenarios", "PASS" if sec.get("held") == sec.get("scenarios") else "FAIL", f"{sec.get('held')}/{sec.get('scenarios')}") r.add("REPORT_CONSISTENCY", "bench.scenarios_eq_rows", "PASS" if sec.get("scenarios") == len(sec.get("rows", [])) else "FAIL", f"{sec.get('scenarios')} vs {len(sec.get('rows', []))}") for row in sec.get("rows", []): r.add("REPORT_CONSISTENCY", f"bench.row.{row.get('id')}", "PASS" if row.get("held") is True else "FAIL", str(row.get("detail"))[:60]) inv = files.get("INVARIANTS.json") if isinstance(inv, dict): bad = [x.get("id") for x in inv.get("rows", []) if not x.get("ok")] r.add("REPORT_CONSISTENCY", "invariants.passed_eq_total", "PASS" if inv.get("passed") == inv.get("total") else "FAIL", bad[:5]) mut = files.get("MUTATION_RESULTS.json") if isinstance(mut, dict): r.add("REPORT_CONSISTENCY", "mutation.killed_eq_mutants", "PASS" if mut.get("killed") == mut.get("mutants") and not mut.get("survived") else "FAIL", mut.get("survived")) e2e = files.get("END_TO_END.json") if isinstance(e2e, dict): r.add("REPORT_CONSISTENCY", "e2e.ok_flag", "PASS" if e2e.get("ok") is True else "FAIL", "reported flag, not observed") for s in e2e.get("steps", []): r.add("REPORT_CONSISTENCY", f"e2e.{s.get('step')}", "PASS" if s.get("ok") is True else "FAIL", s.get("step")) lim = files.get("KNOWN_LIMITATIONS.json") if isinstance(lim, dict): r.add("REPORT_CONSISTENCY", "limitations.published", "PASS" if len(lim.get("limitations", [])) >= 8 else "FAIL", len(lim.get("limitations", []))) # --- master signature --- sig = files.get("SIGNATURE.json") if isinstance(sig, dict): r.add("REPORT_CONSISTENCY", "signature.files_match_sha256sums", "PASS" if sig.get("files") == sums else "FAIL", "files") r.add("SEMANTIC_RECOMPUTATION", "signature.master_digest", "PASS" if sig.get("master", {}).get("hashes_digest") == hh(sig.get("files", {})) else "FAIL", "digest") r.add("SCHEMA", "signature.valid", "PASS" if sig_ok(sig.get("signer_public_key_b64"), sig.get("signature_b64"), "CAIN42/E42-MASTER/v1", sig.get("master", {})) else "FAIL", "sig") return r def _recompute_decision(states: dict) -> str: blocked = [s for s in EXPECTED_AUTH_PATH[:-2] if states.get(s) in ("DENY", "REVOKED", "EXPIRED", "FAILED")] missing = [s for s in EXPECTED_AUTH_PATH[:-2] if states.get(s) in (None, "UNKNOWN")] if blocked: return "DENY" if missing: return "UNKNOWN" if states.get("RISK") == "REQUIRE_APPROVAL": return "REQUIRE_APPROVAL" if all(states.get(s) in ("ALLOW", "VERIFIED", "ACTIVE") for s in EXPECTED_AUTH_PATH[:-2]): return "ALLOW" return "UNKNOWN" def _classify_fixture(cls, m) -> str: """Return PASS only when the fixture is genuinely rejected by the recomputation.""" o = m["object"] if cls == "unknown_to_allow": return "PASS" if _recompute_decision(o.get("states", {})) != o.get("decision") else "FAIL" if cls == "coverage_upgrade": return "PASS" if COVERAGE_STRENGTH.get(o.get("claimed"), 0) > COVERAGE_STRENGTH.get(o.get("actual"), 0) else "FAIL" if cls == "receipt_no_authority": return "PASS" if o.get("body", {}).get("authority") in (None, "", "UNKNOWN", "NONE") else "FAIL" if cls == "false_completion": return "PASS" if o.get("false_completion") is False and o.get("evidence") in (None, {}, []) else "FAIL" return "FAIL" def load_bundle(d: Path) -> dict: files: dict = {"_raw": {}} names = ["PLATFORM.json", "MALICIOUS.json", "SECURITY_RESULTS.json", "INVARIANTS.json", "MUTATION_RESULTS.json", "END_TO_END.json", "KNOWN_LIMITATIONS.json", "SIGNATURE.json", "SHA256SUMS"] for n in names: p = d / n if not p.is_file(): continue raw = p.read_bytes() files["_raw"][n] = raw if n == "SHA256SUMS": sums = {} for line in raw.decode().splitlines(): if " " in line: hx, name = line.split(" ", 1) sums[name] = hx files["SHA256SUMS"] = sums else: try: files[n] = json.loads(raw) except ValueError: pass for cand in [d / f"{d.name}.attestation.json", d.parent / "_publisher" / f"{d.name}.json", d / "_publisher" / f"{d.name}.json"]: if cand.is_file(): try: files["_attestation"] = json.loads(cand.read_text()) except ValueError: pass break # Read every file named in SHA256SUMS so integrity can be checked on real bytes. for name in list(files.get("SHA256SUMS", {}).keys()): p = d / name if p.is_file(): files["_raw"].setdefault(name, p.read_bytes()) return files def main(argv=None) -> int: argv = argv or sys.argv[1:] d = Path(argv[0]) if argv else Path(".") r = run_bundle(load_bundle(d)) out = r.out() print(json.dumps(out, indent=2)) # Exit 0 only when there are no failures/harness errors. A NOT_ESTABLISHED / # UNSUPPORTED result is reported, not hidden, and does not by itself fail the run. return 0 if out["result"] == "NO_FAILURES" else 1 if __name__ == "__main__": sys.exit(main())