diff --git a/cmd_chat/sor/analysis/__init__.py b/cmd_chat/sor/analysis/__init__.py new file mode 100644 index 0000000..f740aaf --- /dev/null +++ b/cmd_chat/sor/analysis/__init__.py @@ -0,0 +1,18 @@ +"""R7 (partial) — offline analysis calibration primitives. + +This package holds only the *offline-validatable* half of R7: the detectors that +the instrument-validation gate calibrates on **synthetic fixtures** (never on +confirmatory-cell data — CLAUDE.md build discipline). Specifically: + +- ``detectors.shannon_entropy_bits`` — the RQ2 anonymity-set entropy metric; gate + item 4 requires it to return ``log2(N)`` for ``N`` equiprobable senders. +- ``detectors.bridge_correlation_auc`` — the RQ1 bridge-linkability scorer; gate + item 3 requires AUC ≈ 1 on a known-linked control pair and ≈ 0.5 on a + known-unlinked pair. + +These are pure functions over in-memory series/distributions — they read no +pcaps, spawn no engine, move no traffic, and touch no VM fabric. The +traffic-moving R7 pieces (``churn.py`` seeded VM spin/kill; the live selector +rebuild loop; the confirmatory battery that writes ``metrics.json``) are NOT +here — they are HELD pending R4/R6 + a live grid (see OVERSEER-STATUS.md). +""" diff --git a/cmd_chat/sor/analysis/detectors.py b/cmd_chat/sor/analysis/detectors.py new file mode 100644 index 0000000..07d5691 --- /dev/null +++ b/cmd_chat/sor/analysis/detectors.py @@ -0,0 +1,132 @@ +"""R7 (partial) — offline detector primitives + seeded synthetic fixtures. + +Pure, stdlib-only calibration instruments (gate items 3 and 4). No I/O, no +engine, no pcaps: they operate on in-memory numeric series/distributions so they +can be validated entirely offline against synthetic ground truth. Determinism is +inherited from the R1 ``SorRng`` so a calibration fixture is reproducible from +its seed alone. +""" + +from __future__ import annotations + +import math +from typing import Dict, List, Sequence, Union + +from cmd_chat.sor.config import Domain, SorRng + +Number = Union[int, float] +Series = Sequence[Number] + + +# --------------------------------------------------------------------------- # +# RQ2 — anonymity-set entropy (gate item 4). +# --------------------------------------------------------------------------- # +def shannon_entropy_bits(weights: Union[Dict[object, Number], Sequence[Number]]) -> float: + """Shannon entropy in **bits** of an observed sender distribution. + + ``weights`` is either a mapping ``sender -> count`` or a sequence of + non-negative weights. For ``N`` equiprobable senders this returns exactly + ``log2(N)`` (the maximum for an N-set), which is the gate item 4 predicate. + An empty or all-zero distribution has entropy 0.""" + vals = list(weights.values()) if isinstance(weights, dict) else list(weights) + total = float(sum(vals)) + if total <= 0: + return 0.0 + h = 0.0 + for c in vals: + if c > 0: + p = c / total + h -= p * math.log2(p) + return h + + +# --------------------------------------------------------------------------- # +# RQ1 — bridge-linkability correlation scorer (gate item 3). +# --------------------------------------------------------------------------- # +def pearson(xs: Series, ys: Series) -> float: + """Pearson correlation coefficient of two equal-length numeric series. + Returns 0.0 for empty, mismatched-length, or zero-variance inputs (a flat + series carries no linkage signal).""" + n = len(xs) + if n == 0 or n != len(ys): + return 0.0 + mx = sum(xs) / n + my = sum(ys) / n + num = sum((x - mx) * (y - my) for x, y in zip(xs, ys)) + dx = math.sqrt(sum((x - mx) ** 2 for x in xs)) + dy = math.sqrt(sum((y - my) ** 2 for y in ys)) + if dx == 0.0 or dy == 0.0: + return 0.0 + return num / (dx * dy) + + +def score_matrix(ingress: Sequence[Series], egress: Sequence[Series]) -> List[List[float]]: + """Full pairwise linkage-score matrix ``S[i][j] = pearson(ingress[i], + egress[j])``. The candidate true pairing is the diagonal (``i == j``).""" + return [[pearson(a, b) for b in egress] for a in ingress] + + +def auc(pos: Series, neg: Series) -> float: + """Rank AUC = P(pos ranked above neg) with ties counted as 0.5 (Mann-Whitney + U normalized). AUC 1.0 = perfect separation, 0.5 = chance. Empty side -> 0.5.""" + if not pos or not neg: + return 0.5 + wins = 0.0 + for p in pos: + for q in neg: + if p > q: + wins += 1.0 + elif p == q: + wins += 0.5 + return wins / (len(pos) * len(neg)) + + +def linkage_auc(scores: Sequence[Sequence[float]]) -> float: + """AUC of the diagonal (linked) scores against the off-diagonal (unlinked) + scores of a square score matrix — the RQ1 bridge-correlation AUC.""" + n = len(scores) + pos = [scores[i][i] for i in range(n)] + neg = [scores[i][j] for i in range(n) for j in range(n) if i != j] + return auc(pos, neg) + + +def bridge_correlation_auc(ingress: Sequence[Series], egress: Sequence[Series]) -> float: + """RQ1 scorer: how well the correlator links each ingress flow to its true + egress flow, as AUC over all candidate pairings. ≈1 for a linked bridge, + ≈0.5 for an unlinked one (gate item 3).""" + return linkage_auc(score_matrix(ingress, egress)) + + +# --------------------------------------------------------------------------- # +# Seeded synthetic calibration fixtures (ground truth known by construction). +# --------------------------------------------------------------------------- # +def synthetic_bridge_fixture( + seed: int, + n_flows: int = 8, + bins: int = 64, + jitter: int = 5, + linked: bool = True, +) -> tuple[List[List[int]], List[List[int]]]: + """Deterministically build ``(ingress, egress)`` flow sets of per-bin byte + counts from ``seed`` alone (reuses the R1 ``SorRng`` streams). + + - ``linked=True``: each egress flow is its ingress flow plus small independent + padding/latency ``jitter`` — the *known-linked* control (diagonal + correlation dominates -> AUC ≈ 1). + - ``linked=False``: egress flows are drawn from an independent domain stream, + uncorrelated with ingress — the *known-unlinked* control (no diagonal + advantage -> AUC ≈ 0.5). + """ + rng = SorRng(seed) + sig = rng.stream(Domain.PATH) # ingress signal + ingress = [[sig.next_below(1000) for _ in range(bins)] for _ in range(n_flows)] + if linked: + noise = rng.stream(Domain.PADDING) + span = 2 * jitter + 1 + egress = [ + [v + (noise.next_below(span) - jitter) for v in flow] for flow in ingress + ] + else: + alt = rng.stream(Domain.CHURN) # independent of the PATH signal + egress = [[alt.next_below(1000) for _ in range(bins)] for _ in range(n_flows)] + return ingress, egress diff --git a/tests/test_sor_analysis.py b/tests/test_sor_analysis.py new file mode 100644 index 0000000..9b89a52 --- /dev/null +++ b/tests/test_sor_analysis.py @@ -0,0 +1,112 @@ +"""R7 (partial) — offline detector calibration checks (gate items 3 & 4). + +Acceptance predicate (roadmap R7, the offline-verifiable half): + - entropy metric returns log2(N) for N equiprobable senders (gate item 4); + - the bridge-correlation scorer gives AUC ≈ 1 on a known-linked control pair + and ≈ 0.5 on a known-unlinked pair (gate item 3). + +Calibrated on synthetic fixtures only (CLAUDE.md: detectors are never fit to +confirmatory-cell data). These detectors read no pcaps, spawn no engine, and +move no traffic — fully offline. The traffic-moving R7 pieces (churn VM +spin/kill, live selector rebuild, metrics.json emission) are HELD for R4/R6 + a +live grid and are intentionally absent. +""" + +import math +import statistics + +import pytest + +from cmd_chat.sor.analysis.detectors import ( + auc, + bridge_correlation_auc, + pearson, + shannon_entropy_bits, + synthetic_bridge_fixture, +) + + +# --------------------------------------------------------------------------- # +# Gate item 4 — anonymity-set entropy. +# --------------------------------------------------------------------------- # +@pytest.mark.parametrize("n", [1, 2, 3, 4, 5, 8, 16, 17]) +def test_entropy_equiprobable_is_log2n(n): + # N equiprobable senders -> exactly log2(N) bits (the gate item 4 predicate). + h = shannon_entropy_bits({f"sender{i}": 1 for i in range(n)}) + assert h == pytest.approx(math.log2(n), abs=1e-12) + + +def test_entropy_is_maximized_by_the_uniform_distribution(): + n = 8 + uniform = shannon_entropy_bits([1] * n) + skewed = shannon_entropy_bits([10, 1, 1, 1, 1, 1, 1, 1]) + assert skewed < uniform == pytest.approx(3.0) + + +def test_entropy_degenerate_cases(): + assert shannon_entropy_bits({"only": 5}) == 0.0 # one sender -> 0 bits + assert shannon_entropy_bits([]) == 0.0 + assert shannon_entropy_bits([0, 0]) == 0.0 + + +def test_entropy_known_skewed_value(): + # 3:1 split -> H = 0.75*log2(4/3) + 0.25*log2(4) = 0.811278... + assert shannon_entropy_bits([3, 1]) == pytest.approx(0.8112781, abs=1e-6) + + +# --------------------------------------------------------------------------- # +# Correlation + AUC primitives. +# --------------------------------------------------------------------------- # +def test_pearson_identities(): + assert pearson([1, 2, 3, 4], [1, 2, 3, 4]) == pytest.approx(1.0) + assert pearson([1, 2, 3, 4], [4, 3, 2, 1]) == pytest.approx(-1.0) + assert pearson([1, 2, 3], [5, 5, 5]) == 0.0 # zero variance -> no signal + assert pearson([], []) == 0.0 + assert pearson([1, 2], [1, 2, 3]) == 0.0 # length mismatch + + +def test_auc_basics(): + assert auc([3, 4, 5], [0, 1, 2]) == 1.0 # perfectly separated + assert auc([0, 1], [2, 3]) == 0.0 # perfectly inverted + assert auc([1, 1], [1, 1]) == 0.5 # all ties -> chance + assert auc([], [1]) == 0.5 + + +# --------------------------------------------------------------------------- # +# Gate item 3 — bridge-correlation AUC calibration. +# --------------------------------------------------------------------------- # +def test_known_linked_pair_scores_auc_near_one(): + # A linked bridge (egress = ingress + small jitter): the diagonal dominates. + ingress, egress = synthetic_bridge_fixture(42, linked=True) + assert bridge_correlation_auc(ingress, egress) == pytest.approx(1.0) + # Robust across seeds — never below near-perfect. + for seed in (1, 123456789, 777, 2024): + ig, eg = synthetic_bridge_fixture(seed, linked=True) + assert bridge_correlation_auc(ig, eg) >= 0.99 + + +def test_known_unlinked_pair_scores_auc_near_half(): + # A single unlinked pair is noisy (finite flows); the estimator is *unbiased*, + # so the ensemble mean over many seeds sits at chance (0.5). + aucs = [ + bridge_correlation_auc(*synthetic_bridge_fixture(seed, n_flows=24, linked=False)) + for seed in range(40) + ] + assert statistics.mean(aucs) == pytest.approx(0.5, abs=0.05) + # A representative single pair is within a chance band (not linkable). + ig, eg = synthetic_bridge_fixture(42, n_flows=24, linked=False) + assert 0.35 <= bridge_correlation_auc(ig, eg) <= 0.65 + + +def test_linked_separates_from_unlinked(): + linked = bridge_correlation_auc(*synthetic_bridge_fixture(7, n_flows=24, linked=True)) + unlinked = bridge_correlation_auc(*synthetic_bridge_fixture(7, n_flows=24, linked=False)) + assert linked > unlinked + + +def test_fixture_is_seed_deterministic(): + a = synthetic_bridge_fixture(99, linked=True) + b = synthetic_bridge_fixture(99, linked=True) + assert a == b + # Different ground truth from the same seed is actually different data. + assert synthetic_bridge_fixture(99, linked=True) != synthetic_bridge_fixture(99, linked=False)