#!/usr/bin/env python3 """Verify the CAIN-42 72-hour soak evidence, checkpoint by checkpoint, without trusting any site. No CAIN imports; needs Python 3.8+, `cryptography`, and verify_pbft_qc_bundle.py (standalone) next to this file. python3 verify_soak.py https://clawx.click/evidence/soak-72h-2026-09-25/ For every published checkpoint: * signed by the CAIN-42 evidence-root key (fetched from all three sites; must match); * hash-chained: previous_checkpoint_hash == hash of the checkpoint before (none missing); * membership configuration hash recomputed; N=4, f=1, Q=3, fast path declared; * every window quorum certificate verifies (Ed25519 signatures by >= 3 pinned members, or all 4 for FAST_COMMIT_QC) and consecutive certificates chain (parent = previous decision); * the two sampled nodes agree on the decision hash at every common sequence; * nodes at the same height report the same application-state hash; * MCPGate: the fresh authorization verifies, the gate ALLOWed it with a signed proof, and its replay was DENIED (AUTHORIZATION_REPLAY); * counters never decrease; divergences == 0. Result: RUNNING (elapsed hours) while the soak is in progress; PASS or FAIL once final. """ from __future__ import annotations import hashlib import json import os import sys import urllib.request from typing import Any, Dict, List sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) import verify_pbft_qc_bundle as Q # noqa: E402 DOMAIN = "CAIN42/SOAK72/CHECKPOINT/v1" KEY_URLS = ["https://cainstudio.online/proof/bundle/claims/evidence-root.pub.json", "https://mcpgate.online/proof/bundle/claims/evidence-root.pub.json", "https://clawx.click/evidence/claims/evidence-root.pub.json"] def get(u: str) -> Any: if u.startswith("http"): with urllib.request.urlopen(u, timeout=30) as r: return json.loads(r.read()) with open(u) as f: return json.load(f) def canon(o) -> bytes: return json.dumps(o, sort_keys=True, separators=(",", ":"), ensure_ascii=True).encode() def verify(base: str, pin: str = None) -> Dict[str, Any]: base = base.rstrip("/") + "/" idx = get(base + "index.json") mem = get(base + "membership.json") mb = Q.Membership(mem["membership"]) problems: List[str] = [] if pin is None: keys = set() for u in KEY_URLS: try: keys.add(get(u)["public_key_b64"]) except Exception as e: # noqa: BLE001 problems.append(f"evidence-root key unreachable at {u}: {e}") if len(keys) != 1: problems.append(f"evidence-root key differs across sites: {keys}") pin = next(iter(keys)) if keys else None want_fast = idx.get("fast_path", True) # soaks declare it; the 2026-09-25 soak predates the field if (mb.n, mb.f, mb.quorum, mb.fast_path) != (4, 1, 3, want_fast) or mb.configuration_hash != mem["configuration_hash"]: problems.append(f"membership does not recompute or is not N=4 f=1 Q=3 with fast_path={want_fast}") prev = hashlib.sha256(canon({"domain": DOMAIN, "genesis": idx["cluster_id"]})).hexdigest() last_counters, rows = None, [] for i, name in enumerate(idx["checkpoints"]): ck = get(base + name) p = [] body = {k: v for k, v in ck.items() if k not in ("checkpoint_hash", "evidence_root_signature_b64")} d = hashlib.sha256(canon(body)).hexdigest() if d != ck["checkpoint_hash"]: p.append("content does not match checkpoint_hash") if not Q.ed25519_ok(pin, ck["evidence_root_signature_b64"], ck["checkpoint_hash"].encode()): p.append("not signed by the evidence-root key") if ck["checkpoint"] != i or ck["previous_checkpoint_hash"] != prev: p.append("hash chain broken (missing, reordered or replaced checkpoint)") if ck["configuration_hash"] != mb.configuration_hash: p.append("configuration hash differs from the membership") decisions = {} nqc = 0 for n, certs in (ck.get("window_certificates") or {}).items(): parent = None for e in certs: for kind in ("commit_qc", "prepare_qc"): ok, why = Q.verify_qc(e[kind], mb, expected_parent=parent if kind == "commit_qc" else None) if not ok: p.append(f"{n} seq {e['sequence']} {kind}: {why}") nqc += 1 parent = Q.qc_decision_hash(e["commit_qc"]) if decisions.setdefault(e["sequence"], parent) != parent: p.append(f"nodes disagree on the decision at sequence {e['sequence']}") by_h = {} for n, s in ck["nodes"].items(): if "commit_index" in s: by_h.setdefault(s["commit_index"], set()).add(s["application_state_hash"]) if any(len(v) > 1 for v in by_h.values()): p.append("nodes at the same height report different application state") enf = ck.get("enforcement") or {} if "certificate" not in enf: p.append(f"no MCPGate enforcement evidence in this checkpoint ({enf.get('error', 'missing')})") else: ok, why = Q.verify_auth_cert(enf["certificate"], mb) if not ok: p.append(f"authorization: {why}") for tag, want in (("allow", "ALLOW"), ("replay", "DENY")): gp = enf[tag] if gp.get("decision") != want or not Q.ed25519_ok(ck["gate_public_key_b64"], gp["gate_signature_b64"], gp["proof_hash"].encode()): p.append(f"MCPGate {tag} proof invalid or not {want}") if enf["replay"].get("code") != "AUTHORIZATION_REPLAY": p.append("replay was not refused as AUTHORIZATION_REPLAY") c = ck["counters"] if last_counters and any(c[k] < last_counters[k] for k in c if k in last_counters): p.append("a counter decreased") if c["divergences"]: p.append(f"{c['divergences']} state divergences recorded") last_counters = c prev = ck["checkpoint_hash"] rows.append({"checkpoint": name, "elapsed_hours": ck["elapsed_hours"], "certificates_verified": nqc, "height": max((s.get("commit_index") or 0) for s in ck["nodes"].values()), "result": "VALID" if not p else "INVALID", "problems": p}) problems += [f"{name}: {x}" for x in p] last = get(base + idx["checkpoints"][-1]) if idx["checkpoints"] else {} if problems: verdict = "FAIL" elif last.get("final"): verdict = "PASS" if last.get("elapsed_hours", 0) >= last.get("planned_hours", 72) - 0.1 else "FAIL (ended early)" else: # A soak that stopped publishing is not "running": checkpoints are hourly, # so one older than 2 h without a final checkpoint means the harness died # (the 2026-09-25 soak kept reading RUNNING after it was OOM-killed). import datetime as _dt age_h = None if last.get("at"): age_h = (_dt.datetime.now(_dt.timezone.utc) - _dt.datetime.strptime(last["at"], "%Y-%m-%dT%H:%M:%SZ") .replace(tzinfo=_dt.timezone.utc)).total_seconds() / 3600 if age_h is not None and age_h > 2: verdict = (f"STOPPED EARLY ({last.get('elapsed_hours', 0)} of {idx['planned_hours']} h; no checkpoint for " f"{age_h:.1f} h; all published checkpoints valid)") else: verdict = f"RUNNING ({last.get('elapsed_hours', 0)} of {idx['planned_hours']} h, all checkpoints valid so far)" return {"verdict": verdict, "checkpoints": rows, "problems": problems, "counters": last.get("counters")} def main(argv: List[str]) -> int: if len(argv) < 2: print(__doc__) return 2 r = verify(argv[1], argv[argv.index("--pin") + 1] if "--pin" in argv else None) if "--json" in argv: print(json.dumps(r, indent=1)) else: for c in r["checkpoints"]: print(f"{c['result']:8s} {c['checkpoint']} t={c['elapsed_hours']:7.2f} h height {c['height']:6d} " f"{c['certificates_verified']} certificates verified") for x in c["problems"][:5]: print(f" - {x}") print(f"\ncounters: {r['counters']}\nVERDICT: {r['verdict']}") return 0 if not r["problems"] else 1 if __name__ == "__main__": sys.exit(main(sys.argv))