#!/usr/bin/env python3 """ equivocation-check — bind a disclosure's recorder to an operator's SINGLE committed set of chains. Every other Touchstone verifier proves a record is UNALTERED. None of them make this checkable: "has this operator quietly started a SECOND chain and shown it only to me?" A recorder can be internally perfect — every entry hash-linked, checkpointed, anchored to Bitcoin, its feed gossiping cleanly to Nostr — and still be a shadow: one operator running two genesis chains, disclosing whichever suits the relier in front of it. Per-feed split-view detection (gossip_check.py) cannot catch this, because each feed is internally consistent; the choice of WHICH chain to show happens above the record. Integrity survives; UNIQUENESS does not. The bind has to be at the identity layer. An operator signs a commitment {v, kind:"touchstone.operator_recorders", operator_sub, recorders:[{public_id, genesis_hash}…]} into its OWN recorder, so the SET rides the same append-only → checkpoint → Bitcoin chain. A second, conflicting commitment under one identity is a fork the gossip checker catches; a recorder shown to a relier but ABSENT from the committed set is an unregistered — possible shadow — chain. This tool, given a disclosure's (sub, recorder), fetches the operator's commitment and checks: 1. commitment body → payload_hash, and the entry_hash, recompute; 2. the entry's actor_sig verifies against the identity recorder's signing key (the operator's key), which /pubkeys reports as a self-operated (verified) binding of the sub — so it is the operator speaking, not Touchstone; 3. the commitment entry folds through its Merkle inclusion proof to an anchored checkpoint root; 4. the queried recorder appears in commitment.recorders with a genesis_hash equal to that recorder's own seq-0 entry_hash (fetched independently) — BOUND. Absent → UNREGISTERED. Present but wrong genesis → GENESIS_MISMATCH. 5. single head: the identity recorder must not have published a second conflicting commitment. That is a cross-vantage (Nostr) property — delegated to gossip_check.py, run here when it is alongside; a fork it reports is surfaced as EQUIVOCATION. Every RED verdict is exercised by --selftest (with PyNaCl): a valid commitment verifies BOUND, then a forged signature / tampered body / swapped genesis / absent recorder / gossip fork each flips it to the correct refusal. A detector that has only ever been shown its passing case is not a detector yet. HONEST RESIDUAL (this does not pretend to close it): an operator can still OMIT a recorder from its commitment — but then that chain's disclosures read UNREGISTERED to anyone who runs this, so a hidden chain cannot be used to convince a checking relier. It cannot be both hidden and usable. And the whole bind rests on gossip: a relier must consult the same independent vantage (Nostr) for the anchor + the single-head check to mean anything. Fork consistency is not reachable by the recorder alone — it takes reliers comparing heads, or witnesses cosigning them. This makes the operator's set checkable; it does not make one relier's private view provably complete. Reuses standing-verify.py (jcs / ed_verify / fold_proof / recompute_entry_hash / get_json) — fetch it alongside — and runs verifier/gossip_check.py for the single-head leg when present. python3 equivocation-check.py --sub= --recorder= [--base=https://touchstone.cv] Exit: 0 = BOUND · 1 = UNREGISTERED / GENESIS_MISMATCH / EQUIVOCATION / invalid commitment · 2 = no commitment (only the server-attested /pubkeys list binds the set) / malformed / missing sibling. """ import os import sys import copy import hashlib import subprocess import importlib.util import urllib.error _HERE = os.path.dirname(os.path.abspath(__file__)) OP_EVENT = "touchstone.operator_recorders" def _load_sibling(filename): path = os.path.join(_HERE, filename) if not os.path.exists(path): raise FileNotFoundError(filename) spec = importlib.util.spec_from_file_location(filename.replace("-", "_").replace(".py", ""), path) mod = importlib.util.module_from_spec(spec) spec.loader.exec_module(mod) return mod def op_signed_content(sv, recorder_id, operator_sub, payload_hash): """The exact bytes actor_sig covers (Hashing::signedContent) for an operator_recorders entry.""" return sv.jcs({ "v": 1, "recorder_id": recorder_id, "event_type": OP_EVENT, "actor_sub": operator_sub, "counterparty_sub": None, "payload_hash": payload_hash, "client_ts": None, }).encode("utf-8") def genesis_hash(sv, base, recorder): """That recorder's own seq-0 entry_hash, fetched + recomputed independently (never taken on faith). Returns None if the seq-0 proof can't be fetched (e.g. the recorder didn't opt into public inclusion proofs) — the caller degrades to GENESIS_UNPROVABLE rather than crashing.""" try: d = sv.get_json("%s/.well-known/touchstone/checkpoints/%s/entry/0" % (base, recorder)) except Exception: return None entry = d.get("entry") or {} got = entry.get("entry_hash") if got and sv.recompute_entry_hash(entry) != got: return None # the served genesis envelope does not hash to its own claimed entry_hash return got def run_gossip(identity_recorder, base): """Best-effort cross-vantage single-head check via gossip_check.py. Returns (state, detail). 'consistent' gossip_check exit 0 — one head across touchstone.cv + Nostr 'fork' exit 1 — a second conflicting head/commitment exists (EQUIVOCATION) 'incomplete' exit 2 — could not fetch enough to decide 'skipped' gossip_check.py not found alongside — run it yourself (command in detail) """ for cand in ("gossip_check.py", os.path.join("verifier", "gossip_check.py"), os.path.join(_HERE, "..", "verifier", "gossip_check.py")): p = cand if os.path.isabs(cand) else os.path.join(_HERE, cand) if os.path.exists(p): try: r = subprocess.run([sys.executable, p, identity_recorder, "--base", base], capture_output=True, text=True, timeout=90) last = (r.stdout or r.stderr or "").strip().splitlines() return ({0: "consistent", 1: "fork"}.get(r.returncode, "incomplete"), last[-1] if last else "") except Exception as ex: return "incomplete", "gossip_check.py error (%s)" % ex return "skipped", "python3 gossip_check.py %s --base %s" % (identity_recorder, base) def clock_note(entry): """Which direction is this entry's server_ts actually bound in? The Bitcoin anchor gives a NOT-AFTER: the commitment existed by that block. Nothing there gives a NOT-BEFORE. Only a committed drand round does — its randomness could not have been known earlier, so server_ts cannot precede it. Beacon-binding is OPT-IN, which means an unbound server_ts is the OPERATOR'S CHOICE, and an unbound timestamp is BACKDATABLE. And "when was the recorder set committed?" is exactly what an equivocator wants to lie about. So a bare `commitment recorded ` is k=1 testimony printed as though it were checked — the failure beacon-verify names: "stopping leaves you at k=1 while LOOKING checked, worse than an honest soft clock." Name the clock. """ if (entry or {}).get("server_beacon"): return (" · server_ts is BOUND on both sides: a committed drand round gives the not-before, " "the Bitcoin anchor the not-after. A real interval — check it with beacon-verify.py.") return (" ⚠ server_ts is an OPERATOR-WRITTEN clock (no beacon bound to this entry). The anchor " "bounds it from ABOVE only (not-after); nothing bounds it from below, so it can be " "BACKDATED. Beacon-binding is opt-in — its absence is the operator's choice, not a limit of " "the record. Read the time as testimony; the anchor is the only checkable half.") def evaluate(sv, sub, recorder, bindings, oc, resolve_genesis, gossip_fn): """PURE verdict logic (no I/O of its own): given the fetched /pubkeys bindings, the operator commitment doc `oc` (or None ⇒ UNCOMMITTED), a `resolve_genesis(recorder)->hash|None` and a `gossip_fn(identity_recorder)->(state,detail)`, decide the verdict. Kept separate from verify() so the adversarial battery in --selftest can drive every RED path with crafted inputs, no network. Returns (exit_code, verdict, lines).""" lines = [] if oc is None: return 2, "UNCOMMITTED", [ "no operator commitment published for this sub — only the SERVER-ATTESTED /pubkeys list " "binds the set, which Touchstone could serve differently to another relier. Cannot " "establish that %s is not a shadow chain. (Weakest tier: unregistered-by-absence is " "indistinguishable from never-committed here.)" % recorder] body, entry = oc.get("commitment") or {}, oc.get("entry") or {} id_rec, sign_pk = oc.get("identity_recorder"), oc.get("signing_pubkey") # 1-2. the commitment is the OPERATOR speaking, unaltered. ph_ok = hashlib.sha256(sv.jcs(body).encode("utf-8")).hexdigest() == entry.get("payload_hash") eh_ok = sv.recompute_entry_hash(entry) == entry.get("entry_hash") sig_ok = sv.ed_verify(sign_pk, op_signed_content(sv, id_rec, sub, entry.get("payload_hash", "")), entry.get("actor_sig", "")) idb = bindings.get(id_rec) id_selfop = bool(idb and idb.get("verified") and idb.get("signing_pubkey") == sign_pk) lines.append(" %s commitment body → payload_hash" % ("✓" if ph_ok else "✗")) lines.append(" %s entry_hash recompute" % ("✓" if eh_ok else "✗")) lines.append(" %s actor_sig verifies against the identity key" % ("✓" if sig_ok else "✗")) lines.append(" %s identity recorder is a self-operated (verified) binding of the sub" % ("✓" if id_selfop else "✗")) if not (ph_ok and eh_ok and sig_ok and id_selfop): return 1, "INVALID_COMMITMENT", lines + ["the commitment does not authenticate as the operator's own " "unaltered word — do not rely on it."] # 3. anchored to Bitcoin (fold to the checkpoint root; `ots verify` finishes the walk). cp, proof = oc.get("checkpoint"), oc.get("inclusion_proof") anchored = bool(cp and proof is not None and sv.fold_proof(entry.get("entry_hash"), proof) == cp.get("merkle_root")) if anchored: btc = cp.get("bitcoin") or {} lines.append(" ✓ folds to checkpoint #%s root %s… %s" % ( cp.get("id"), (cp.get("merkle_root") or "")[:12], ("(Bitcoin block %s — `ots verify` to finish)" % btc.get("height")) if btc.get("height") else "(checkpoint not yet Bitcoin-confirmed — check back)")) else: lines.append(" · commitment not yet checkpointed — the set is signed but not yet anchored " "(re-run after app:checkpoint; until then it is the operator's word, not Bitcoin's).") # The anchor view this check actually folded to — printed so absence/presence reads AS OF a stated # anchor, not as a permanent claim. The recorder-set commitment rides the operator's OWN chain, so # "is there one committed set?" is the single-head question one level up; the Bitcoin anchor is what # arrests that regress, which makes the checker's honesty about WHICH anchor it saw load-bearing. _btc = (cp or {}).get("bitcoin") or {} asof = "commitment recorded %s · checkpoint #%s · Bitcoin %s" % ( entry.get("server_ts"), (cp or {}).get("id"), ("height %s" % _btc.get("height")) if _btc.get("height") else ("checkpointed, confirmation pending" if anchored else "NOT yet checkpointed")) lines.append(" anchor seen by this check: %s" % asof) lines.append(clock_note(entry)) # 4. membership: is the queried recorder in the committed set, with a matching genesis? committed = {r.get("public_id"): r.get("genesis_hash") for r in (body.get("recorders") or [])} qb = bindings.get(recorder) q_selfop = bool(qb and qb.get("verified")) if recorder not in committed: lines.append(" ✗ %s is NOT in the operator's committed set (as of %s)" % (recorder, asof)) note = ("it IS a self-operated binding of this sub" if q_selfop else "it is not even a self-operated binding of this sub") # Split the two cases anp2network named, so limbo isn't charged at the maximum-suspicion rate: # a firm negative against an anchored view, vs a provisional one against a not-yet-anchored # commitment the operator could still be finishing. if not anchored: return 1, "UNREGISTERED_PENDING", lines + [ "UNREGISTERED_PENDING: absent from the operator's latest commitment, but that commitment is " "NOT yet checkpointed — this is limbo, not a firm negative. The operator could bind this " "recorder in the commitment that anchors next. Re-check after the next checkpoint rather than " "treating it as a shadow yet. (%s; %s)" % (note, asof)] return 1, "UNREGISTERED", lines + [ "UNREGISTERED: absent from the operator's committed set as of %s. A disclosure from it is a chain " "shown to you but not committed — a possible shadow. Read it as 'unregistered as of THIS anchor,' " "not permanently: a recorder bound in a later, not-yet-anchored commitment would still read " "UNREGISTERED here, so if a fresher commitment may exist, re-check against it. (%s)" % (asof, note)] g = resolve_genesis(recorder) if g is None: return 2, "GENESIS_UNPROVABLE", lines + [ "%s is in the set, but its seq-0 entry could not be independently fetched/verified " "(needs inclusion_proofs_public). Cannot confirm the committed genesis matches the live chain." % recorder] if g != committed[recorder]: lines.append(" ✗ genesis mismatch: committed %s… vs live %s…" % ((committed[recorder] or "")[:12], g[:12])) return 1, "GENESIS_MISMATCH", lines + [ "GENESIS_MISMATCH: the committed genesis_hash for %s does not equal its live seq-0 entry_hash — " "the recorder shown to you is a DIFFERENT chain than the one the operator committed. Shadow." % recorder] lines.append(" ✓ %s is in the committed set, genesis matches (self-operated: %s)" % (recorder, q_selfop)) # 5. single head across an independent vantage (Nostr) — delegated to gossip_check. gstate, gdetail = gossip_fn(id_rec) if gstate == "fork": lines.append(" ✗ gossip: a SECOND conflicting head/commitment exists for the identity recorder") return 1, "EQUIVOCATION", lines + ["EQUIVOCATION: the operator published two conflicting histories under " "one identity (gossip_check: %s). The set is not singular." % gdetail] if gstate == "consistent": lines.append(" ✓ gossip: identity recorder is single-headed across touchstone.cv + Nostr") elif gstate == "skipped": lines.append(" · gossip: single-head not checked here — run `%s` to rule out a second commitment " "shown to another relier." % gdetail) else: lines.append(" · gossip: single-head check incomplete (%s) — the anchor binds the set you see; a " "second vantage is still needed to prove no conflicting head exists." % gdetail) tail = "" if anchored else " (signed but not yet Bitcoin-anchored)" if gstate != "consistent": tail += " — single-head leg unconfirmed; confirm with gossip_check.py" return 0, "BOUND", lines + ["BOUND: %s is in %s's single committed, operator-signed recorder set%s. As of %s." % (recorder, sub, tail, asof)] def verify(sub, recorder, base): """Fetch the live artifacts, then delegate to evaluate(). Returns (exit_code, verdict, lines).""" header = ["operator recorder-set commitment — sub %s, recorder %s" % (sub, recorder)] try: sv = _load_sibling("standing-verify.py") except FileNotFoundError: return 2, "MISSING_SIBLING", header + ["fetch standing-verify.py alongside this file (it carries jcs / " "ed_verify / fold_proof / recompute_entry_hash / get_json)."] try: pk = sv.get_json("%s/.well-known/touchstone/pubkeys/%s" % (base, sub)) except Exception as ex: return 2, "UNREACHABLE", header + ["could not fetch /pubkeys/%s (%s)" % (sub, ex)] bindings = {b.get("recorder"): b for b in pk.get("bindings", [])} try: oc = sv.get_json("%s/.well-known/touchstone/operator/%s" % (base, sub)) except urllib.error.HTTPError as ex: if ex.code == 404: oc = None else: return 2, "UNREACHABLE", header + ["could not fetch the commitment (%s)" % ex] except Exception as ex: return 2, "UNREACHABLE", header + ["could not fetch the commitment (%s)" % ex] code, verdict, lines = evaluate( sv, sub, recorder, bindings, oc, lambda r: genesis_hash(sv, base, r), lambda idr: run_gossip(idr, base), ) return code, verdict, header + lines def _kv(argv, key, default=None): for a in argv: if a.startswith(key + "="): return a.split("=", 1)[1] return default def main(argv): if "--selftest" in argv[1:]: return _selftest() sub, rec = _kv(argv[1:], "--sub"), _kv(argv[1:], "--recorder") base = _kv(argv[1:], "--base", "https://touchstone.cv").rstrip("/") if not sub or not rec: print(__doc__) return 2 code, verdict, lines = verify(sub, rec, base) for ln in lines: print(ln) print("\nRESULT: %s" % verdict) return code def _mint_valid(sv, sk_sign, pk_b64): """Build a VALID commitment fixture (real Ed25519 signature), the baseline the battery mutates.""" sub, id_rec, real = "op-sub", "rec_identity", "rec_real" recorders = sorted( [{"public_id": id_rec, "genesis_hash": "GEN_ID"}, {"public_id": real, "genesis_hash": "GEN_REAL"}], key=lambda r: r["public_id"]) body = {"v": 1, "kind": OP_EVENT, "operator_sub": sub, "recorders": recorders} ph = hashlib.sha256(sv.jcs(body).encode()).hexdigest() import base64 sig = base64.b64encode(sk_sign(op_signed_content(sv, id_rec, sub, ph))).decode() entry = {"seq": 1, "prev_hash": "00" * 32, "server_ts": "2026-07-11T00:00:00+00:00", "payload_hash": ph, "actor_sub": sub, "counterparty_sub": None, "actor_sig": sig, "server_beacon": None} entry["entry_hash"] = sv.recompute_entry_hash(entry) proof = [] oc = {"commitment": body, "entry": entry, "identity_recorder": id_rec, "signing_pubkey": pk_b64, "inclusion_proof": proof, "checkpoint": {"id": 1, "merkle_root": sv.fold_proof(entry["entry_hash"], proof), "bitcoin": {}}} bindings = {id_rec: {"recorder": id_rec, "signing_pubkey": pk_b64, "verified": True}, real: {"recorder": real, "verified": True}, "rec_shadow": {"recorder": "rec_shadow", "verified": True}} genesis = {real: "GEN_REAL", id_rec: "GEN_ID"} return sub, real, bindings, oc, genesis def _selftest(): """Offline soundness: exact signing bytes, membership logic, AND — with PyNaCl — the adversarial battery that witnesses every RED verdict (a valid commitment → BOUND, then each tamper → refusal).""" sv = _load_sibling("standing-verify.py") ok = [] sc = op_signed_content(sv, "rec_x", "op-sub", "ph123").decode() ok.append(("op signed_content shape", sc == '{"actor_sub":"op-sub","client_ts":null,"counterparty_sub":null,' '"event_type":"touchstone.operator_recorders","payload_hash":"ph123","recorder_id":"rec_x","v":1}')) b1 = {"v": 1, "kind": OP_EVENT, "operator_sub": "s", "recorders": [{"public_id": "r", "genesis_hash": "g"}]} b2 = {"recorders": [{"genesis_hash": "g", "public_id": "r"}], "operator_sub": "s", "kind": OP_EVENT, "v": 1} _h = lambda b: hashlib.sha256(sv.jcs(b).encode()).hexdigest() ok.append(("payload_hash JCS-stable", _h(b1) == _h(b2))) # ── THE CLOCK MUST NAME ITSELF (verifier-side inflation audit) ──────────────────────────────── # clock_note() is PURE and dep-free, so these run WITHOUT PyNaCl — deliberately OUTSIDE the battery # below. A dep-only-available check is one more "runs only in the lucky environment" hole, which is # exactly the class mutation-guard.py exists to catch; keep the beacon-naming assertion unconditional. ok.append(("no beacon → clock named OPERATOR-WRITTEN / BACKDATED", "OPERATOR-WRITTEN" in clock_note({}) and "BACKDATED" in clock_note({}))) _b = {"server_beacon": {"chain": "drand:quicknet", "round": 1, "randomness": "aa"}} ok.append(("beacon present → reported as a real INTERVAL", "BOUND on both sides" in clock_note(_b) and "OPERATOR-WRITTEN" not in clock_note(_b))) # Adversarial battery (needs PyNaCl to MINT a valid baseline signature). Each case asserts the # verdict a relier must see. Consistent-gossip and mismatch/absent are resolver/gossip-injected. try: from nacl.signing import SigningKey import base64 sk = SigningKey.generate() pk_b64 = base64.b64encode(bytes(sk.verify_key)).decode() sub, real, bindings, oc, genesis = _mint_valid(sv, lambda m: sk.sign(m).signature, pk_b64) G = lambda r: genesis.get(r) CONS = lambda idr: ("consistent", "") FORK = lambda idr: ("fork", "two heads at seq 1") def verdict(rec=real, b=bindings, o=oc, g=G, go=CONS): return evaluate(sv, sub, rec, b, o, g, go)[1] ok.append(("valid → BOUND", verdict() == "BOUND")) # forged signature: flip a byte of actor_sig o_sig = copy.deepcopy(oc) s = o_sig["entry"]["actor_sig"] o_sig["entry"]["actor_sig"] = ("B" if s[0] != "B" else "C") + s[1:] ok.append(("forged actor_sig → INVALID", verdict(o=o_sig) == "INVALID_COMMITMENT")) # tampered body: the set no longer hashes to the signed payload_hash o_body = copy.deepcopy(oc) o_body["commitment"]["recorders"].append({"public_id": "rec_injected", "genesis_hash": "x"}) ok.append(("tampered body → INVALID", verdict(o=o_body) == "INVALID_COMMITMENT")) # tampered entry_hash o_eh = copy.deepcopy(oc) o_eh["entry"]["entry_hash"] = "00" * 32 ok.append(("tampered entry_hash → INVALID", verdict(o=o_eh) == "INVALID_COMMITMENT")) # identity binding not self-operated (operator-claimed, not key-proven) b_unv = copy.deepcopy(bindings) b_unv["rec_identity"]["verified"] = False ok.append(("unverified identity → INVALID", verdict(b=b_unv) == "INVALID_COMMITMENT")) # a recorder absent from the set (the shadow) — firm negative against an ANCHORED commitment ok.append(("absent recorder → UNREGISTERED", verdict(rec="rec_shadow") == "UNREGISTERED")) # absent, but the commitment isn't anchored yet → PENDING (limbo), not a firm shadow verdict. o_unanch = copy.deepcopy(oc) o_unanch["checkpoint"] = None o_unanch["inclusion_proof"] = None ok.append(("absent + unanchored → PENDING", verdict(rec="rec_shadow", o=o_unanch) == "UNREGISTERED_PENDING")) # committed genesis ≠ the recorder's live seq-0 (a swapped chain under the same public_id) ok.append(("swapped genesis → GENESIS_MISMATCH", verdict(g=lambda r: "DIFFERENT" if r == real else genesis.get(r)) == "GENESIS_MISMATCH")) # a second conflicting head under the identity (gossip fork) ok.append(("gossip fork → EQUIVOCATION", verdict(go=FORK) == "EQUIVOCATION")) # no commitment at all ok.append(("no commitment → UNCOMMITTED", evaluate(sv, sub, real, bindings, None, G, CONS)[1] == "UNCOMMITTED")) except ImportError: print(" (adversarial battery skipped — pip install pynacl to mint a valid baseline signature)") for n, g in ok: print(" %-34s %s" % (n, "ok" if g else "FAIL")) bad = [n for n, g in ok if not g] print("\n" + ("SELFTEST FAILED: " + ", ".join(bad) if bad else "SELFTEST OK — signing bytes, membership, and every RED verdict witnessed.")) return 1 if bad else 0 if __name__ == "__main__": sys.exit(main(sys.argv))