#!/usr/bin/env python3 """CAIN-42 BFT clean-room judge (ZERO CAIN imports; stdlib + `cryptography`). Input: what each replica claims it committed (exported certificates) + the pinned membership. It re-derives from the PBFT rules alone (Castro & Liskov 1999, restated in BFT_PROTOCOL_SPEC.md): R1 digest binding request_digest == SHA256(canonical {"op","req_id","client"} of the PRE_PREPARE payload) R2 primary PRE_PREPARE sender == sorted(members)[view % n], valid Ed25519 under the PINNED key R3 prepared |{pre_prepare sender} U {PREPARE senders}| >= 2f+1, all distinct pinned members, each PREPARE for the same (view, seq, digest), each signature valid R4 committed >= 2f+1 COMMITs from distinct pinned members for the same (view, seq, digest), valid sigs R5 agreement no two replicas hold committed certificates for one seq with different digests R6 integrity a replica's commit_index prefix 1..k has a valid certificate for every seq (no gaps) Signature rule (wire spec): Ed25519 over the UTF-8 hex of SHA256(canonical JSON of the signing payload {msg_type, sender_id, cluster_id, view, sequence, request_digest, payload, epoch, nonce, timestamp rounded to 4 dp, engine_version}). Canonical JSON = sort_keys, separators (",", ":"), default=str. A key carried INSIDE a message is never used. It does not trust the engine's own safety report or any "valid" flag. Same-project and same-author as the vector harness; written from the spec, not by importing the engine. """ from __future__ import annotations import base64 import hashlib import json import sys from typing import Any, Dict, List, Optional, Tuple from cryptography.exceptions import InvalidSignature from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PublicKey def cj(x: Any) -> bytes: return json.dumps(x, sort_keys=True, separators=(",", ":"), default=str).encode("utf-8") def sha(x: Any) -> str: return hashlib.sha256(cj(x)).hexdigest() def signing_payload(m: Dict[str, Any]) -> Dict[str, Any]: return {"msg_type": m["msg_type"], "sender_id": m["sender_id"], "cluster_id": m["cluster_id"], "view": m["view"], "sequence": m["sequence"], "request_digest": m["request_digest"], "payload": json.loads(cj(m.get("payload", {}))), "epoch": m.get("epoch", 1), "nonce": m["nonce"], "timestamp": round(m["timestamp"], 4), "engine_version": m["engine_version"]} def sig_ok(m: Dict[str, Any], pins: Dict[str, str]) -> bool: pub = pins.get(m.get("sender_id")) if not pub: return False try: Ed25519PublicKey.from_public_bytes(base64.b64decode(pub)).verify( base64.b64decode(m["signature_b64"]), sha(signing_payload(m)).encode("utf-8")) return True except (InvalidSignature, ValueError, TypeError, KeyError): return False def proposal_digest(payload: Dict[str, Any]) -> Optional[str]: if not isinstance(payload, dict) or not isinstance(payload.get("operation"), dict): return None return sha({"op": payload["operation"], "req_id": payload.get("request_id"), "client": payload.get("client_id")}) def judge_entry(e: Dict[str, Any], pins: Dict[str, str], cluster_id: str) -> Tuple[bool, str]: members = sorted(pins); n = len(members); f = (n - 1) // 3; q = 2 * f + 1 v, s, d = e["view"], e["sequence"], e["request_digest"] pp = e.get("pre_prepare") or {} if pp.get("msg_type") != "PRE_PREPARE" or (pp.get("view"), pp.get("sequence"), pp.get("request_digest")) != (v, s, d): return False, "R2:no matching PRE_PREPARE" if pp.get("cluster_id") != cluster_id: return False, "R2:wrong cluster" if proposal_digest(pp.get("payload")) != d: return False, "R1:digest not bound to operation" if pp.get("sender_id") != members[v % n] or not sig_ok(pp, pins): return False, "R2:PRE_PREPARE not from the view's primary or bad signature" voters = {pp["sender_id"]} for sender, m in (e.get("prepares") or {}).items(): if m.get("msg_type") == "PREPARE" and m.get("sender_id") == sender and sender in pins and \ (m["view"], m["sequence"], m["request_digest"]) == (v, s, d) and m.get("cluster_id") == cluster_id \ and sig_ok(m, pins): voters.add(sender) if len(voters) < q: return False, f"R3:prepared by {len(voters)} < {q}" committers = set() for sender, m in (e.get("commit_certificate") or {}).items(): if m.get("msg_type") == "COMMIT" and m.get("sender_id") == sender and sender in pins and \ (m["view"], m["sequence"], m["request_digest"]) == (v, s, d) and m.get("cluster_id") == cluster_id \ and sig_ok(m, pins): committers.add(sender) if len(committers) < q: return False, f"R4:committed by {len(committers)} < {q}" return True, "OK" def judge(replicas: Dict[str, Dict[str, Any]], pins: Dict[str, str], cluster_id: str) -> Dict[str, Any]: """replicas: {node: {"commit_index": k, "entries": [certificate entries]}}""" problems: List[str] = [] by_seq: Dict[int, Dict[str, str]] = {} valid_entries = 0 for node, rep in sorted(replicas.items()): good = {} k = int(rep.get("commit_index", 0)) for e in rep.get("entries", []): if not e.get("commit_certificate") and int(e.get("sequence", 0)) > k: continue # prepared, not claimed committed ok, why = judge_entry(e, pins, cluster_id) if not ok: problems.append(f"{node}@{e.get('sequence')}:{why}") continue valid_entries += 1 good[int(e["sequence"])] = e["request_digest"] by_seq.setdefault(int(e["sequence"]), {})[node] = e["request_digest"] missing = [s for s in range(1, k + 1) if s not in good] if missing: problems.append(f"{node}:R6:executed prefix 1..{k} lacks valid certificates for {missing[:5]}") conflicts = {s: d for s, d in by_seq.items() if len(set(d.values())) > 1} for s, d in sorted(conflicts.items()): problems.append(f"R5:conflicting commits at seq {s}: {d}") return {"verdict": "SAFE" if not problems else "VIOLATION", "valid_certificates": valid_entries, "conflicting_sequences": sorted(conflicts), "problems": problems, "committed_sequences": {n: sorted(s for s, x in by_seq.items() if n in x) for n in sorted(replicas)}} if __name__ == "__main__": data = json.load(open(sys.argv[1])) out = {v["vector_id"]: judge(v["export"]["replicas"], v["export"]["pins"], v["export"]["cluster_id"]) for v in data["vectors"]} print(json.dumps(out, indent=2, sort_keys=True)) sys.exit(0 if all(o["verdict"] == "SAFE" for o in out.values()) else 1)