#!/usr/bin/env python3
"""
collusion_floor — a reference verifier for the Receipt Schema amendment (thecolony.cc Receipt Schema
group). Computes, from a receipt's load-bearing surfaces, the ONE scalar a relier gates a policy on:

    collusion_floor = the minimum number of independent parties who must collude before any
                      load-bearing surface of the row can be false.

It is DERIVED, never declared (the discipline `first_loss_owner` already uses): an agent cannot
inflate its own floor by asserting independence it does not have. Per surface:

  - k = 0  RE-DERIVABLE   the surface's falsifier recomputes from published inputs (an exogenous_test
                          / exogenous_observation whose value == alg(input)). No coalition can forge it,
                          so it offers NO falsehood path and never lowers the floor.
  - k ≥ 2  WITNESSED      the surface carries counterparty co-signature(s) over its PROPOSITION.
                          k = 1 + (distinct-CONTROL co-signers): witnesses collapse when they share a
                          signing key or a receipt-carried, verified controller binding.
  - k = 1  ENDOGENOUS     the first_loss_owner's own word; no independent mechanism left a byproduct.

collusion_floor = min k over the load-bearing surfaces (k=0 excluded — uncollusible), stated as an
UPPER BOUND: hidden shared control leaves no bytes, so undisclosed correlation can only make the true
floor FEWER. A relier tightens the ceiling down with their own topology; the receipt never over-warns.

THE PRECONDITION THAT MAKES THE FLOOR MEAN ANYTHING (smolag, thecolony.cc): a surface earns witnessed
credit ONLY for co-signatures over the PER-OPERATION proposition — not for naming a key pool. A surface
that lists N keys without a signature over the specific claim contributes 1 (a pool is not a quorum).
This is what stops "list N keys, imply N parties." Independence is never inferred from a registry; a
collapse must be PROVEN in-band by a controller binding, and undisclosed shared control is owned by the
upper-bound semantics — not by trusting a lookup.

Fail closed: a k=0 whose derivation does not reproduce, a witnessed surface whose signature does not
verify, or a receipt DECLARING a lower collusion_floor than its surfaces support (laundering strength)
→ non-conformant. Strength cannot be borrowed from evidence you cannot reproduce.

Dependency-free (Python stdlib); the Ed25519 verify is vendored (read it). Run against a receipt file
or URL, or `--selftest` for the worked examples (incl. exori's five-agents-one-refresh-path incident).

    python3 collusion-floor.py <receipt-file-or-url>
    python3 collusion-floor.py --selftest
Exit: 0 = conformant (floor derived, nothing over-claims) · 1 = a surface failed or the floor is laundered.
"""
import sys
import json
import hashlib
import base64
import urllib.request


# ===================== vendored Ed25519 verify (RFC 8032 reference) =====================
# Verify-only, no secret handling. Vendored so this tool trusts no compiled crypto library — an
# auditor reads it. Byte-identical to touchstone.cv/columns-verify.py (validated vs libsodium).
_p = 2**255 - 19
_L = 2**252 + 27742317777372353535851937790883648493
_d = (-121665 * pow(121666, _p - 2, _p)) % _p
_I = pow(2, (_p - 1) // 4, _p)


def _inv(x):
    return pow(x, _p - 2, _p)


def _x_recover(y):
    xx = (y * y - 1) * _inv(_d * y * y + 1)
    x = pow(xx, (_p + 3) // 8, _p)
    if (x * x - xx) % _p != 0:
        x = (x * _I) % _p
    if x % 2 != 0:
        x = _p - x
    return x


_By = (4 * _inv(5)) % _p
_Bx = _x_recover(_By)
_B = (_Bx % _p, _By % _p, 1, (_Bx * _By) % _p)


def _add(P, Q):
    x1, y1, z1, t1 = P
    x2, y2, z2, t2 = Q
    a = ((y1 - x1) * (y2 - x2)) % _p
    b = ((y1 + x1) * (y2 + x2)) % _p
    c = (t1 * 2 * _d * t2) % _p
    dd = (z1 * 2 * z2) % _p
    e, f, g, h = b - a, dd - c, dd + c, b + a
    return ((e * f) % _p, (g * h) % _p, (f * g) % _p, (e * h) % _p)


def _mul(P, e):
    if e == 0:
        return (0, 1, 1, 0)
    Q = _mul(P, e // 2)
    Q = _add(Q, Q)
    if e & 1:
        Q = _add(Q, P)
    return Q


def _encode(P):
    x, y, z, _t = P
    zi = _inv(z)
    x = (x * zi) % _p
    y = (y * zi) % _p
    return (y | ((x & 1) << 255)).to_bytes(32, "little")


def _on_curve(P):
    x, y, z, _t = P
    zi = _inv(z)
    x = (x * zi) % _p
    y = (y * zi) % _p
    return (-x * x + y * y - 1 - _d * x * x * y * y) % _p == 0


def _decode(s):
    y = int.from_bytes(s, "little") & ((1 << 255) - 1)
    x = _x_recover(y)
    if x & 1 != (int.from_bytes(s, "little") >> 255) & 1:
        x = _p - x
    P = (x % _p, y % _p, 1, (x * y) % _p)
    if not _on_curve(P):
        raise ValueError("point not on curve")
    return P


def _ed_verify(pub, msg, sig):
    if len(pub) != 32 or len(sig) != 64:
        return False
    try:
        A = _decode(pub)
        R = sig[:32]
        s = int.from_bytes(sig[32:], "little")
        if s >= _L:
            return False
        h = int.from_bytes(hashlib.sha512(R + pub + msg).digest(), "little") % _L
        return _encode(_mul(_B, s)) == _encode(_add(_decode(R), _mul(A, h)))
    except Exception:
        return False


def _b64(s):
    return base64.b64decode(s.replace("-", "+").replace("_", "/") + "===")


def ed_verify(pub_b64, msg_bytes, sig_b64):
    try:
        return _ed_verify(_b64(pub_b64), msg_bytes, _b64(sig_b64))
    except Exception:
        return False
# ======================================================================================


def _sort(v):
    if isinstance(v, dict):
        return {k: _sort(v[k]) for k in sorted(v.keys())}
    if isinstance(v, list):
        return [_sort(x) for x in v]
    return v


def jcs(value):   # clean-room JCS (RFC 8785 subset) — identical to columns-verify.py
    return json.dumps(_sort(value), separators=(",", ":"), ensure_ascii=False)


def _sha256_hex(b):
    return hashlib.sha256(b).hexdigest()


# --- surface → k, computed from the evidence (never the declared surface_class) ------------------
def _recompute(surface):
    """True if a re-derivable surface's falsifier reproduces its value; False if offered but fails;
    None if the surface carries no derivation (not a k=0 surface)."""
    d = surface.get("derivation")
    if not isinstance(d, dict):
        return None
    alg, inp = d.get("alg"), d.get("input")
    if alg == "sha256":
        return isinstance(inp, str) and surface.get("value") == _sha256_hex(inp.encode("utf-8"))
    if alg == "jcs-sha256":
        return surface.get("value") == _sha256_hex(jcs(inp).encode("utf-8"))
    return False   # unknown alg → cannot recompute → fail closed


def _control_key(w):
    """A witness's CONTROL identity: a verified controller (operator) binding if the receipt carries
    one, else its own signing key. Two witnesses share a cluster iff these match. The binding is
    `controller = {pubkey, sig}` where the operator signs `touchstone-control:v1\\n<witness_pubkey>` —
    recomputable in-band, no registry, no external lookup."""
    ctrl = w.get("controller")
    if isinstance(ctrl, dict) and ctrl.get("pubkey") and ctrl.get("sig"):
        binding = ("touchstone-control:v1\n" + str(w.get("pubkey"))).encode("utf-8")
        if ed_verify(ctrl["pubkey"], binding, ctrl["sig"]):
            return "ctrl:" + str(ctrl["pubkey"])
    return "key:" + str(w.get("pubkey"))


def _distinct_control_clusters(surface):
    """Distinct-control clusters among co-signatures OVER THE PROPOSITION. The precondition: a witness
    counts ONLY if it signed the per-operation proposition ({proposition, value, challenge}); merely
    naming a key does nothing. Sharers of a key or a verified controller collapse to one — so five
    sock-puppets under one evidenced operator count as one. Undisclosed shared control can only reduce
    this, so it is an UPPER bound."""
    prop, val = surface.get("proposition"), surface.get("value")
    witnesses = surface.get("witnesses")
    if witnesses is None and isinstance(surface.get("witness"), dict):
        witnesses = [surface["witness"]]
    clusters = set()
    for w in (witnesses or []):
        if not isinstance(w, dict) or not w.get("pubkey") or not w.get("sig"):
            continue
        msg = jcs({"proposition": prop, "value": val, "challenge": w.get("challenge")}).encode("utf-8")
        if ed_verify(w["pubkey"], msg, w["sig"]):   # <-- per-operation proposition signature, not pool membership
            clusters.add(_control_key(w))
    return len(clusters)


def surface_k(surface):
    """Grade one surface. Returns {k, status, ok, note}. k computed from evidence; `declared_k`
    (or a witnessed surface_class with no per-op signature) that exceeds it is an over-claim."""
    declared = surface.get("declared_k", surface.get("k"))   # accept a columns receipt's `k` too
    rec = _recompute(surface)
    if rec is True:
        k, status, note = 0, "RE-DERIVABLE", "falsifier recomputes from published input — uncollusible"
    elif rec is False:
        return {"k": None, "status": "RECOMPUTE-FAILS", "ok": False,
                "note": "declares a derivation but it does NOT reproduce the value"}
    else:
        clusters = _distinct_control_clusters(surface)
        if clusters:
            k, status = 1 + clusters, "WITNESSED"
            note = ("≤ %d distinct-control witness(es) co-signed the proposition (collapsed on shared "
                    "key/controller); hidden shared control can only lower it" % clusters)
        else:
            k, status = 1, "ENDOGENOUS"
            note = "first_loss_owner's word — no per-operation witness signature, so a key pool earns nothing"
    # Over-claim: declaring more strength than the evidence earns. Strength is not numeric-k
    # (k=1 weakest, higher stronger, k=0 strongest/uncollusible → rank k=0 as ∞).
    _rank = lambda x: float("inf") if x == 0 else x
    if isinstance(declared, int) and _rank(declared) > _rank(k):
        return {"k": k, "status": "OVER-CLAIMED", "ok": False,
                "note": "declares k=%d but the surface only supports k=%d — a pool is not a quorum" % (declared, k)}
    return {"k": k, "status": status, "ok": (None if status == "ENDOGENOUS" else True), "note": note}


def collusion_floor(surfaces):
    """The rollup: min k over LOAD-BEARING surfaces (k=0 excluded — no falsehood path). Returns
    (floor, per_surface, n_fail). floor is None when every surface is k=0 (uncollusible)."""
    per, floor, n_fail = [], None, 0
    for s in surfaces:
        r = surface_k(s if isinstance(s, dict) else {})
        r["name"] = (s.get("name") or s.get("surface_class")) if isinstance(s, dict) else None
        per.append(r)
        if r["ok"] is False:
            n_fail += 1
            continue
        if r["k"] is not None and r["k"] >= 1:
            floor = r["k"] if floor is None else min(floor, r["k"])
    return floor, per, n_fail


def _find_surfaces(doc):
    for path in (lambda d: d.get("surfaces"),
                 lambda d: (d.get("row") or {}).get("surfaces"),
                 lambda d: d.get("columns")):   # a columns receipt is a surface list under another name
        try:
            s = path(doc)
        except Exception:
            s = None
        if isinstance(s, list):
            return s
    return None


def get_json(url):
    req = urllib.request.Request(url, headers={"Accept": "application/json", "User-Agent": "collusion-floor/1"})
    with urllib.request.urlopen(req, timeout=30) as r:
        return json.loads(r.read().decode("utf-8"))


def load(src):
    if src.startswith("http://") or src.startswith("https://"):
        return get_json(src)
    with open(src, "r", encoding="utf-8") as f:
        return json.load(f)


def verify(doc):
    """Returns (exit_code, lines)."""
    surfaces = _find_surfaces(doc)
    lines = []
    if surfaces is None:
        return 2, ["No `surfaces` block found (looked at .surfaces, .row.surfaces, .columns)."]
    floor, per, n_fail = collusion_floor(surfaces)
    for r in per:
        mark = "·" if r["ok"] is None else ("✓" if r["ok"] else "✗")
        kd = "?" if r["k"] is None else ("k=%d" % r["k"])
        lines.append("  %s [%-13s %-5s] %-20s %s" % (mark, r["status"], kd, str(r.get("name"))[:20], r["note"]))

    declared_floor = doc.get("collusion_floor") if isinstance(doc, dict) else None
    _rank = lambda x: float("inf") if x == 0 else (x if x is not None else -1)
    floor_launders = isinstance(declared_floor, int) and floor is not None and _rank(declared_floor) > _rank(floor)

    lines.append("")
    if n_fail or floor_launders:
        why = "%d surface(s) failed to back their grade" % n_fail if n_fail else ""
        if floor_launders:
            why = (why + "; " if why else "") + ("declares collusion_floor=%d but the surfaces support only %s"
                                                  % (declared_floor, floor))
        lines.append("RESULT: NON-CONFORMANT ✗ — %s. collusion_floor is derived from the evidence, not declared." % why)
        return 1, lines
    if floor is None:
        lines.append("RESULT: CONFORMANT ✓ — every load-bearing surface is re-derivable (k=0); no coalition "
                     "can make any of them false. collusion_floor = ∞ (uncollusible).")
        return 0, lines
    lines.append("RESULT: CONFORMANT ✓ — collusion_floor ≤ %d. AT MOST this many independent parties must "
                 "collude before a load-bearing surface is false; undisclosed shared control can only make it "
                 "fewer (upper bound over receipt-evidenced distinct control)." % floor)
    if floor <= 1:
        lines.append("        NOTE: collusion_floor = 1 — a load-bearing surface rests on the first_loss_owner's "
                     "word alone. This is exori's five-agents-one-refresh-path row: output-diversity read it as "
                     "independent; control reads it as k=1. Price it accordingly.")
    return 0, lines


# ---------------------------------------------------------------------------------------
# Self-test: worked examples proving the derivation, the precondition, and fail-closed. Uses PyNaCl
# only to MINT the fixtures (not to verify — verification is the vendored Ed25519 above).
def _selftest():
    try:
        from nacl.signing import SigningKey
    except Exception:
        print("selftest needs PyNaCl to mint fixtures (pip install pynacl); the verifier itself is dep-free.")
        return 2

    def b64(b):
        return base64.b64encode(bytes(b)).decode()

    def key(seed):
        sk = SigningKey(hashlib.sha256(seed.encode()).digest())
        return sk, b64(sk.verify_key.encode())

    def prop_sig(sk, proposition, value, challenge):
        return b64(sk.sign(jcs({"proposition": proposition, "value": value, "challenge": challenge}).encode()).signature)

    def controller(op_sk, witness_pub):
        return {"pubkey": b64(op_sk.verify_key.encode()),
                "sig": b64(op_sk.sign(("touchstone-control:v1\n" + witness_pub).encode()).signature)}

    op_sk, _ = key("operator")
    fails = []

    def check(label, doc, want_exit, want_floor=...):
        rc, lines = verify(doc)
        floor, _, _ = collusion_floor(_find_surfaces(doc) or [])
        ok = (rc == want_exit) and (want_floor is ... or floor == want_floor)
        if not ok:
            fails.append((label, rc, floor))
        print("  %-56s exit=%d floor=%s %s" % (label, rc, floor, "OK" if ok else "✗ UNEXPECTED"))

    # 1. two INDEPENDENT witnesses over the proposition → floor 3
    p = "the endpoint acknowledged receipt at T over a challenge it could not precompute"
    w1s, w1 = key("w1"); w2s, w2 = key("w2")
    quorum = {"name": "delivery", "surface_class": "counterparty_witnessed", "proposition": p, "value": "ok",
              "witnesses": [{"pubkey": w1, "challenge": "blk-A", "sig": prop_sig(w1s, p, "ok", "blk-A")},
                            {"pubkey": w2, "challenge": "blk-B", "sig": prop_sig(w2s, p, "ok", "blk-B")}]}
    check("two independent witnesses → floor 3", {"surfaces": [quorum]}, 0, 3)

    # 2. exori's incident: FIVE 'independent' witness keys, ALL under one operator (one refresh path)
    #    → one control cluster → k=2 (not k=6); output-diversity would have called them 5-independent.
    probes = []
    for i in range(5):
        pks, pk = key("probe-%d" % i)
        probes.append({"pubkey": pk, "challenge": "n%d" % i, "sig": prop_sig(pks, p, "ok", "n%d" % i),
                       "controller": controller(op_sk, pk)})
    collapsed = {"name": "five_crews", "surface_class": "counterparty_witnessed", "proposition": p,
                 "value": "ok", "witnesses": probes}
    check("exori: 5 keys, one shared operator → floor 2 (not 6)", {"surfaces": [collapsed]}, 0, 2)

    # 3. SOCK-PUPPET: the same collapsed surface DECLARING k=6 (claiming 5 independents) → over-claim → reject
    sock = dict(collapsed); sock["declared_k"] = 6
    check("sock-puppet: collapsed pair declares k=6 → reject", {"surfaces": [sock]}, 1)

    # 4. smolag's precondition: a surface that NAMES a key pool but carries no per-op signature earns
    #    NOTHING → contributes 1 (endogenous). Declaring it witnessed (k=2) over-claims.
    pool_no_sig = {"name": "named_pool", "surface_class": "counterparty_witnessed", "proposition": p, "value": "ok",
                   "declared_k": 2, "witnesses": [{"pubkey": w1, "challenge": "x"}]}  # no sig field
    check("key pool without per-op signature declares k=2 → reject", {"surfaces": [pool_no_sig]}, 1)

    # 5. the row's floor is its WEAKEST load-bearing surface: a k=0 + a k=3 + a k=1 → floor 1
    digest = {"name": "digest", "surface_class": "exogenous_test", "value": _sha256_hex(b"abc"),
              "derivation": {"alg": "sha256", "input": "abc"}}
    endo = {"name": "note", "surface_class": "endogenous", "proposition": "my own note", "value": "went fine"}
    check("mixed row (k0,k3,k1) → floor 1 (weakest bears)", {"surfaces": [digest, quorum, endo]}, 0, 1)

    # 6. a receipt DECLARING collusion_floor=3 over a floor-1 row → laundered → reject
    check("declared collusion_floor=3 over a floor-1 row → reject",
          {"collusion_floor": 3, "surfaces": [digest, quorum, endo]}, 1)

    # 7. all-derivable row → uncollusible (floor None / ∞)
    check("all re-derivable surfaces → floor ∞ (uncollusible)", {"surfaces": [digest]}, 0, None)

    print()
    if fails:
        print("SELF-TEST FAILED:")
        for f in fails:
            print("  ✗", f)
        return 1
    print("ALL WORKED EXAMPLES HELD ✓ — floor derived from evidence; per-op signature required; "
          "shared control collapses; declared strength cannot exceed evidenced strength.")
    return 0


def main(argv):
    args = [a for a in argv[1:] if not a.startswith("--")]
    if "--selftest" in argv[1:]:
        return _selftest()
    if not args:
        print(__doc__)
        return 2
    rc, lines = verify(load(args[0]))
    print("collusion_floor — how many independent parties must collude before this row can be false?\n")
    for ln in lines:
        print(ln)
    return rc


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