#!/usr/bin/env python3 """Standalone verifier for CAIN-42 cluster evidence. Imports NOTHING from CAIN; stdlib + `cryptography`. It re-checks, from raw data embedded in a bundle (or fetched live from a node with --node), the cryptographic facts: * a node's signed state proof: Ed25519 over sha256(canonical JSON of the proof without signature/public_key) * a quorum certificate (QC): every vote is Ed25519 over the raw sha256 digest of the canonical statement, verified against the membership public keys; counts DISTINCT valid signers and compares with the quorum the QC claims * agreement: replicas that report the same sequence must report the same state root * membership keys in a QC must equal the keys the nodes themselves report (not merely what the QC asserts) Canonical JSON = json.dumps(obj, sort_keys=True, separators=(",", ":"), default=str) (ASCII-escaped), as the system defines it. WHAT THIS DOES NOT PROVE: that the nodes are on independent machines or run independent code, that no node was compromised, or that the cluster tolerates any fault. It proves the signatures and the recorded agreement are mathematically consistent. Not independent review. """ import argparse, base64, hashlib, json, sys, urllib.request def canon(o): return json.dumps(o, sort_keys=True, separators=(",", ":"), default=str).encode() def _verify(pub_b64, sig_b64, msg): from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PublicKey try: Ed25519PublicKey.from_public_bytes(base64.b64decode(pub_b64)).verify(base64.b64decode(sig_b64), msg) return True except Exception: return False def verify_state_proof(p): body = {k: v for k, v in p.items() if k not in ("signature_b64", "public_key_b64")} return _verify(p.get("public_key_b64", ""), p.get("signature_b64", ""), hashlib.sha256(canon(body)).digest()) def qc_statement_digest(qc): stmt = {"cluster_id": qc["cluster_id"], "authority_epoch": qc["authority_epoch"], "view": qc["view"], "sequence": qc["sequence"], "request_id": qc["request_id"], "request_digest": qc["request_digest"], "state_root": qc["state_root"], "membership_hash": hashlib.sha256(canon(qc["membership_config"])).hexdigest(), "created_at": qc["created_at"], "expires_at": qc["expires_at"]} return hashlib.sha256(canon(stmt)).digest() def verify_qc(qc, required_quorum, pinned_keys=None): """Returns (ok, detail). pinned_keys: {node_id: pubkey_b64} as reported by the nodes themselves.""" d = {"valid_signers": [], "invalid_signers": [], "unknown_signers": [], "required": required_quorum, "membership_matches_pinned": None} digest = qc_statement_digest(qc) members = qc["membership_config"] for node, sig in qc.get("votes", {}).items(): if node not in members: d["unknown_signers"].append(node) elif _verify(members[node], sig, digest): d["valid_signers"].append(node) else: d["invalid_signers"].append(node) if pinned_keys is not None: d["membership_matches_pinned"] = all(pinned_keys.get(n) == k for n, k in members.items()) and set(members) == set(pinned_keys) ok = len(set(d["valid_signers"])) >= required_quorum and d["membership_matches_pinned"] in (None, True) return ok, d def verify_agreement(proofs): """Replicas at the same sequence must share one state root. Returns (ok, detail).""" by_seq = {} for p in proofs: by_seq.setdefault(p["sequence"], set()).add(p["state_root"]) return all(len(v) == 1 for v in by_seq.values()), {str(k): sorted(v) for k, v in by_seq.items()} def verify_bundle(b): out = [] def rec(name, ok, detail=""): out.append((name, "PASS" if ok else "FAIL", detail)); print(f"{'PASS' if ok else 'FAIL':6} {name}" + (f" -- {detail}" if detail else "")) for step in b.get("steps", []): for pr in step.get("state_proofs", []): rec(f"[{step['name']}] state proof signature {pr['node_id']}", verify_state_proof(pr)) if step.get("state_proofs"): ok, det = verify_agreement(step["state_proofs"]) rec(f"[{step['name']}] replicas at one sequence share one state root", ok, json.dumps(det)[:120]) for qc_rec in step.get("quorum_certificates", []): ok, det = verify_qc(qc_rec["qc"], qc_rec["required_quorum"], qc_rec.get("pinned_keys")) rec(f"[{step['name']}] QC {qc_rec['qc']['request_id']}: {len(set(det['valid_signers']))} valid signers >= {qc_rec['required_quorum']}", ok, f"invalid={det['invalid_signers']} unknown={det['unknown_signers']} membership_matches_pinned={det['membership_matches_pinned']}") claimed = qc_rec.get("claimed_accepted_signers") if claimed is not None: rec(f"[{step['name']}] node-claimed signer count equals independently verified count", claimed == len(set(det["valid_signers"])), f"claimed={claimed} verified={len(set(det['valid_signers']))}") return out def main(): ap = argparse.ArgumentParser() ap.add_argument("bundle", nargs="?") ap.add_argument("--node", action="append", help="live node base URL (repeatable): verifies its state proof and cross-node agreement") a = ap.parse_args() res = [] if a.node: proofs = [] for n in a.node: p = json.load(urllib.request.urlopen(n.rstrip("/") + "/api/v1/cluster/pbft/state-proof", timeout=15)); proofs.append(p) ok = verify_state_proof(p); res.append(ok); print(f"{'PASS' if ok else 'FAIL':6} live state proof {p['node_id']} seq={p['sequence']} root={p['state_root'][:12]}") ok, det = verify_agreement(proofs); res.append(ok); print(f"{'PASS' if ok else 'FAIL':6} live agreement {json.dumps(det)[:160]}") ks = {p["public_key_b64"] for p in proofs} ok = len(ks) == len(proofs); res.append(ok); print(f"{'PASS' if ok else 'FAIL':6} distinct signing keys: {len(ks)}/{len(proofs)}") if a.bundle: b = json.load(open(a.bundle)) res += [r[1] == "PASS" for r in verify_bundle(b)] if not res: print("give a bundle path or --node URL"); return 2 print(f"\n{'VERIFIED' if all(res) else 'REJECTED'}: {sum(res)}/{len(res)} checks. Signatures and recorded agreement only; not independence, not fault tolerance, not independent review.") return 0 if all(res) else 1 if __name__ == "__main__": sys.exit(main())