#!/usr/bin/env python3
"""Standalone verifier for a CAIN-42 PBFT quorum-certificate evidence bundle.

No CAIN imports. Needs Python 3.8+ and the `cryptography` package (Ed25519).

    python3 verify_pbft_qc_bundle.py PBFT_QC_BUNDLE.json
    python3 verify_pbft_qc_bundle.py https://cainstudio.online/proof/bundle/pbft-evolution2-2026-09-24/PBFT_QC_BUNDLE.json

What it checks, from the bytes in the bundle only:
  * membership: configuration_hash recomputed from the member ids + Ed25519
    public keys; every node and every certificate carries that hash;
  * every COMMIT_QC and PREPARE_QC held by every node: each vote is an
    Ed25519 signature by a distinct member over the SHA-256 of the canonical
    signed message, of the right type (COMMIT vs PREPARE: domain separation),
    for exactly this cluster / epoch / view / sequence / digest; at least
    2f+1 = 3 distinct signers; the leader proposal is signed by the primary of
    that view and its digest binds the proposed operation;
  * certificate_hash and bundle_hash recomputed from content;
  * the decision chain: decision_hash(seq) = H(..., seq, digest, parent)
    folded from genesis, contiguous, and identical on all four nodes;
  * view changes recorded in the run: VIEW_CHANGE_QC signatures and quorum;
  * authorization certificates: certificate_id recomputed, issuer signature,
    bound to the COMMIT_QC it embeds and to the committed request;
  * negative controls: it tampers with a valid certificate in seven ways and
    requires every tampered copy to FAIL. A verifier that accepts everything
    would fail these.

It does not prove: that the run happened on independent machines (it did not:
one host), that the software is free of bugs, or anything about a cluster
other than the one that produced the bundle.
"""
from __future__ import annotations

import base64
import copy
import hashlib
import json
import sys
import urllib.request
from typing import Any, Dict, List, Optional, Tuple

from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PublicKey

QC_DOMAIN = "CAIN42/PBFT/QC/v1"
DECISION_DOMAIN = "CAIN42/PBFT/DECISION/v1"
BUNDLE_DOMAIN = "CAIN42/PBFT/SIGBUNDLE/v1"
AUTH_DOMAIN = "CAIN42/PBFT/AUTHORIZATION-CERTIFICATE/v1"
VOTE_TYPE = {"PREPARE_QC": "PREPARE", "COMMIT_QC": "COMMIT", "VIEW_CHANGE_QC": "VIEW_CHANGE",
             "CHECKPOINT_QC": "CHECKPOINT", "FAST_COMMIT_QC": "PREPARE"}
LEADER_TYPES = ("PREPARE_QC", "FAST_COMMIT_QC")      # carry the leader's signed proposal
DECIDING_TYPES = ("COMMIT_QC", "FAST_COMMIT_QC")     # decide a sequence
CHAINED_TYPES = ("PREPARE_QC", "COMMIT_QC", "FAST_COMMIT_QC")


# ----------------------------------------------------------------- primitives
def canon(o: Any) -> bytes:
    return json.dumps(o, sort_keys=True, separators=(",", ":"), default=str).encode("utf-8")


def h(o: Any) -> str:
    return hashlib.sha256(canon(o)).hexdigest()


def msg_digest(m: Dict[str, Any]) -> str:
    """SHA-256 of the canonical signed form of a PBFT message."""
    return h({
        "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(canon(m["payload"])), "epoch": m["epoch"], "nonce": m["nonce"],
        "timestamp": round(m["timestamp"], 4), "engine_version": m["engine_version"],
    })


def ed25519_ok(pub_b64: str, sig_b64: str, data: bytes) -> bool:
    try:
        sig = base64.b64decode(sig_b64, validate=True)
        if base64.b64encode(sig).decode() != sig_b64:
            return False
        Ed25519PublicKey.from_public_bytes(base64.b64decode(pub_b64)).verify(sig, data)
        return True
    except Exception:
        return False


def proposal_request_digest(payload: Any) -> Optional[str]:
    if not isinstance(payload, dict) or not isinstance(payload.get("operation"), dict):
        return None
    return h({"op": payload["operation"], "req_id": payload.get("request_id"), "client": payload.get("client_id")})


def genesis(cluster_id: str, epoch: int) -> str:
    return h({"domain": DECISION_DOMAIN, "genesis": True, "cluster_id": cluster_id, "epoch": epoch})


def decision_hash(cluster_id: str, epoch: int, seq: int, digest: str, parent: str) -> str:
    return h({"domain": DECISION_DOMAIN, "cluster_id": cluster_id, "epoch": epoch, "sequence": seq,
              "proposal_digest": digest, "parent": parent})


# ----------------------------------------------------------------- membership
class Membership:
    def __init__(self, cfg: Dict[str, Any]):
        self.cfg = cfg
        self.cluster_id = cfg["cluster_id"]
        self.epoch = cfg["epoch"]
        self.members = sorted(m["node_id"] for m in cfg["members"])
        self.keys = {m["node_id"]: m["public_key_b64"] for m in cfg["members"]}
        n = max(len(self.members), 4)
        self.n, self.f = n, (n - 1) // 3
        self.quorum = -(-(n + self.f + 1) // 2)       # ceil((N+f+1)/2) = 3 for N=4
        self.configuration_hash = h(cfg)
        # Evolution 3: a FAST_COMMIT_QC (all N members' votes) is a commitment
        # only on a membership whose NEW_VIEW rule honours vote histories.
        self.fast_path = cfg.get("fast_path") == "vote_history/v1"

    def primary_for(self, view: int) -> str:
        return self.members[view % len(self.members)]


# ----------------------------------------------------------------- certificates
def bundle_hash(b: Dict[str, Any]) -> str:
    sigs = b["individual_signatures"]
    return h({"domain": BUNDLE_DOMAIN, "algorithm": b["aggregation_algorithm"], "version": b["algorithm_version"],
              "aggregate_signature": b["aggregate_signature"],
              "signatures": [[s, msg_digest(sigs[s]), sigs[s].get("signature_b64")] for s in sorted(sigs)]})


def qc_decision_hash(qc: Dict[str, Any]) -> str:
    if qc["certificate_type"] not in CHAINED_TYPES or not qc["parent_decision_hash"]:
        return ""
    return decision_hash(qc["cluster_id"], qc["epoch"], qc["sequence"], qc["proposal_digest"],
                         qc["parent_decision_hash"])


def qc_hash(qc: Dict[str, Any]) -> str:
    b = qc["bundle"]
    return h({
        "domain": QC_DOMAIN, "certificate_type": qc["certificate_type"], "cluster_id": qc["cluster_id"],
        "epoch": qc["epoch"], "view": qc["view"], "sequence": qc["sequence"],
        "proposal_digest": qc["proposal_digest"], "parent_decision_hash": qc["parent_decision_hash"],
        "decision_hash": qc_decision_hash(qc), "configuration_hash": qc["configuration_hash"],
        "quorum": qc["quorum"], "quorum_policy_hash": h({"domain": "CAIN42/PBFT/QUORUM-POLICY/v1",
                                                         **qc["quorum_policy"]}),
        "signer_set": sorted(b["individual_signatures"]), "bundle_hash": bundle_hash(b),
        "leader_proposal_digest": msg_digest(qc["leader_proposal"]) if qc.get("leader_proposal") else None,
    })


def verify_qc(qc: Dict[str, Any], mb: Membership, expected_parent: Optional[str] = None) -> Tuple[bool, str]:
    try:
        t = qc["certificate_type"]
        want = VOTE_TYPE.get(t)
        if want is None:
            return False, f"unknown certificate type {t!r}"
        if (qc["cluster_id"], qc["epoch"]) != (mb.cluster_id, mb.epoch):
            return False, "wrong cluster or epoch"
        if qc["configuration_hash"] != mb.configuration_hash:
            return False, "configuration_hash is not the membership's"
        if expected_parent is not None and qc["parent_decision_hash"] != expected_parent:
            return False, "parent_decision_hash does not continue the chain"
        b = qc["bundle"]
        if (b["aggregation_algorithm"], b["algorithm_version"], b["aggregate_signature"]) != ("ed25519-individual", 1, None):
            return False, "unsupported signature bundle"
        votes = b["individual_signatures"]
        for signer, m in votes.items():
            if signer not in mb.keys:
                return False, f"signer {signer!r} is not a member"
            if m["msg_type"] != want or m["sender_id"] != signer:
                return False, f"vote from {signer!r} is a {m['msg_type']} by {m['sender_id']!r}, need {want}"
            if (m["cluster_id"], m["epoch"]) != (qc["cluster_id"], qc["epoch"]):
                return False, f"vote from {signer!r} is for another cluster/epoch"
            if t != "VIEW_CHANGE_QC" and (m["sequence"], m["request_digest"]) != (qc["sequence"], qc["proposal_digest"]):
                return False, f"vote from {signer!r} is for another (sequence, digest)"
            if t != "CHECKPOINT_QC" and m["view"] != qc["view"]:
                return False, f"vote from {signer!r} is for view {m['view']}"
            if not ed25519_ok(mb.keys[signer], m["signature_b64"], msg_digest(m).encode()):
                return False, f"signature of {signer!r} does not verify"
        signers = set(votes)
        if t in LEADER_TYPES:
            lp = qc.get("leader_proposal")
            if not lp:
                return False, f"{t} without leader proposal"
            if lp["msg_type"] != "PRE_PREPARE" or (lp["view"], lp["sequence"], lp["request_digest"], lp["cluster_id"]) \
                    != (qc["view"], qc["sequence"], qc["proposal_digest"], qc["cluster_id"]):
                return False, "leader proposal does not match the certificate"
            if lp["sender_id"] != mb.primary_for(qc["view"]):
                return False, f"leader proposal not from the primary of view {qc['view']}"
            if proposal_request_digest(lp["payload"]) != qc["proposal_digest"]:
                return False, "proposal digest does not bind the proposed operation"
            if not ed25519_ok(mb.keys[lp["sender_id"]], lp["signature_b64"], msg_digest(lp).encode()):
                return False, "leader proposal signature does not verify"
            signers.add(lp["sender_id"])
        need = mb.quorum
        if t == "FAST_COMMIT_QC":
            if not mb.fast_path:
                return False, "FAST_COMMIT_QC but the membership has no fast path"
            need = mb.n
        if qc["quorum"] != need:
            return False, f"certificate claims quorum {qc['quorum']}, membership requires {need}"
        if len(signers) < need:
            return False, f"{len(signers)} distinct signers, need {need}"
        if qc.get("certificate_hash") != qc_hash(qc):
            return False, "certificate_hash does not match content"
        if t in CHAINED_TYPES and qc.get("decision_hash") != qc_decision_hash(qc):
            return False, "decision_hash does not match content"
        return True, ""
    except (KeyError, TypeError, ValueError, AttributeError) as e:
        return False, f"malformed: {type(e).__name__}: {e}"


def verify_auth_cert(cert: Dict[str, Any], mb: Membership) -> Tuple[bool, str]:
    try:
        body, qc, req = cert["body"], cert["commit_qc"], cert["request"]
        if qc["certificate_type"] not in DECIDING_TYPES:
            return False, "not backed by a COMMIT_QC or FAST_COMMIT_QC"
        ok, why = verify_qc(qc, mb)
        if not ok:
            return False, f"consensus certificate: {why}"
        if proposal_request_digest(req) != qc["proposal_digest"]:
            return False, "request does not hash to the committed digest"
        op = req["operation"]
        expect = {"domain": AUTH_DOMAIN, "cluster_id": qc["cluster_id"], "epoch": qc["epoch"], "view": qc["view"],
                  "sequence": qc["sequence"], "request_digest": qc["proposal_digest"],
                  "decision_hash": qc_decision_hash(qc), "consensus_certificate_hash": qc_hash(qc),
                  "authorized_action": {k: op.get(k) for k in ("action", "resource", "data")},
                  "action_hash": h({"action": op.get("action"), "resource": op.get("resource"), "data": op.get("data")})}
        for k, v in expect.items():
            if body.get(k) != v:
                return False, f"body field {k} does not match the committed request/certificate"
        if body["identity"]["client_id"] != req["client_id"]:
            return False, "identity does not match the request"
        if h({k: v for k, v in body.items() if k != "certificate_id"}) != body["certificate_id"]:
            return False, "certificate_id does not match the body"
        if body["issuer"] not in mb.keys or not ed25519_ok(mb.keys[body["issuer"]], cert["issuer_signature_b64"],
                                                           body["certificate_id"].encode()):
            return False, "issuer signature does not verify"
        return True, ""
    except (KeyError, TypeError, ValueError, AttributeError) as e:
        return False, f"malformed: {type(e).__name__}: {e}"


# ----------------------------------------------------------------- negative controls
def _tamper(qc: Dict[str, Any], mb: Membership) -> List[Tuple[str, Dict[str, Any]]]:
    out = []
    signers = sorted(qc["bundle"]["individual_signatures"])
    s0 = signers[0]

    t = copy.deepcopy(qc)                          # 1. flip a signature byte
    sig = bytearray(base64.b64decode(t["bundle"]["individual_signatures"][s0]["signature_b64"]))
    sig[0] ^= 1
    t["bundle"]["individual_signatures"][s0]["signature_b64"] = base64.b64encode(bytes(sig)).decode()
    out.append(("forged signature", t))

    t = copy.deepcopy(qc)                          # 2. below quorum
    for s in signers[: len(signers) - mb.quorum + 1]:
        del t["bundle"]["individual_signatures"][s]
    t["certificate_hash"] = qc_hash(t)
    out.append(("below quorum (hash recomputed)", t))

    t = copy.deepcopy(qc)                          # 3. relabel COMMIT votes as PREPARE votes
    t["certificate_type"] = "PREPARE_QC"
    out.append(("COMMIT votes presented as PREPARE (domain separation)", t))

    t = copy.deepcopy(qc)                          # 4. change the decided digest
    t["proposal_digest"] = "0" * 64
    t["certificate_hash"] = qc_hash(t)
    out.append(("different decision digest (hash recomputed)", t))

    t = copy.deepcopy(qc)                          # 5. replay into another cluster
    t["cluster_id"] = t["cluster_id"] + "-other"
    out.append(("cross-cluster replay", t))

    t = copy.deepcopy(qc)                          # 6. replay into another view
    t["view"] = t["view"] + 1
    t["certificate_hash"] = qc_hash(t)
    out.append(("cross-view replay (hash recomputed)", t))

    t = copy.deepcopy(qc)                          # 7. vote altered after signing
    t["bundle"]["individual_signatures"][s0]["timestamp"] += 1
    out.append(("vote content altered after signing", t))
    return out


# ----------------------------------------------------------------- main
def load(src: str) -> Dict[str, Any]:
    if src.startswith("http://") or src.startswith("https://"):
        with urllib.request.urlopen(src, timeout=30) as r:
            return json.loads(r.read())
    with open(src, "rb") as f:
        return json.load(f)


def verify_bundle(bundle: Dict[str, Any]) -> Dict[str, Any]:
    checks: List[Dict[str, Any]] = []

    def check(name: str, ok: bool, detail: str = "") -> None:
        checks.append({"check": name, "result": "PASS" if ok else "FAIL", "detail": detail})

    mb = Membership(bundle["membership"])
    check("membership: 4 members, f=1, quorum 3", (mb.n, mb.f, mb.quorum, len(mb.members)) == (4, 1, 3, 4),
          f"n={mb.n} f={mb.f} quorum={mb.quorum}")
    check("membership: configuration_hash recomputed", mb.configuration_hash == bundle["configuration_hash"],
          mb.configuration_hash)

    chains: Dict[str, List[Tuple[int, str, str]]] = {}
    n_qc = 0
    for node_id, node in sorted(bundle["nodes"].items()):
        check(f"{node_id}: reports the same configuration_hash", node["configuration_hash"] == mb.configuration_hash)
        parent, chain = genesis(mb.cluster_id, mb.epoch), []
        for i, entry in enumerate(node["certificates"], start=1):
            cq, pq = entry["commit_qc"], entry["prepare_qc"]
            ok_c, why_c = verify_qc(cq, mb, expected_parent=parent)
            ok_p, why_p = verify_qc(pq, mb, expected_parent=parent)
            same = (cq["sequence"], cq["proposal_digest"]) == (pq["sequence"], pq["proposal_digest"]) == (i, cq["proposal_digest"])
            if not (ok_c and ok_p and same):
                check(f"{node_id}: sequence {i}", False, why_c or why_p or "commit/prepare certificates disagree or gap")
                break
            n_qc += 2
            parent = qc_decision_hash(cq)
            chain.append((i, cq["proposal_digest"], parent))
        else:
            check(f"{node_id}: {len(chain)} COMMIT_QC + {len(chain)} PREPARE_QC verified, chain contiguous from genesis",
                  len(chain) > 0, f"head {parent[:16]}")
        chains[node_id] = chain
        check(f"{node_id}: stated head_decision_hash matches the folded chain",
              bool(chain) and node.get("head_decision_hash") == chain[-1][2])

    heads = {tuple(c) for c in map(tuple, chains.values())}
    check("all nodes: identical decision chain (same sequence -> same digest -> same decision_hash)",
          len(heads) == 1 and all(chains.values()),
          f"{len(next(iter(chains.values())))} decisions" if len(heads) == 1 else "DIVERGED")

    states = {n["application_state_hash"] for n in bundle["nodes"].values()}
    check("all nodes: identical application state hash after the run", len(states) == 1, next(iter(states))[:16])

    views = sorted({e["commit_qc"]["view"] for n in bundle["nodes"].values() for e in n["certificates"]})
    check("run spans a view change (decisions committed in more than one view)", len(views) > 1, f"views {views}")
    for vc in bundle.get("view_change_qcs", []):
        ok, why = verify_qc(vc, mb)
        check(f"VIEW_CHANGE_QC for view {vc.get('view')}: signatures + quorum", ok, why)

    for ac in bundle.get("authorization_certificates", []):
        ok, why = verify_auth_cert(ac, mb)
        check(f"AuthorizationCertificate seq {ac['body'].get('sequence')} issued by {ac['body'].get('issuer')}", ok, why)

    entries = next(iter(bundle["nodes"].values()))["certificates"]
    normal = [e["commit_qc"] for e in entries if e["commit_qc"]["certificate_type"] == "COMMIT_QC"]
    sample = normal[-1] if normal else entries[-1]["commit_qc"]
    fast = sum(1 for n in bundle["nodes"].values() for e in n["certificates"]
               if e["commit_qc"]["certificate_type"] == "FAST_COMMIT_QC")
    if mb.fast_path or fast:
        check("fast-path certificates only on a fast-path membership", mb.fast_path or not fast,
              f"{fast} FAST_COMMIT_QC across nodes")
    for name, t in _tamper(sample, mb):
        ok, why = verify_qc(t, mb)
        check(f"negative control: {name} -> rejected", not ok, why)

    failed = [c for c in checks if c["result"] == "FAIL"]
    return {"verifier": "verify_pbft_qc_bundle.py (no CAIN imports)", "certificates_verified": n_qc,
            "checks": checks, "passed": len(checks) - len(failed), "total": len(checks),
            "verdict": "ALL_CHECKS_PASSED" if not failed else "FAILED"}


def main(argv: List[str]) -> int:
    if len(argv) < 2:
        print(__doc__)
        return 2
    res = verify_bundle(load(argv[1]))
    if "--json" in argv:
        print(json.dumps(res, indent=1))
    else:
        for c in res["checks"]:
            print(f"[{c['result']}] {c['check']}" + (f"  ({c['detail']})" if c["detail"] else ""))
        print(f"\n{res['verdict']}: {res['passed']}/{res['total']} checks, "
              f"{res['certificates_verified']} certificates verified")
    return 0 if res["verdict"] == "ALL_CHECKS_PASSED" else 1


if __name__ == "__main__":
    sys.exit(main(sys.argv))