#!/usr/bin/env python3 """CAIN-42 Auditor Kit verifier: the 42-Second Initial Evidence Verdict. Standard library only. No network access, no third-party packages, no code from the package is executed. The archive is read in memory and never extracted. python3 -I cain42_verify.py KIT.zip --trust-anchor PUBKEY_B64 [--expect-sha256 HEX] [--expect-version X.Y.Z] python3 -I cain42_verify.py KIT.zip --check-result RESULT.json (detect a replayed or altered result) The trust anchor (the publisher's Ed25519 public key) is NEVER taken from the package being verified. You must obtain it out of band; see KEYS_AND_TRUST.md. Verdicts and exit codes (most severe wins): VERIFICATION_ERROR 70 the verifier itself could not complete (bad arguments, unreadable input) FAIL_INTEGRITY 10 bytes, signature, structure, identity or evidence contradict each other UNSUPPORTED_SCHEMA 20 unknown schema version or algorithm (includes schema downgrade) UNTRUSTED_SIGNING_KEY 30 no trust anchor given, signature missing, or signed by a different/revoked key INCOMPLETE_EVIDENCE 40 a mandatory evidence record is missing or of an unrecognised mandatory type STALE_EVIDENCE 50 mandatory evidence is older than the freshness policy allows PASS_INTEGRITY_WITH_WARNINGS 1 every mandatory check passed; warnings need human attention PASS_INTEGRITY 0 every mandatory check passed; no warnings A PASS verdict means only: these are the exact bytes the holder of the trust-anchor key signed, they are internally consistent, and mandatory evidence is present and fresh. It does NOT mean any CAIN-42 claim is true, that anyone independent reviewed it, or that the platform is production ready. See ASSURANCE_MODEL.md. """ from __future__ import annotations import argparse import base64 import gzip import hashlib import io import json import os import re import sys import time import zipfile from datetime import datetime, timezone VERIFIER_VERSION = "8.3.2" RESULT_SCHEMA = "cain42.auditor_kit.verdict.v1" SUPPORTED_MANIFEST_SCHEMAS = {"cain42.auditor_kit.manifest.v1"} SUPPORTED_HASH = {"sha256"} SUPPORTED_SIG = {"ed25519"} MANIFEST_NAME = "RELEASE_MANIFEST.json" SIGNATURE_NAME = "SIGNATURES/RELEASE_MANIFEST.sig.json" # Archive safety limits (applied to declared sizes BEFORE anything is decompressed). MAX_ARCHIVE_BYTES = 64 * 1024 * 1024 MAX_ENTRIES = 400 MAX_ENTRY_BYTES = 16 * 1024 * 1024 MAX_TOTAL_BYTES = 96 * 1024 * 1024 MAX_RATIO = 200 # Freshness ceilings in days. A package may declare a STRICTER policy, never a looser one. FRESHNESS_CEILING_DAYS = { "platform_release_gate_snapshot": 14, "kit_build_record": 30, "quorum_certificate_sample": 90, "merkle_fixture_set": 365, "independent_review_record": 365, "ip_valuation_record": 60, "operational_proof_chain": 365, "fault_campaign_record": 90, } MANDATORY_EVIDENCE = ["kit_build_record", "platform_release_gate_snapshot", "quorum_certificate_sample", "merkle_fixture_set"] KNOWN_EVIDENCE = set(FRESHNESS_CEILING_DAYS) PROVENANCE_LABELS = {"DEVELOPER_SELF_ATTESTED", "DEVELOPER_MEASURED", "TEST_VECTOR", "INDEPENDENTLY_REPRODUCED", "INDEPENDENT_REVIEW"} INDEPENDENT_LABELS = {"INDEPENDENTLY_REPRODUCED", "INDEPENDENT_REVIEW"} GATE_STATUSES = {"PASS", "PASSED_TESTED_SCOPE", "FAIL", "BLOCKED", "NOT_RUN", "STALE", "UNVERIFIED"} CLAIM_STATUSES = {"SUPPORTED_WITHIN_TESTED_SCOPE", "PARTIALLY_SUPPORTED", "UNSUPPORTED", "NOT_VERIFIED", "CONTRADICTED", "BLOCKED", "PENDING_EXTERNAL_REVIEW"} SUPPORTING_STATUSES = {"SUPPORTED_WITHIN_TESTED_SCOPE", "PARTIALLY_SUPPORTED"} SHA_RE = re.compile(r"^[0-9a-f]{64}$") COMMIT_RE = re.compile(r"^[0-9a-f]{40}$") VERSION_RE = re.compile(r"^(0|[1-9][0-9]*)\.(0|[1-9][0-9]*)\.(0|[1-9][0-9]*)$") ID_RE = re.compile(r"^[A-Z][A-Z0-9]*(-[A-Z0-9]+)+$") # Text that tries to instruct a reader (human or AI). Flagged, never obeyed. INJECTION_RE = re.compile( r"(ignore\s+(all\s+)?(previous|prior|above)\s+instructions|disregard\s+(the\s+)?(previous|prior|above)" r"|you\s+are\s+now\s+|new\s+system\s+prompt|system\s+prompt\s*:|report\s+(this\s+)?(as\s+)?(a\s+)?pass" r"|mark\s+(all\s+)?(checks|findings)\s+as\s+(pass|passed|resolved))", re.I) DOC_VERSION_FILES = ("README.md", "QUICKSTART.md", "REPRODUCE.md") ADVERSARIAL_DIRS = ("FIXTURES/", "AI_REVIEW/fixtures/") VERDICT_EXIT = {"PASS_INTEGRITY": 0, "PASS_INTEGRITY_WITH_WARNINGS": 1, "FAIL_INTEGRITY": 10, "UNSUPPORTED_SCHEMA": 20, "UNTRUSTED_SIGNING_KEY": 30, "INCOMPLETE_EVIDENCE": 40, "STALE_EVIDENCE": 50, "VERIFICATION_ERROR": 70} SEVERITY_ORDER = ["VERIFICATION_ERROR", "FAIL_INTEGRITY", "UNSUPPORTED_SCHEMA", "UNTRUSTED_SIGNING_KEY", "INCOMPLETE_EVIDENCE", "STALE_EVIDENCE"] PROVES = [ "The archive bytes hash to the reported archive_sha256 (and to --expect-sha256 when given).", "When signature_valid passes: the release manifest was signed by the private key matching the trust anchor you supplied.", "Every file in the archive is listed in the signed manifest with a matching SHA-256, and nothing is missing or extra.", "Version, archive root and source-commit identifiers inside the package agree with each other.", "Mandatory evidence records exist, parse, are internally consistent and are within the freshness policy.", "The sample quorum certificate carries >= quorum valid Ed25519 signatures from the cluster keys listed in the signed package.", ] DOES_NOT_PROVE = [ "That the trust-anchor key belongs to who you think it does (you must establish that out of band; see KEYS_AND_TRUST.md).", "That any CAIN-42 claim is true. A valid signature binds a statement to a key; it does not make the statement correct.", "That the source commit exists, or that the platform was built from it (the repository is not public; provenance is DECLARED only).", "That anyone independent of the developer reviewed, reproduced or audited anything.", "Consensus safety or liveness of the CAIN-42 cluster, or the current state of any live system.", "That the CAIN-42 platform is production ready. Its release status is a separate record with its own gates.", ] # ---------------------------------------------------------------- Ed25519 (RFC 8032), verification only _P = 2 ** 255 - 19 _L = 2 ** 252 + 27742317777372353535851937790883648493 _D = (-121665 * pow(121666, _P - 2, _P)) % _P _I = pow(2, (_P - 1) // 4, _P) def _recover_x(y, sign): if y >= _P: return None x2 = (y * y - 1) * pow(_D * y * y + 1, _P - 2, _P) if x2 == 0: return None if sign else 0 x = pow(x2, (_P + 3) // 8, _P) if (x * x - x2) % _P != 0: x = x * _I % _P if (x * x - x2) % _P != 0: return None if (x & 1) != sign: x = _P - x return x def _add(a, b): x1, y1, z1, t1 = a x2, y2, z2, t2 = b A = (y1 - x1) * (y2 - x2) % _P B = (y1 + x1) * (y2 + x2) % _P C = 2 * t1 * t2 * _D % _P Dd = 2 * z1 * z2 % _P E, F, G, H = B - A, Dd - C, Dd + C, B + A return (E * F % _P, G * H % _P, F * G % _P, E * H % _P) def _mul(s, pt): q = (0, 1, 1, 0) while s > 0: if s & 1: q = _add(q, pt) pt = _add(pt, pt) s >>= 1 return q def _eq(a, b): x1, y1, z1, _ = a x2, y2, z2, _ = b return (x1 * z2 - x2 * z1) % _P == 0 and (y1 * z2 - y2 * z1) % _P == 0 _GY = 4 * pow(5, _P - 2, _P) % _P _GX = _recover_x(_GY, 0) _G = (_GX, _GY, 1, _GX * _GY % _P) def _decompress(s): if len(s) != 32: return None y = int.from_bytes(s, "little") sign = y >> 255 y &= (1 << 255) - 1 x = _recover_x(y, sign) if x is None: return None return (x, y, 1, x * y % _P) def ed25519_verify(public: bytes, msg: bytes, sig: bytes) -> bool: """Strict RFC 8032 verification: rejects malformed keys, non-canonical S and off-curve points.""" if len(public) != 32 or len(sig) != 64: return False A = _decompress(public) R = _decompress(sig[:32]) if A is None or R is None: return False s = int.from_bytes(sig[32:], "little") if s >= _L: return False h = int.from_bytes(hashlib.sha512(sig[:32] + public + msg).digest(), "little") % _L return _eq(_mul(s, _G), _add(R, _mul(h, A))) # ---------------------------------------------------------------- helpers def sha256_hex(b: bytes) -> str: return hashlib.sha256(b).hexdigest() def canonical(o) -> bytes: return json.dumps(o, sort_keys=True, separators=(",", ":"), ensure_ascii=False).encode("utf-8") class DuplicateKey(ValueError): pass def _no_dupes(pairs): seen = {} for k, v in pairs: if k in seen: raise DuplicateKey(f"duplicate JSON key {k!r}") seen[k] = v return seen def _reject_constant(c): raise ValueError(f"non-standard JSON constant {c}") def strict_json(b: bytes): """Parse UTF-8 JSON rejecting duplicate keys, NaN/Infinity and a byte-order mark.""" if b.startswith(b"\xef\xbb\xbf"): raise ValueError("byte-order mark not allowed") return json.loads(b.decode("utf-8"), object_pairs_hook=_no_dupes, parse_constant=_reject_constant) def parse_utc(s): if not isinstance(s, str) or not s.endswith("Z"): raise ValueError(f"timestamp must be RFC 3339 UTC ending in Z: {s!r}") return datetime.strptime(s, "%Y-%m-%dT%H:%M:%SZ").replace(tzinfo=timezone.utc) def b64_strict(s, n=None): if not isinstance(s, str): raise ValueError("not a string") raw = base64.b64decode(s, validate=True) if n is not None and len(raw) != n: raise ValueError(f"expected {n} bytes, got {len(raw)}") return raw def key_fingerprint(raw_pub: bytes) -> str: return sha256_hex(raw_pub) # ---------------------------------------------------------------- verifier class Verifier: def __init__(self, archive_path, trust_anchor_b64=None, expect_sha256=None, expect_version=None, revoked=(), auditor_keys=(), now=None): self.archive_path = archive_path self.trust_anchor_b64 = trust_anchor_b64 self.expect_sha256 = (expect_sha256 or "").lower() or None self.expect_version = expect_version self.revoked = {r.lower() for r in revoked} self.auditor_keys = list(auditor_keys) self.now = now or datetime.now(timezone.utc).replace(microsecond=0) self.checks = [] self.files = {} self.manifest = None self.root = None self.release = {} # each check: id, status PASS|FAIL|WARN|SKIP, the verdict a failure maps to, and detail def check(self, cid, ok, fail_verdict="FAIL_INTEGRITY", detail="", warn=False): status = "PASS" if ok else ("WARN" if warn else "FAIL") self.checks.append({"id": cid, "status": status, "on_fail": None if ok or warn else fail_verdict, "detail": detail}) return ok def skip(self, cid, reason, verdict): """A skipped mandatory check is never a pass; it carries the verdict it would have guarded.""" self.checks.append({"id": cid, "status": "SKIP", "on_fail": verdict, "detail": reason}) # ------------------------------------------------------------ stage 1: archive def read_archive(self): try: size = os.path.getsize(self.archive_path) except OSError as e: raise RuntimeError(f"cannot read archive: {e}") if not self.check("archive_size_within_limit", size <= MAX_ARCHIVE_BYTES, detail=f"{size} bytes (limit {MAX_ARCHIVE_BYTES})"): return None with open(self.archive_path, "rb") as fh: data = fh.read() self.archive_sha256 = sha256_hex(data) self.archive_bytes = len(data) if self.expect_sha256: self.check("archive_matches_expected_sha256", self.archive_sha256 == self.expect_sha256, detail=f"expected {self.expect_sha256}, got {self.archive_sha256}") else: self.check("archive_matches_expected_sha256", False, warn=True, detail="no --expect-sha256 given: cannot confirm this is the published release") try: zf = zipfile.ZipFile(io.BytesIO(data)) infos = zf.infolist() except Exception as e: self.check("archive_parses", False, detail=f"not a valid ZIP: {e}") return None self.check("archive_parses", True) if not self.check("archive_entry_count", len(infos) <= MAX_ENTRIES, detail=f"{len(infos)} entries (limit {MAX_ENTRIES})"): return None names = [i.filename for i in infos] dupes = sorted({n for n in names if names.count(n) > 1}) self.check("archive_no_duplicate_entries", not dupes, detail=dupes) bad, roots, total = [], set(), 0 for i in infos: n = i.filename mode = (i.external_attr >> 16) & 0o170000 parts = n.split("/") if (n.startswith("/") or "\\" in n or ":" in parts[0] or any(p in ("..", ".") for p in parts) or "\x00" in n or any(ord(c) < 32 for c in n)): bad.append(f"unsafe path: {n!r}") if mode == 0o120000: bad.append(f"symlink entry: {n!r}") if i.flag_bits & 0x1: bad.append(f"encrypted entry: {n!r}") if i.file_size > MAX_ENTRY_BYTES: bad.append(f"entry too large: {n!r} {i.file_size}") if i.compress_size and i.file_size / max(i.compress_size, 1) > MAX_RATIO: bad.append(f"compression ratio too high: {n!r}") if i.compress_type not in (zipfile.ZIP_STORED, zipfile.ZIP_DEFLATED): bad.append(f"unsupported compression: {n!r}") total += i.file_size roots.add(parts[0]) if total > MAX_TOTAL_BYTES: bad.append(f"total uncompressed size {total} exceeds {MAX_TOTAL_BYTES}") if len(roots) != 1: bad.append(f"archive must have exactly one top-level directory, found {sorted(roots)}") if not self.check("archive_entries_safe", not bad, detail=bad): return None self.root = roots.pop() for i in infos: if i.is_dir(): continue try: with zf.open(i) as fh: b = fh.read(MAX_ENTRY_BYTES + 1) except Exception as e: self.check("archive_entry_crc", False, detail=f"{i.filename}: {e}") return None if len(b) > MAX_ENTRY_BYTES: self.check("archive_entries_safe", False, detail=f"{i.filename} expands beyond limit") return None self.files[i.filename[len(self.root) + 1:]] = b self.check("archive_entry_crc", True, detail=f"{len(self.files)} files read and CRC-checked") return True # ------------------------------------------------------------ stage 2: manifest + signature def check_manifest(self): raw = self.files.get(MANIFEST_NAME) if raw is None: self.check("manifest_present", False, "FAIL_INTEGRITY", f"{MANIFEST_NAME} missing") return False try: m = strict_json(raw) except Exception as e: self.check("manifest_parses_strictly", False, detail=str(e)) return False self.check("manifest_parses_strictly", True) if not isinstance(m, dict): self.check("manifest_schema_supported", False, "UNSUPPORTED_SCHEMA", "manifest is not an object") return False if not self.check("manifest_schema_supported", m.get("schema") in SUPPORTED_MANIFEST_SCHEMAS, "UNSUPPORTED_SCHEMA", f"schema={m.get('schema')!r}; supported={sorted(SUPPORTED_MANIFEST_SCHEMAS)}"): return False self.check("manifest_algorithms_supported", m.get("hash_algorithm") in SUPPORTED_HASH and m.get("signature_algorithm") in SUPPORTED_SIG, "UNSUPPORTED_SCHEMA", f"hash={m.get('hash_algorithm')!r} signature={m.get('signature_algorithm')!r}") # One byte representation only: the signed bytes must be the canonical form. self.check("manifest_bytes_canonical", raw == canonical(m), detail="manifest bytes must equal sorted-key compact UTF-8 JSON (no ambiguity in what was signed)") required = {"schema": str, "kit_name": str, "kit_version": str, "archive_root": str, "source_commit": str, "built_at_utc": str, "hash_algorithm": str, "signature_algorithm": str, "signing_key_id": str, "signing_key_fingerprint_sha256": str, "files": list, "evidence": list, "freshness_policy_days": dict, "revoked_key_fingerprints": list} missing = [k for k, t in required.items() if not isinstance(m.get(k), t)] if not self.check("manifest_required_fields", not missing, detail=f"missing or wrong type: {missing}"): return False self.manifest = m self.release = {"kit_name": m["kit_name"], "kit_version": m["kit_version"], "source_commit": m["source_commit"], "built_at_utc": m["built_at_utc"], "signing_key_id": m["signing_key_id"], "signing_key_fingerprint_sha256": m["signing_key_fingerprint_sha256"]} return True def check_signature(self): m = self.manifest raw_manifest = self.files[MANIFEST_NAME] sig_raw = self.files.get(SIGNATURE_NAME) if not self.trust_anchor_b64: self.check("trust_anchor_supplied", False, "UNTRUSTED_SIGNING_KEY", "no --trust-anchor given; the package's own key information is never trusted") anchor = None else: try: anchor = b64_strict(self.trust_anchor_b64, 32) self.check("trust_anchor_supplied", True, detail=f"fingerprint {key_fingerprint(anchor)}") except Exception as e: raise RuntimeError(f"--trust-anchor is not a base64 32-byte Ed25519 public key: {e}") fp = key_fingerprint(anchor) if anchor else None self.release["trust_anchor_fingerprint_sha256"] = fp revoked = self.revoked | {r.lower() for r in m["revoked_key_fingerprints"] if isinstance(r, str)} if fp: self.check("trust_anchor_not_revoked", fp not in revoked, "UNTRUSTED_SIGNING_KEY", "trust anchor appears on a revocation list" if fp in revoked else "") if sig_raw is None: self.check("signature_present", False, "UNTRUSTED_SIGNING_KEY", f"{SIGNATURE_NAME} missing: release is unsigned") self.skip("signature_valid", "no signature to verify", "UNTRUSTED_SIGNING_KEY") return self.check("signature_present", True) try: s = strict_json(sig_raw) sig = b64_strict(s["signature_b64"], 64) alg, signed_fp, signed_sha = s["algorithm"], s["key_fingerprint_sha256"], s["signed_manifest_sha256"] except Exception as e: self.check("signature_record_well_formed", False, detail=f"malformed signature record: {e}") return self.check("signature_record_well_formed", True) if not self.check("signature_algorithm_supported", alg in SUPPORTED_SIG, "UNSUPPORTED_SCHEMA", f"algorithm={alg!r}"): return self.check("signature_binds_this_manifest", signed_sha == sha256_hex(raw_manifest), detail="signature record names a different manifest digest" if signed_sha != sha256_hex(raw_manifest) else "") self.check("manifest_names_signing_key", m["signing_key_fingerprint_sha256"] == signed_fp, detail="manifest and signature record disagree on the signing key") if anchor is None: self.skip("signature_valid", "no trust anchor supplied", "UNTRUSTED_SIGNING_KEY") return if signed_fp != fp: self.check("signature_key_matches_anchor", False, "UNTRUSTED_SIGNING_KEY", f"signed by key {signed_fp}, but your trust anchor is {fp}") self.skip("signature_valid", "signature is from a key you do not trust", "UNTRUSTED_SIGNING_KEY") return self.check("signature_key_matches_anchor", True) self.check("signature_valid", ed25519_verify(anchor, raw_manifest, sig), detail="Ed25519 over the exact RELEASE_MANIFEST.json bytes") # ------------------------------------------------------------ stage 3: contents + identity def check_contents(self): m = self.manifest listed, ids_bad = {}, [] for e in m["files"]: if not (isinstance(e, dict) and isinstance(e.get("path"), str) and SHA_RE.match(str(e.get("sha256", ""))) and isinstance(e.get("bytes"), int)): ids_bad.append(repr(e)[:120]) continue if e["path"] in listed: ids_bad.append(f"duplicate manifest path {e['path']}") listed[e["path"]] = e self.check("manifest_file_entries_valid", not ids_bad, detail=ids_bad) for p in (MANIFEST_NAME, SIGNATURE_NAME): if p in listed: ids_bad.append(p) self.check("manifest_does_not_list_itself", False, detail=f"{p} must not appear in files[]") actual = set(self.files) - {MANIFEST_NAME, SIGNATURE_NAME} missing = sorted(set(listed) - actual) extra = sorted(actual - set(listed)) self.check("no_missing_files", not missing, detail=missing) self.check("no_unlisted_files", not extra, detail=extra) mism = [] for p, e in sorted(listed.items()): b = self.files.get(p) if b is not None and (sha256_hex(b) != e["sha256"] or len(b) != e["bytes"]): mism.append(p) self.check("file_hashes_match_manifest", not mism, detail=mism) sums = self.files.get("SHA256SUMS") if sums is not None: want = {p: e["sha256"] for p, e in listed.items() if p != "SHA256SUMS"} got = {} for line in sums.decode("utf-8", "replace").splitlines(): if line.strip(): h, _, name = line.partition(" ") got[name] = h self.check("sha256sums_agrees_with_manifest", got == want, detail="SHA256SUMS disagrees with RELEASE_MANIFEST.json" if got != want else "") # identity v = m["kit_version"] self.check("kit_version_well_formed", bool(VERSION_RE.match(v)), detail=v) self.check("archive_root_matches_version", self.root == m["archive_root"] == f"{m['kit_name']}-{v}", detail=f"archive root {self.root!r}, manifest archive_root {m['archive_root']!r}, version {v!r}") if self.expect_version: self.check("version_is_expected_release", v == self.expect_version, detail=f"expected {self.expect_version}, package says {v}") self.check("source_commit_well_formed", bool(COMMIT_RE.match(m["source_commit"])), detail=m["source_commit"]) try: built = parse_utc(m["built_at_utc"]) self.check("build_time_not_in_future", built <= self.now, detail=m["built_at_utc"]) except Exception as e: self.check("build_time_not_in_future", False, detail=str(e)) vf = self.files.get("VERIFIER/cain42_verify.py") mine = None try: with open(os.path.abspath(__file__), "rb") as fh: mine = sha256_hex(fh.read()) except OSError: pass self.release["running_verifier_sha256"] = mine if vf is not None: self.check("embedded_verifier_matches_running_verifier", sha256_hex(vf) == mine, warn=True, detail="you are running a different verifier than the one in the package (fine if you obtained it independently)" if sha256_hex(vf) != mine else "") stale_refs = [] for doc in DOC_VERSION_FILES: t = self.files.get(doc, b"").decode("utf-8", "replace") for found in set(re.findall(r"cain42-auditor-kit-(\d+\.\d+\.\d+)", t)): if found != v: stale_refs.append(f"{doc} references kit {found}") if "cain42-verify.zip" in t: stale_refs.append(f"{doc} references the superseded cain42-verify.zip") self.check("entry_docs_reference_current_release", not stale_refs, detail=stale_refs) # ------------------------------------------------------------ stage 4: evidence def load(self, path): b = self.files.get(path) if b is None: raise ValueError(f"{path} not in package") return strict_json(b) def check_evidence(self): m = self.manifest policy = {} for t, ceiling in FRESHNESS_CEILING_DAYS.items(): declared = m["freshness_policy_days"].get(t, ceiling) policy[t] = min(ceiling, declared) if isinstance(declared, (int, float)) and declared > 0 else ceiling records, problems, seen_ids = {}, [], set() for r in m["evidence"]: if not isinstance(r, dict): problems.append("non-object evidence record") continue rid = r.get("id") if not (isinstance(rid, str) and ID_RE.match(rid)): problems.append(f"bad evidence id {rid!r}") continue if rid in seen_ids: problems.append(f"duplicate evidence id {rid}") seen_ids.add(rid) if r.get("provenance") not in PROVENANCE_LABELS: problems.append(f"{rid}: unknown provenance label {r.get('provenance')!r}") if r.get("path") not in self.files: problems.append(f"{rid}: evidence file {r.get('path')!r} not in package") if r.get("source_commit") != m["source_commit"]: problems.append(f"{rid}: source_commit {r.get('source_commit')!r} != release {m['source_commit']}") records.setdefault(r.get("type"), []).append(r) self.check("evidence_index_consistent", not problems, detail=problems) self.evidence_by_id = {r["id"]: r for t in records.values() for r in t if isinstance(r, dict) and "id" in r} unknown_mandatory = [r.get("id") for t, rs in records.items() for r in rs if t not in KNOWN_EVIDENCE and r.get("mandatory")] unknown_optional = [r.get("id") for t, rs in records.items() for r in rs if t not in KNOWN_EVIDENCE and not r.get("mandatory")] self.check("no_unrecognized_mandatory_evidence", not unknown_mandatory, "INCOMPLETE_EVIDENCE", detail=f"mandatory evidence of a type this verifier cannot evaluate: {unknown_mandatory}") if unknown_optional: self.check("unrecognized_optional_evidence", False, warn=True, detail=unknown_optional) for t in MANDATORY_EVIDENCE: self.check(f"evidence_present:{t}", t in records, "INCOMPLETE_EVIDENCE", detail="" if t in records else f"no {t} record in the manifest") for t, rs in sorted(records.items(), key=lambda kv: str(kv[0])): if t not in KNOWN_EVIDENCE: continue for r in rs: rid = r.get("id") try: obs = parse_utc(r.get("observed_at_utc")) age = (self.now - obs).total_seconds() / 86400 if obs > self.now: self.check(f"fresh:{rid}", False, detail=f"observed_at {r['observed_at_utc']} is in the future") else: self.check(f"fresh:{rid}", age <= policy[t], "STALE_EVIDENCE", detail=f"{age:.1f} days old; policy for {t} is {policy[t]} days") except Exception as e: self.check(f"fresh:{rid}", False, detail=f"bad observed_at_utc: {e}") try: getattr(self, "ev_" + t)(r) except Exception as e: self.check(f"evidence_valid:{rid}", False, detail=f"{type(e).__name__}: {e}") def ev_kit_build_record(self, r): d = self.load(r["path"]) m = self.manifest bad = [k for k in ("kit_version", "source_commit", "kit_name") if d.get(k) != m.get(k)] self.check(f"evidence_valid:{r['id']}", not bad, detail=f"build record disagrees with manifest on {bad}" if bad else "") def ev_platform_release_gate_snapshot(self, r): d = self.load(r["path"]) gates = d["gates"] problems = [] ids = [g["id"] for g in gates] if len(ids) != len(set(ids)): problems.append("duplicate gate ids") for g in gates: if g["status"] not in GATE_STATUSES: problems.append(f"{g['id']}: unknown status {g['status']!r}") # Rules of scripts/ironclad/release_gate_engine.py: any FAIL -> FAILED; else any BLOCKED -> BLOCKED; else any # NOT_RUN/UNVERIFIED/STALE -> INCOMPLETE; else PASSED. production_ready needs PASSED and every gate passing. st = [g["status"] for g in gates] if not isinstance(d["production_ready"], bool) or not gates: problems.append("production_ready must be a boolean and gates must be non-empty") expected = ("RELEASE_GATE_FAILED" if "FAIL" in st else "RELEASE_GATE_BLOCKED" if "BLOCKED" in st else "RELEASE_GATE_INCOMPLETE" if set(st) & {"NOT_RUN", "UNVERIFIED", "STALE"} else "RELEASE_GATE_PASSED") overall = d.get("overall_gate_status") if overall != expected: problems.append(f"overall_gate_status={overall!r} but gate statuses imply {expected}") if d["production_ready"] is True and (expected != "RELEASE_GATE_PASSED" or any(s not in ("PASS", "PASSED_TESTED_SCOPE") for s in st)): problems.append(f"production_ready=true contradicts gate statuses { {g['id']: g['status'] for g in gates} }") self.platform = {"production_ready": d["production_ready"], "overall": overall, "gates": {g["id"]: g["status"] for g in gates}} self.check(f"evidence_valid:{r['id']}", not problems, detail=problems) def ev_quorum_certificate_sample(self, r): cert = self.load(r["path"]) keys_doc = self.load(r["cluster_keys_path"]) if keys_doc.get("cluster_id") != cert.get("cluster_id"): raise ValueError("cluster key set and certificate name different clusters") pub = {k: b64_strict(v, 32) for k, v in keys_doc["keys"].items()} n = len(pub) f = (n - 1) // 3 quorum = 2 * f + 1 valid = set() for sender, msg in sorted(cert["individual_signatures"].items()): if sender not in pub or msg.get("sender_id") != sender or msg.get("cluster_id") != cert["cluster_id"]: continue payload = {"msg_type": msg["msg_type"], "sender_id": msg["sender_id"], "cluster_id": msg["cluster_id"], "view": msg["view"], "sequence": msg["sequence"], "request_digest": msg["request_digest"], "payload": msg["payload"], "epoch": msg["epoch"], "nonce": msg["nonce"], "timestamp": round(msg["timestamp"], 4), "engine_version": msg["engine_version"]} digest = sha256_hex(json.dumps(payload, sort_keys=True, separators=(",", ":"), default=str).encode()).encode() if ed25519_verify(pub[sender], digest, b64_strict(msg["signature_b64"], 64)): valid.add(sender) self.check(f"evidence_valid:{r['id']}", len(valid) >= quorum, detail=f"{len(valid)} valid distinct signers {sorted(valid)}; quorum {quorum} of n={n} (f={f})") def ev_merkle_fixture_set(self, r): d = self.load(r["path"]) wrong = [] for c in d["cases"]: if c["kind"] == "inclusion": got = _merkle_inclusion(bytes.fromhex(c["leaf"]), c["index"], c["tree_size"], [bytes.fromhex(x) for x in c["proof"]], bytes.fromhex(c["root"])) elif c["kind"] == "consistency": got = _merkle_consistency(c["old_size"], c["new_size"], bytes.fromhex(c["old_root"]), bytes.fromhex(c["new_root"]), [bytes.fromhex(x) for x in c["proof"]]) else: raise ValueError(f"unsupported Merkle case kind {c['kind']!r}") if got != c["expect"]: wrong.append(c["id"]) self.check(f"evidence_valid:{r['id']}", not wrong and bool(d["cases"]), detail=f"{len(d['cases'])} RFC 6962 cases; mismatches {wrong}") def ev_ip_valuation_record(self, r): """Recompute the IP register arithmetic and the company / founder figures derived from it.""" reg = self.load(r["path"]) comp = self.load(r["company_record_path"]) problems, ip_base = [], {} for sc, c in sorted(reg["combined"].items()): disc = reg["deployed_discounts"][sc] net = round(c["prompt45_gross"] * (1 - disc)) if c["deployed_discount"] != disc or reg["prompt45_gross"][sc] != c["prompt45_gross"]: problems.append(f"{sc}: discount/gross inputs disagree inside the register") if net != c["prompt45_new_net"]: problems.append(f"{sc}: gross {c['prompt45_gross']} x (1-{disc}) = {net}, register says {c['prompt45_new_net']}") if c["prompt23_base"] + c["prompt45_new_net"] != c["total_net"]: problems.append(f"{sc}: {c['prompt23_base']} + {c['prompt45_new_net']} != total {c['total_net']}") ip_base[sc] = c["total_net"] if comp["ip_base"] != ip_base: problems.append(f"company record ip_base {comp['ip_base']} != register totals {ip_base}") company, founder = {}, {} share = comp["founder_paper_equity"]["ownership_percentage"] / 100.0 for sc, v in comp["company_value"].items(): company[sc] = {"investor": v["investor"], "strategic": v["strategic"]} for basis in ("investor", "strategic"): if ip_base.get(sc, 0) + v["premium_over_ip"][basis] != v[basis]: problems.append(f"{sc}/{basis}: ip_base {ip_base.get(sc)} + premium {v['premium_over_ip'][basis]} != {v[basis]}") fe = comp["founder_paper_equity"][f"{sc.lower()}_scenario"] founder[sc] = {"investor": fe["investor"], "strategic": fe["strategic"]} for basis in ("investor", "strategic"): if round(v[basis] * share) != fe[basis]: problems.append(f"{sc}/{basis}: founder equity {fe[basis]} != {share:.0%} of company value {v[basis]}") self.ip = {"ip_base": ip_base, "company": company, "founder": founder, "founder_name": comp["founder_paper_equity"].get("founder_name"), "disclaimer": comp.get("disclaimer", "")[:160]} self.check(f"evidence_valid:{r['id']}", not problems, detail=problems or f"arithmetic reproduces: IP base {ip_base}; company BASE {company.get('BASE')}; " f"founder share {share:.0%} (internal analytical scenario, not an appraisal)") def ev_operational_proof_chain(self, r): """Verify every signed hourly proof: hash, Ed25519 signature, hash chain, replica agreement; measure the longest unbroken OPERATIONAL streak. (Embedded quorum certificates are re-checked by the published full-chain verifier named in chain_meta.json.)""" meta = self.load(r["meta_path"]) key = b64_strict(meta["evidence_root_public_key_b64"], 32) canon_a = lambda o: json.dumps(o, sort_keys=True, separators=(",", ":"), ensure_ascii=True).encode() # noqa: E731 prev = sha256_hex(canon_a({"domain": meta["domain"], "genesis": meta["cluster_id"]})) lines = [l for l in self.files[r["path"]].decode("utf-8").splitlines() if l.strip()] problems, best, start, last_at, start_seq, best_seq = [], (0.0, None, None, 0), None, None, None, None for i, line in enumerate(lines): p = strict_json(line.encode()) b = p["body"] if sha256_hex(canon_a(b)) != p["proof_hash"]: problems.append(f"proof {i}: proof_hash does not match content") if not ed25519_verify(key, p["proof_hash"].encode(), b64_strict(p["evidence_root_signature_b64"], 64)): problems.append(f"proof {i}: evidence-root signature invalid") if b["n"] != i or b["previous_proof_hash"] != prev or b["cluster_id"] != meta["cluster_id"] \ or b["configuration_hash"] != meta["configuration_hash"]: problems.append(f"proof {i}: hash chain / cluster / membership mismatch") ok = b["verdict"] == "OPERATIONAL" if ok: up = [x for x in b["replicas"].values() if x.get("reachable")] by_h = {} for x in up: by_h.setdefault(x["commit_index"], set()).add(x["application_state_hash"]) if len(up) < 3 or any(len(v) > 1 for v in by_h.values()) or not b["write"].get("committed"): problems.append(f"proof {i}: OPERATIONAL but replicas disagree or the write did not commit") at = parse_utc(b["at"]) if ok and start is not None and last_at is not None and (at - last_at).total_seconds() <= 5400: pass elif ok: start, start_seq = at, b["write"].get("sequence") else: start = None if ok and start is not None: h = (at - start).total_seconds() / 3600 if h > best[0]: best = (h, start, at, i) best_seq = (start_seq, b["write"].get("sequence")) last_at = at if ok else None prev = p["proof_hash"] self.ops = {"cluster_id": meta["cluster_id"], "longest_hours": round(best[0], 1), "proofs": len(lines), "from": best[1].strftime("%Y-%m-%dT%H:%M:%SZ") if best[1] else None, "to": best[2].strftime("%Y-%m-%dT%H:%M:%SZ") if best[2] else None, "commit_sequence": best_seq} need = r.get("claimed_min_hours", 72) if best[0] < need: problems.append(f"longest unbroken OPERATIONAL streak is {best[0]:.1f} h, below the claimed {need} h") self.check(f"evidence_valid:{r['id']}", not problems, detail=problems[:10] or f"{len(lines)} signed proofs verified; longest unbroken OPERATIONAL streak " f"{best[0]:.1f} h ({self.ops['from']} -> {self.ops['to']}), commit sequence " f"{best_seq[0]} -> {best_seq[1]} on {meta['cluster_id']}") def ev_independent_review_record(self, r): """An independence claim must be signed by a reviewer key the auditor supplied out of band.""" d = self.load(r["path"]) body = {k: v for k, v in d.items() if k != "reviewer_signature_b64"} ok = False for k in self.auditor_keys: try: if ed25519_verify(b64_strict(k, 32), canonical(body), b64_strict(d.get("reviewer_signature_b64", ""), 64)): ok = True except Exception: pass self.check(f"evidence_valid:{r['id']}", ok, detail="independence claim verified under a supplied --auditor-key" if ok else "record claims independent review but no --auditor-key verifies its signature: treated as forged") # ------------------------------------------------------------ stage 5: claims def ev_fault_campaign_record(self, r): """The published fault-campaign summary must follow from the raw per-trial records in the package. Nothing is assumed about the outcome: a campaign with failing trials verifies if its summary says so.""" published = self.load(r["path"]) tp = r.get("trials_path") blob = self.files.get(tp) if blob is None: raise ValueError(f"trials file {tp!r} not in package") if str(tp).endswith(".gz"): with gzip.GzipFile(fileobj=io.BytesIO(blob)) as fh: raw = fh.read(MAX_TRIALS_BYTES + 1) else: raw = blob if len(raw) > MAX_TRIALS_BYTES: raise ValueError(f"trial records exceed {MAX_TRIALS_BYTES} bytes") mine = _fc_summarize(raw) diffs = list(mine["record_mismatches"][:5]) for k in sorted(set(mine) | (set(published) - {"run"})): if canonical(mine.get(k)) != canonical(published.get(k)): diffs.append(f"summary field {k!r} differs from the value recomputed from the records") t = mine["totals"] self.check(f"evidence_valid:{r['id']}", not diffs, detail=diffs or (f"summary recomputed from {t['all_trials']} trial records and it matches: " f"{t['scenarios']} scenarios / {t['trials']} trials, PASS {t['PASS']}, FAIL {t['FAIL']}, " f"ERROR {t['ERROR']}, safety violations {t['safety_violations']}; must-fail controls " f"{t['must_fail_scenarios']} / {t['must_fail_trials']} trials, FAIL {t['must_fail_FAIL']}; " f"campaign verdict {mine['campaign_verdict']}. This shows the summary follows from " f"the records, not that the records describe a real run.")) def check_claims(self): try: cm = self.load("CLAIM_EVIDENCE_MATRIX.json") except Exception as e: self.check("claim_matrix_parses", False, detail=str(e)) return self.check("claim_matrix_parses", True) problems, warnings, seen = [], [], set() for c in cm.get("claims", []): cid = c.get("id") if cid in seen: problems.append(f"duplicate claim id {cid}") seen.add(cid) st = c.get("status") if st not in CLAIM_STATUSES: problems.append(f"{cid}: unknown status {st!r}") ev = c.get("evidence_ids", []) ghost = [e for e in ev if e not in self.evidence_by_id] if ghost: problems.append(f"{cid}: cites evidence that does not exist {ghost}") if st in SUPPORTING_STATUSES and not ev: problems.append(f"{cid}: status {st} with no evidence") if not c.get("cannot_establish"): warnings.append(f"{cid}: does not state what its procedure cannot establish") if c.get("provenance") in INDEPENDENT_LABELS and not any( self.evidence_by_id.get(e, {}).get("type") == "independent_review_record" for e in ev): problems.append(f"{cid}: labelled {c.get('provenance')} without an independent_review_record") a = c.get("assertion") if isinstance(a, dict): problems.extend(self._contradiction(cid, st, a)) self.check("claims_traceable_and_uncontradicted", not problems, detail=problems) if warnings: self.check("claims_state_limits", False, warn=True, detail=warnings) def _contradiction(self, cid, status, a): plat = getattr(self, "platform", None) kind = a.get("kind") asserted = status not in {"UNSUPPORTED", "CONTRADICTED", "NOT_VERIFIED", "BLOCKED", "PENDING_EXTERNAL_REVIEW"} out = [] if kind in ("platform_production_ready", "gate_status") and plat is None: out.append(f"{cid}: asserts {kind} but no valid platform gate snapshot is available to check it against") elif kind == "platform_production_ready": if asserted and a.get("value") != plat["production_ready"]: out.append(f"{cid}: signed claim production_ready={a.get('value')} contradicts gate snapshot " f"production_ready={plat['production_ready']}") elif kind == "gate_status": actual = plat["gates"].get(a.get("gate")) if asserted and actual != a.get("value"): out.append(f"{cid}: claims {a.get('gate')}={a.get('value')} but snapshot says {actual}") elif kind == "kit_version": if asserted and a.get("value") != self.manifest["kit_version"]: out.append(f"{cid}: claims kit version {a.get('value')} but manifest says {self.manifest['kit_version']}") elif kind in ("ip_base_value", "company_value", "founder_value"): ip = getattr(self, "ip", None) if ip is None: out.append(f"{cid}: asserts {kind} but no verified ip_valuation_record is available to check it against") else: sc = a.get("scenario", "BASE") actual = ip["ip_base"].get(sc) if kind == "ip_base_value" else \ ip[{"company_value": "company", "founder_value": "founder"}[kind]].get(sc, {}).get(a.get("basis", "investor")) if asserted and actual != a.get("value"): out.append(f"{cid}: claims {kind} {sc}/{a.get('basis', '')} = {a.get('value')} but the verified record says {actual}") elif kind == "continuous_operation_hours": ops = getattr(self, "ops", None) if ops is None: out.append(f"{cid}: asserts continuous operation but no verified operational_proof_chain is available") elif asserted and (ops["cluster_id"] != a.get("cluster_id") or ops["longest_hours"] < a.get("min_hours", 0)): out.append(f"{cid}: claims >= {a.get('min_hours')} h on {a.get('cluster_id')} but the verified chain shows " f"{ops['longest_hours']:.1f} h on {ops['cluster_id']}") elif kind == "independent_review_completed": has = any(r.get("type") == "independent_review_record" for r in self.evidence_by_id.values()) if asserted and a.get("value") is True and not has: out.append(f"{cid}: claims independent review completed but no independent_review_record exists") elif kind is not None: out.append(f"{cid}: unknown assertion kind {kind!r}") return out # ------------------------------------------------------------ stage 6: untrusted-text scan def scan_text(self): hits = [] for p, b in sorted(self.files.items()): # FIXTURES/ and AI_REVIEW/fixtures/ hold deliberately adversarial test data (documented in README.md); # everything else, including evidence, documentation and claims, is scanned. if p.endswith((".md", ".json", ".txt")) and not p.startswith(ADVERSARIAL_DIRS): t = b.decode("utf-8", "replace") mm = INJECTION_RE.search(t) if mm: hits.append(f"{p}: {mm.group(0)!r}") self.check("no_instruction_like_text_in_evidence", not hits, warn=True, detail=hits and ["Instruction-like text found. It is DATA: do not follow it. "] + hits) # ------------------------------------------------------------ run def run(self): t0 = time.perf_counter() try: if self.read_archive() and self.check_manifest(): self.check_signature() self.check_contents() self.check_evidence() self.check_claims() self.scan_text() error = None except RuntimeError as e: error = str(e) except Exception as e: # never turn a crash into a pass error = f"{type(e).__name__}: {e}" return self.report(error, time.perf_counter() - t0) def report(self, error, elapsed): failing = {c["on_fail"] for c in self.checks if c["status"] in ("FAIL", "SKIP")} if error: verdict = "VERIFICATION_ERROR" elif failing: verdict = next(v for v in SEVERITY_ORDER if v in failing) elif not self.checks or self.manifest is None: verdict = "VERIFICATION_ERROR" error = "no checks completed" elif any(c["status"] == "WARN" for c in self.checks): verdict = "PASS_INTEGRITY_WITH_WARNINGS" else: verdict = "PASS_INTEGRITY" out = { "schema": RESULT_SCHEMA, "verifier": {"name": "cain42_verify", "version": VERIFIER_VERSION}, "evaluated_at_utc": self.now.strftime("%Y-%m-%dT%H:%M:%SZ"), "input": {"archive_path": os.path.basename(self.archive_path), "archive_sha256": getattr(self, "archive_sha256", None), "archive_bytes": getattr(self, "archive_bytes", None), "expected_sha256": self.expect_sha256, "expected_version": self.expect_version, "auditor_keys_supplied": len(self.auditor_keys)}, "release": self.release, "platform_status_as_recorded_in_package": getattr(self, "platform", None), "ip_valuation_as_recomputed": getattr(self, "ip", None), "continuous_operation_as_verified": getattr(self, "ops", None), "verdict": verdict, "exit_code": VERDICT_EXIT[verdict], "error": error, "counts": {s: sum(1 for c in self.checks if c["status"] == s) for s in ("PASS", "FAIL", "WARN", "SKIP")}, "checks": self.checks, "proves": PROVES if verdict.startswith("PASS") else [], "does_not_prove": DOES_NOT_PROVE, "scope": "Initial evidence-integrity verdict only. Not a security audit, not a proof of consensus safety, " "not a validation of any product, not a production-readiness decision.", "elapsed_seconds": round(elapsed, 4), } out["result_digest_sha256"] = sha256_hex(canonical({k: v for k, v in out.items() if k != "elapsed_seconds"})) return out # ---------------------------------------------------------------- RFC 6962 Merkle proofs def _leaf(d): return hashlib.sha256(b"\x00" + d).digest() def _node(a, b): return hashlib.sha256(b"\x01" + a + b).digest() def _merkle_inclusion(leaf, index, size, proof, root): if index >= size: return False fn, sn, r = index, size - 1, _leaf(leaf) for p in proof: if sn == 0: return False if fn & 1 or fn == sn: r = _node(p, r) 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 def _merkle_consistency(old, new, old_root, new_root, proof): if old == 0 or old > new: return False if old == new: return old_root == new_root and not proof p = list(proof) if not p: return False if old & (old - 1) == 0: p = [old_root] + p fn, sn = old - 1, new - 1 while fn & 1: fn >>= 1 sn >>= 1 fr = sr = p[0] for c in p[1:]: if sn == 0: return False if fn & 1 or fn == sn: fr, sr = _node(c, fr), _node(c, sr) while fn and not fn & 1: fn >>= 1 sn >>= 1 else: sr = _node(sr, c) fn >>= 1 sn >>= 1 return fr == old_root and sr == new_root and sn == 0 # ---------------------------------------------------------------- replayed / altered result detection # ---------------------------------------------------------------------------------------------- # Fault campaign: recompute the published summary from the raw per-trial records. # Same rules as fault-campaign/recompute_summary.py, restated here so this file stays self-contained. # ---------------------------------------------------------------------------------------------- FAULT_CAMPAIGN_SUMMARY_SCHEMA = "cain42.fault_campaign.summary.v1" MAX_TRIALS_BYTES = 256 * 1024 * 1024 def _fc_evaluate(rec): expected = "FAIL" if rec.get("designed_to_fail") else "PASS" if rec.get("harness_error") or "replicas" not in rec: return {"result": "ERROR", "expected_result": expected, "as_expected": False, "safety_ok": None, "checks": [{"name": "trial_ran", "ok": False, "detail": str(rec.get("harness_error"))[:200]}]} honest = [h for h in rec["honest"] if h in rec["replicas"]] reps = {h: rec["replicas"][h] for h in honest} exp = rec.get("expect", {}) checks = [] def add(name: str, ok: bool, detail: str = "", safety: bool = False): checks.append({"name": name, "ok": bool(ok), "detail": detail, "safety": safety}) by_seq = {} for h, r in reps.items(): for seq, digest in r["committed"].items(): by_seq.setdefault(seq, set()).add(digest) conflicts = sorted(s for s, d in by_seq.items() if len(d) > 1) add("no_two_honest_replicas_commit_different_digests_at_one_sequence", not conflicts, f"conflicting sequences {conflicts}" if conflicts else "", safety=True) by_index = {} for h, r in reps.items(): by_index.setdefault(r["commit_index"], set()).add(r["application_hash"]) diverged = sorted(i for i, hs in by_index.items() if len(hs) > 1) add("same_commit_index_means_same_application_state", not diverged, f"diverged at commit_index {diverged}" if diverged else "", safety=True) framed = sorted({f"{h}->{q}" for h, r in reps.items() for q in r["quarantined"] if q in honest}) add("no_honest_replica_quarantined_by_an_honest_replica", not framed, ",".join(framed), safety=True) indexes = [r["commit_index"] for r in reps.values()] if "min_commit" in exp: add("honest_replicas_reach_expected_commit_index", bool(indexes) and min(indexes) >= exp["min_commit"], f"min {min(indexes) if indexes else None}, expected >= {exp['min_commit']}") if "max_commit" in exp: add("no_commit_without_quorum", bool(indexes) and max(indexes) <= exp["max_commit"], f"max {max(indexes) if indexes else None}, expected <= {exp['max_commit']}") if exp.get("converged"): same = len(set(indexes)) == 1 and len({r["application_hash"] for r in reps.values()}) == 1 add("honest_replicas_converge", same, f"commit indexes {sorted(indexes)}") if exp.get("no_attack_commit") or exp.get("attacks_refused"): committed = {d for r in reps.values() for d in r["committed"].values()} hit = sorted(set(rec.get("attack_digests", [])) & committed) add("attack_digest_never_committed", not hit, ",".join(x[:12] for x in hit)) if exp.get("attacks_refused"): inj = rec.get("injections", []) accepted = [i for i in inj if i.get("probe_accepted")] add("every_injected_message_refused_by_the_replica", bool(inj) and not accepted, f"{len(accepted)} of {len(inj)} accepted") if exp.get("faulty_replica_detects_divergence"): faulty = [b for b in rec.get("cluster", {}).get("byzantine", []) if b in rec["replicas"]] seen_div = [b for b in faulty if rec["replicas"][b]["counters"].get("state_divergence", 0) >= 1] add("corrupted_replica_detects_its_own_divergence", bool(faulty) and seen_div == faulty, f"corrupted {faulty}, detected by {seen_div}") safety_ok = all(c["ok"] for c in checks if c["safety"]) result = "PASS" if all(c["ok"] for c in checks) else "FAIL" return {"result": result, "expected_result": expected, "as_expected": result == expected, "safety_ok": safety_ok, "min_honest_commit_index": min(indexes) if indexes else None, "requests_allowed": sum(1 for r in rec.get("requests", []) if r.get("decision") == "ALLOW"), "requests": len(rec.get("requests", [])), "injections": len(rec.get("injections", [])), "injections_refused": sum(1 for i in rec.get("injections", []) if not i.get("probe_accepted")), "checks": checks} def _fc_summarize(raw): scenarios = {} order = [] mismatches = [] seen = set() for n, line in enumerate(raw.decode().splitlines(), 1): if not line.strip(): continue rec = json.loads(line) key = (rec["scenario_id"], rec["trial"]) if key in seen: mismatches.append(f"line {n}: duplicate trial {key}") continue seen.add(key) ev = _fc_evaluate(rec) stored = rec.get("evaluation") if stored is not None and json.dumps(stored, sort_keys=True) != json.dumps(ev, sort_keys=True): mismatches.append(f"line {n}: stored evaluation of {key} differs from the recomputed one") sid = rec["scenario_id"] if sid not in scenarios: order.append(sid) scenarios[sid] = {"id": sid, "name": rec["scenario"], "designed_to_fail": bool(rec.get("designed_to_fail")), "trials": 0, "PASS": 0, "FAIL": 0, "ERROR": 0, "as_expected": 0, "safety_violations": 0, "trials_all_requests_allowed": 0, "requests": 0, "requests_allowed": 0, "injections": 0, "injections_refused": 0, "failed_checks": {}, "not_as_expected_trials": []} s = scenarios[sid] s["trials"] += 1 s[ev["result"]] += 1 s["as_expected"] += int(ev["as_expected"]) if not ev["as_expected"]: s["not_as_expected_trials"].append(rec["trial"]) if ev["safety_ok"] is False: s["safety_violations"] += 1 for k in ("requests", "requests_allowed", "injections", "injections_refused"): s[k] += ev.get(k) or 0 if ev.get("requests") and ev["requests_allowed"] == ev["requests"]: s["trials_all_requests_allowed"] += 1 for c in ev["checks"]: if not c["ok"]: s["failed_checks"][c["name"]] = s["failed_checks"].get(c["name"], 0) + 1 rows = [scenarios[i] for i in order] core = [r for r in rows if not r["designed_to_fail"]] ctrl = [r for r in rows if r["designed_to_fail"]] def tot(rs, k): return sum(r[k] for r in rs) totals = { "scenarios": len(core), "trials": tot(core, "trials"), "PASS": tot(core, "PASS"), "FAIL": tot(core, "FAIL"), "ERROR": tot(core, "ERROR"), "safety_violations": tot(core, "safety_violations"), "must_fail_scenarios": len(ctrl), "must_fail_trials": tot(ctrl, "trials"), "must_fail_FAIL": tot(ctrl, "FAIL"), "must_fail_PASS": tot(ctrl, "PASS"), "must_fail_ERROR": tot(ctrl, "ERROR"), "must_fail_safety_violations": tot(ctrl, "safety_violations"), "all_trials": tot(rows, "trials"), "trials_as_expected": tot(rows, "as_expected"), } ok = totals["all_trials"] > 0 and totals["trials_as_expected"] == totals["all_trials"] and not mismatches return {"schema": FAULT_CAMPAIGN_SUMMARY_SCHEMA, "trials_sha256": hashlib.sha256(raw).hexdigest(), "trials_bytes": len(raw), "totals": totals, "campaign_verdict": "AS_EXPECTED" if ok else "NOT_AS_EXPECTED", "record_mismatches": mismatches, "scenarios": rows} def check_result(result_path, archive_path): try: with open(result_path, "rb") as fh: res = strict_json(fh.read()) with open(archive_path, "rb") as fh: actual = sha256_hex(fh.read()) except Exception as e: return {"result_check": "VERIFICATION_ERROR", "detail": str(e)}, 70 problems = [] if res.get("schema") != RESULT_SCHEMA: problems.append(f"unsupported result schema {res.get('schema')!r}") body = {k: v for k, v in res.items() if k not in ("elapsed_seconds", "result_digest_sha256")} if sha256_hex(canonical(body)) != res.get("result_digest_sha256"): problems.append("result_digest_sha256 does not match the result body: the result was edited") if res.get("input", {}).get("archive_sha256") != actual: problems.append(f"result is for archive {res.get('input', {}).get('archive_sha256')}, not this archive {actual} " "(replayed or substituted result)") if res.get("exit_code") != VERDICT_EXIT.get(res.get("verdict")): problems.append("exit_code does not match verdict") ok = not problems return {"result_check": "RESULT_CONSISTENT" if ok else "RESULT_REJECTED", "problems": problems, "note": "A consistent result is not a signed attestation. Re-run the verifier yourself; it takes seconds."}, (0 if ok else 10) def main(argv=None): ap = argparse.ArgumentParser(description="CAIN-42 Auditor Kit: 42-Second Initial Evidence Verdict (offline, stdlib only)") ap.add_argument("archive", help="path to the downloaded kit ZIP") ap.add_argument("--trust-anchor", help="publisher Ed25519 public key, base64 (32 bytes), obtained OUT OF BAND") ap.add_argument("--trust-anchor-file", help="file containing the base64 public key") ap.add_argument("--expect-sha256", help="the published SHA-256 of the ZIP, obtained out of band") ap.add_argument("--expect-version", help="the kit version you intend to verify, e.g. 8.3.2") ap.add_argument("--revoked", action="append", default=[], help="revoked key fingerprint (sha256 hex); repeatable") ap.add_argument("--auditor-key", action="append", default=[], help="base64 Ed25519 key of an independent reviewer you trust") ap.add_argument("--now", help="evaluate freshness as of this UTC time (YYYY-MM-DDTHH:MM:SSZ) for reproducible runs") ap.add_argument("--check-result", help="check that a saved result JSON matches this archive and was not edited") ap.add_argument("--json", action="store_true", help="JSON output (the default)") ap.add_argument("--human", action="store_true", help="human-readable summary instead of JSON") ap.add_argument("--out", help="also write the JSON result to this path") a = ap.parse_args(argv) if a.check_result: res, code = check_result(a.check_result, a.archive) print(json.dumps(res, indent=1)) return code try: anchor = a.trust_anchor if a.trust_anchor_file: with open(a.trust_anchor_file, encoding="utf-8") as fh: anchor = fh.read().strip() now = parse_utc(a.now) if a.now else None except Exception as e: print(json.dumps({"schema": RESULT_SCHEMA, "verdict": "VERIFICATION_ERROR", "exit_code": 70, "error": str(e)})) return 70 rep = Verifier(a.archive, anchor, a.expect_sha256, a.expect_version, a.revoked, a.auditor_key, now).run() text = json.dumps(rep, indent=1, ensure_ascii=False) if a.out: with open(a.out, "w", encoding="utf-8") as fh: fh.write(text + "\n") if a.human: print("=" * 72) print("CAIN-42 AUDITOR KIT — INITIAL EVIDENCE VERDICT (integrity only)") print("=" * 72) r = rep["release"] print(f"Archive SHA-256 : {rep['input']['archive_sha256']}") print(f"Kit version : {r.get('kit_version')} source commit: {r.get('source_commit')}") print(f"Trust anchor : {r.get('trust_anchor_fingerprint_sha256')}") for c in rep["checks"]: if c["status"] != "PASS": print(f" [{c['status']}] {c['id']}: {c['detail']}") print(f"Checks : {rep['counts']}") p = rep["platform_status_as_recorded_in_package"] if p: print(f"Platform status (as the PUBLISHER recorded it): production_ready={p['production_ready']} {p['overall']}") print("-" * 72) print(f"VERDICT: {rep['verdict']} (exit {rep['exit_code']}) in {rep['elapsed_seconds']}s") if rep["error"]: print(f"ERROR: {rep['error']}") print("This is NOT an audit and NOT a production-readiness decision. See does_not_prove in --json output.") else: print(text) return rep["exit_code"] if __name__ == "__main__": sys.exit(main())