#!/usr/bin/env python3 """Independent Byzantine-cluster prober. Imports NOTHING from CAIN. Needs Python 3.9+ and, for signature checks, the `cryptography` package. Read-only GET requests only. python3 cluster_probe.py [BASE_URL] [--strict] [--out FILE] BASE_URL (default https://cainstudio.online) is asked for /api/v1/cluster/nodes; every node endpoint it lists is then probed directly, plus BASE_URL itself. For each node it records /health, /api/v1/cluster/status, /nodes (its own view of membership) and, where served, /api/v1/cluster/pbft/state-proof, whose Ed25519 signature is verified here (signature is over sha256 of the canonical JSON of the proof without signature_b64/public_key_b64). Everything in the verdict is DERIVED from what the nodes returned; nothing is assumed: N = distinct node ids seen; f = floor((N-1)/3); a Byzantine quorum needs Q = 2f+1 (=3 for N=4). BFT_F1_ESTABLISHED requires ALL of: N>=4; every node reports quorum >= Q; every node reachable; >= Q nodes each serve a signed state proof that VERIFIES, under distinct public keys, agreeing on cluster/view/commit_index/state_root; and all membership views agree. Anything else is NOT_ESTABLISHED with the exact reasons. A node's own claim (for example "byzantine_f1_readiness") is reported next to the derived verdict and flagged if unsupported. --strict exits 1 unless the verdict is BFT_F1_ESTABLISHED. """ import base64, hashlib, json, sys, time, urllib.request, urllib.error TIMEOUT = 15 def get(url): t = time.time() try: with urllib.request.urlopen(urllib.request.Request(url, headers={"Accept": "application/json", "User-Agent": "cluster-probe/1"}), timeout=TIMEOUT) as r: body = r.read() code = r.status except urllib.error.HTTPError as e: body, code = e.read(), e.code except Exception as e: # unreachable, timeout, TLS... return {"url": url, "ok": False, "error": type(e).__name__ + ": " + str(e)[:100], "ms": round((time.time() - t) * 1000)} try: data = json.loads(body) except Exception: data = None return {"url": url, "ok": code == 200 and data is not None, "http": code, "json": data, "ms": round((time.time() - t) * 1000)} def verify_proof(proof): """Returns (valid, reason). Signature over sha256(canonical json without signature_b64/public_key_b64).""" try: from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PublicKey except ImportError: return None, "cryptography not installed" p = dict(proof) sig, pk = p.pop("signature_b64", None), p.pop("public_key_b64", None) if not sig or not pk: return False, "missing signature or public key" for canon in (json.dumps(p, sort_keys=True, separators=(",", ":"), ensure_ascii=True), json.dumps(p, sort_keys=True, separators=(",", ":"), ensure_ascii=False)): try: Ed25519PublicKey.from_public_bytes(base64.b64decode(pk)).verify(base64.b64decode(sig), hashlib.sha256(canon.encode()).digest()) return True, "ok" except Exception: continue return False, "signature does not verify" def main(): args = [a for a in sys.argv[1:] if not a.startswith("--")] base = (args[0] if args else "https://cainstudio.online").rstrip("/") strict = "--strict" in sys.argv out = sys.argv[sys.argv.index("--out") + 1] if "--out" in sys.argv else None started = time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()) listing = get(base + "/api/v1/cluster/nodes") endpoints = {"gateway@" + base: base} for n in (listing.get("json") or {}).get("nodes", []): if n.get("endpoint") and not n["endpoint"].startswith("http://localhost"): endpoints[n["node_id"] + "@" + n["endpoint"]] = n["endpoint"].rstrip("/") probes = {} for label, ep in endpoints.items(): probes[label] = {"endpoint": ep, "health": get(ep + "/health"), "status": get(ep + "/api/v1/cluster/status"), "nodes": get(ep + "/api/v1/cluster/nodes"), "state_proof": get(ep + "/api/v1/cluster/pbft/state-proof")} node_ids, reasons, per_node = set(), [], {} for label, p in probes.items(): st = (p["status"].get("json") or {}) if p["status"]["ok"] else {} nid = st.get("node_id") if nid: node_ids.add(nid) ni = st.get("node_identity") or {} view = sorted(n["node_id"] for n in ((p["nodes"].get("json") or {}).get("nodes", []))) if p["nodes"]["ok"] else None proof = p["state_proof"].get("json") if p["state_proof"]["ok"] else None valid, why = verify_proof(proof) if proof else (None, "no state proof served") per_node[label] = { "reachable": p["health"]["ok"], "node_id": nid, "reported_quorum": st.get("quorum"), "reported_node_count": st.get("node_count"), "reported_healthy_nodes": st.get("healthy_nodes"), "reported_is_quorum": st.get("is_quorum"), "self_claim_byzantine_f1_readiness": st.get("byzantine_f1_readiness"), "software_version": ni.get("software_version"), "public_key_in_status": ni.get("public_key_b64"), "membership_view": view, "cluster_routes_served": None, "state_proof": None if not proof else {"valid_signature": valid, "reason": why, "cluster_id": proof.get("cluster_id"), "view": proof.get("view"), "commit_index": proof.get("commit_index"), "state_root": proof.get("state_root"), "signer_public_key": proof.get("public_key_b64"), "pbft_engine_version": proof.get("pbft_engine_version")}, "state_proof_status": "served" if proof else ("HTTP %s" % p["state_proof"].get("http") if "http" in p["state_proof"] else p["state_proof"].get("error")), } N = len(node_ids) f = (N - 1) // 3 if N else 0 Q = 2 * f + 1 if N < 4: reasons.append(f"only {N} distinct node ids observed; f=1 needs N>=4") unreachable = [l for l, v in per_node.items() if not v["reachable"]] if unreachable: reasons.append("unreachable: " + ", ".join(unreachable)) low_q = {l: v["reported_quorum"] for l, v in per_node.items() if v["reported_quorum"] is not None and v["reported_quorum"] < Q} if low_q: reasons.append(f"nodes reporting a quorum below the Byzantine quorum {Q} (2f+1 for N={N}): {low_q}; a quorum of {min(low_q.values())} lets two conflicting decisions both commit") no_q = [l for l, v in per_node.items() if v["reported_quorum"] is None] if no_q: reasons.append("no quorum reported by: " + ", ".join(no_q)) # count each NODE once (the same node can be reached under two labels, e.g. as the base URL and as a listed peer) verified = {} for l, v in per_node.items(): sp = v["state_proof"] if sp and sp["valid_signature"] is True: verified.setdefault(v["node_id"] or l, sp) keys = {s["signer_public_key"] for s in verified.values()} if len(verified) < Q: reasons.append(f"only {len(verified)} node(s) serve a verifiable signed state proof; need >= {Q}") if len(keys) < len(verified): reasons.append("verified proofs share a signing key") if len({(s["cluster_id"], s["view"], s["commit_index"], s["state_root"]) for s in verified.values()}) > 1: reasons.append("verified proofs disagree on cluster/view/commit_index/state_root") views = {tuple(v["membership_view"]) for v in per_node.values() if v["membership_view"] is not None} if len(views) > 1: reasons.append("nodes disagree on membership: " + "; ".join(sorted(",".join(v) for v in views))) versions = {v["software_version"] for v in per_node.values() if v["software_version"]} no_ver = [l for l, v in per_node.items() if not v["software_version"]] hetero = None if no_ver and versions: hetero = f"software version reported by only some nodes (missing: {', '.join(no_ver)}): the cluster is not running one verified build" if hetero: reasons.append(hetero) claims = {l: v["self_claim_byzantine_f1_readiness"] for l, v in per_node.items() if v["self_claim_byzantine_f1_readiness"]} verdict = "BFT_F1_ESTABLISHED" if not reasons else "NOT_ESTABLISHED" flagged = [f"{l} claims {c!r} but the derived verdict is {verdict}" for l, c in claims.items() if "PROVEN" in str(c).upper() and verdict != "BFT_F1_ESTABLISHED"] result = {"schema": "cluster-probe.v1", "probed_utc": started, "base": base, "derived": {"N": N, "f": f, "byzantine_quorum_2f_plus_1": Q, "verified_signed_proofs": len(verified), "distinct_signing_keys": len(keys)}, "verdict": verdict, "reasons": reasons, "unsupported_self_claims": flagged, "nodes": per_node, "note": "Read-only probe. A single host's signed proof shows that host's state is authentic, not that the cluster tolerates Byzantine faults."} text = json.dumps(result, indent=1, sort_keys=True) if out: open(out, "w").write(text + "\n") print(text) sys.exit(1 if strict and verdict != "BFT_F1_ESTABLISHED" else 0) if __name__ == "__main__": main()