"""The validity layer: when an origin attribution is invalid, or means less than it appears to, say so — and say which. Calibration answers "how often is a 0.8 right?". This module answers prior questions: can an attribution be about this transaction at all, and if so, what may it claim? docs/VALIDITY.md is the taxonomy and the pre-registered metric revision. Every reason that fires is reported; the verdict's tier is the most severe among them: ABSTAIN DEGENERATE candidate_count 0 or 1: nothing to rank. NOT_REACHABLE every candidate is a known public relay (the matrix's `scope_out`): the zero-ceiling case. QUALIFIED COINJOIN the answer is the broadcasting peer only; the inputs' ownership is not attributable. TOR_ONION the answer is an onion identity, never an IP. ANNOTATE DANDELION_STEM the first announcement stands alone before a burst, the shape a stem leaves. Flag only: Dandelion is not deployed in Bitcoin Core. V2_PASSIVE_TAP a pcap capture held port-8333 flows it could not decode (BIP-324 v2): v2 peers may be missing from the candidates. Never on debug.log/.btcap, which the node — a session endpoint — writes. Only ABSTAIN withholds an answer, and it does so through the existing rule — `low_confidence_origin` (engines.propagation) and `eval.origin.flagged_at` — not a second path. `validity.enforce: false` switches the layer off there; `validity.mode: binary` restores P6's gate, where every non-PASS verdict abstained, so that policy can still be scored. """ from __future__ import annotations import statistics from dataclasses import dataclass import pandas as pd import config from graph.builder import Tx from graph.clustering import equal_value_group, is_coinjoin PASS = "PASS" DEGENERATE = "DEGENERATE" NOT_REACHABLE = "NOT_REACHABLE" COINJOIN = "COINJOIN" TOR_ONION = "TOR_ONION" DANDELION_STEM = "DANDELION_STEM" V2_PASSIVE_TAP = "V2_PASSIVE_TAP" #: Most severe first; `assess` reports them in this order. REASONS = (DEGENERATE, NOT_REACHABLE, COINJOIN, TOR_ONION, DANDELION_STEM, V2_PASSIVE_TAP) ABSTAIN, QUALIFIED, ANNOTATE = "ABSTAIN", "QUALIFIED", "ANNOTATE" TIER = {DEGENERATE: ABSTAIN, NOT_REACHABLE: ABSTAIN, COINJOIN: QUALIFIED, TOR_ONION: QUALIFIED, DANDELION_STEM: ANNOTATE, V2_PASSIVE_TAP: ANNOTATE} #: The columns a frame of origins carries. COLUMNS = ["validity", "validity_tier", "validity_reasons", "validity_confidence", "validity_evidence"] @dataclass(frozen=True) class Verdict: reason: str = PASS # the most severe reason that fired confidence: float | None = None # that it is present; None on PASS evidence: tuple[str, ...] = () reasons: tuple[str, ...] = () # every reason that fired, in REASONS order @property def passed(self) -> bool: return self.reason == PASS @property def tier(self) -> str: return TIER.get(self.reason, PASS if self.passed else self.reason) def as_dict(self) -> dict: return {"tier": self.tier, "reason": None if self.passed else self.reason, "reasons": list(self.reasons), "confidence": self.confidence, "evidence": list(self.evidence)} def columns(self) -> dict: return {"validity": self.reason, "validity_tier": self.tier, "validity_reasons": list(self.reasons), "validity_confidence": self.confidence, "validity_evidence": list(self.evidence)} VALID = Verdict(PASS, None, ("no invalidating condition detected",)) #: Not a detector: what an output stored before this layer existed carries, so #: that it is never served without a verdict and never passed off as PASS. NOT_ASSESSED = Verdict("NOT_ASSESSED", None, ( "produced before the validity layer existed; rerun the pipeline to assess it",), ("NOT_ASSESSED",)) def enforced(cfg: dict) -> bool: return bool(cfg.get("validity", {}).get("enforce", True)) def tiered(cfg: dict) -> bool: """Qualifications are in force: the layer is on and not in P6's binary mode.""" return enforced(cfg) and cfg.get("validity", {}).get("mode", "tiered") == "tiered" def withholds(tier: str, cfg: dict) -> bool: """Whether a verdict of this tier abstains under the configured policy.""" if not enforced(cfg): return False return tier == ABSTAIN if tiered(cfg) else tier != PASS def answer(peer: str | None, verdict: Verdict, cfg: dict) -> dict | None: """What an origin answer may claim. A QUALIFIED answer is never shaped like an IP attribution: an onion identity has no `ip`, and a CoinJoin broadcaster carries an explicit `input_ownership: not attributable`.""" if peer is None or withholds(verdict.tier, cfg): return None qualified = tiered(cfg) if qualified and TOR_ONION in verdict.reasons: out = {"kind": "onion_identity", "onion": peer, "actionable": "not for IP-level follow-up"} elif qualified and COINJOIN in verdict.reasons: out = {"kind": "broadcasting_peer", "ip": peer} else: return {"kind": "ip_attribution", "ip": peer} if COINJOIN in verdict.reasons: out["input_ownership"] = "not attributable: a CoinJoin's inputs have many owners" return out # --- the detectors ---------------------------------------------------------- def degenerate(candidate_count: int) -> Verdict | None: if candidate_count > 1: return None return Verdict(DEGENERATE, 1.0, ( f"{candidate_count} peer{'s' if candidate_count != 1 else ''} announced this " "transaction to the observer, so there is nothing to compare",)) def not_reachable(scope_out: bool) -> Verdict | None: if not scope_out: return None return Verdict(NOT_REACHABLE, 1.0, ( "every peer that announced it is a known public relay; the sender is not " "among the candidates this observer can see",)) def coinjoin(tx: Tx | None, cfg: dict) -> Verdict | None: """`graph.clustering.is_coinjoin` — the check clustering already uses to skip common-input-ownership, so a mix is one thing everywhere.""" if tx is None or not is_coinjoin(tx, cfg): return None value, group = equal_value_group(tx.output_values, cfg["graph"]["coinjoin"]["equal_value_tolerance"]) target = cfg["engines"]["rules"]["coinjoin"]["target_equal_outputs"] return Verdict(COINJOIN, round(min(1.0, len(group) / target), 3), ( f"{len(group)} of {len(tx.outputs)} outputs pay {value:.8f} BTC, from " f"{len(tx.inputs)} inputs", "a CoinJoin has one broadcaster and many owners: the answer names who " "broadcast it, never who owns its inputs")) def tor_onion(peer: str | None) -> Verdict | None: if not (peer and str(peer).endswith(".onion")): return None return Verdict(TOR_ONION, 1.0, ( f"{peer} is a Tor hidden service: the answer is that onion identity, which " "names no IP and supports no IP-level follow-up",)) def v2_passive_tap(capture_source: str | None, unreadable_flows) -> Verdict | None: """Only a packet capture is blind to BIP-324: bitcoind's own log, and a .btcap its collector writes, come from the node, which decrypts.""" if not (capture_source and str(capture_source).startswith("pcap:")): return None if not unreadable_flows or pd.isna(unreadable_flows) or int(unreadable_flows) <= 0: return None n = int(unreadable_flows) return Verdict(V2_PASSIVE_TAP, 1.0, ( f"this packet capture held {n} port-8333 flow{'s' if n != 1 else ''} it could not " "decode, consistent with BIP-324 v2 encryption: peers on those links are missing " "from the candidates",)) def dandelion_stem(times, cfg: dict) -> Verdict | None: """First-announcement isolation. A stem hands the transaction to one peer at a time; an observer on or beside the stem hears one announcement, then nothing, then the broadcast arrives as a burst. Ordinary diffusion reaches an observer's neighbours at a steadier pace. The statistic is the gap after the first announcement over the median gap among the rest. Its threshold is fixed in config.yaml, before any result. """ d = cfg["validity"]["dandelion"] t = sorted(float(x) for x in times) if len(t) < d["min_candidates"]: return None gaps = [b - a for a, b in zip(t, t[1:])] later = statistics.median(gaps[1:]) if gaps[0] <= 0: return None ratio = gaps[0] / later if later > 0 else float("inf") if ratio < d["isolation_ratio"]: return None return Verdict(DANDELION_STEM, round(1.0 - 0.5 * d["isolation_ratio"] / ratio, 3), ( f"the first announcement stood alone for {gaps[0] * 1000:.0f} ms, " f"{ratio:.1f}x the median gap between the {len(t) - 1} that followed", "consistent with a Dandelion stem, where the first announcer may be a relay; " "a flag only — Bitcoin Core has no stem phase")) # --- one origin -------------------------------------------------------------- def combine(fired: list[Verdict]) -> Verdict: """Every verdict that fired -> one: the most severe leads, all are listed.""" fired = sorted(fired, key=lambda v: REASONS.index(v.reason)) if not fired: return VALID return Verdict(fired[0].reason, fired[0].confidence, tuple(line for v in fired for line in v.evidence), tuple(v.reason for v in fired)) def assess(candidate_count: int, scope_out: bool = False, peer: str | None = None, times=(), tx: Tx | None = None, cfg: dict | None = None, capture_source: str | None = None, unreadable_flows=None) -> Verdict: cfg = cfg or config.load() fired = [v for v in (degenerate(candidate_count), not_reachable(scope_out), coinjoin(tx, cfg), tor_onion(peer), dandelion_stem(times, cfg), v2_passive_tap(capture_source, unreadable_flows)) if v is not None] return combine(fired) def assess_tree(tree, peer: str | None, intel, cfg: dict | None = None, tx: Tx | None = None, exclude=()) -> Verdict: """For an `engines.propagation` tree. Candidates and announcement times are the tree's senders (`first_sent`), less `exclude` (an observer).""" cfg = cfg or config.load() exclude = set(exclude) announced = tree.first_sent or tree.first_seen candidates = [ip for ip in tree.ips if ip not in exclude] count = 1 if tree.is_single_observation else len(candidates) relay = set(cfg["engines"]["propagation"]["low_confidence_classes"]) scope_out = bool(candidates) and all( intel.classify(ip, tree.graph.nodes.get(ip, {}).get("asn")).ip_class in relay for ip in candidates) return assess(count, scope_out, peer, [ts for ip, ts in announced.items() if ip not in exclude], tx, cfg) def tx_of(txid: str, shape) -> Tx | None: """A transaction's structure from a corpus row (`generator.typologies.shape`).""" if shape is None: return None return Tx(txid, list(zip(shape["in_addrs"], shape["in_vals"])), list(zip(shape["out_addrs"], shape["out_vals"]))) def assess_matrix(matrix: pd.DataFrame, named: pd.DataFrame, shapes: pd.DataFrame | None = None, cfg: dict | None = None) -> pd.DataFrame: """For the `features.relay` matrix: one verdict per (capture_id, txid), about the peer `named` says was chosen (`estimated_origin_ip`).""" cfg = cfg or config.load() key = ["capture_id", "txid"] chosen = named.set_index(key)["estimated_origin_ip"] shape_of = shapes.set_index(key) if shapes is not None and len(shapes) else None rows = [] for (capture_id, txid), group in matrix.groupby(key, sort=False): peer = chosen.get((capture_id, txid)) shape = (shape_of.loc[(capture_id, txid)] if shape_of is not None and (capture_id, txid) in shape_of.index else None) first = group.iloc[0] verdict = assess(int(first["candidate_count"]), bool(first["scope_out"]), peer, group["delta_vs_first_s"].to_numpy(), tx_of(txid, shape), cfg, first.get("capture_source"), first.get("unreadable_flows")) rows.append({"capture_id": capture_id, "txid": txid, **verdict.columns()}) return pd.DataFrame(rows, columns=key + COLUMNS)