#!/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 hashlib import io import json import os import re import sys import time import zipfile from datetime import datetime, timezone VERIFIER_VERSION = "8.1.0" 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, } 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 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 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.1.0") 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())