Research/v0.4

CHANGELOG.md

@@ -0,0 +1,20 @@
+# Changelog
+
+## 0.4.0
+
+### Added
+- Multipath splitting (`routing.split_strategy`): `iid`, weighted `round_robin`, `batch` and lowest-RTT `latency`, with `routing.merge: common_exit`, per-leg observation (`adversary.observed_legs`), a merged exit-side view (`adversary.egress_observation: merged`) and measured per-leg shares in the metrics.
+- Relay-side mixing at every hop (`mixing.strategy`: `constant`, `exponential`, `pool`) and random-hop cover drop (`cover_behaviour.drop_mode: random_hop`).
+- A bursty traffic generator (`traffic.distribution: burst`, `burst_mean_cells`, `burst_gap_ms`).
+- The `correlation_suite` adversary: training-free attackers A0 (zero-lag), A1 (lag-aware binned correlation), A2 (likelihood ratio with the known delay density) and A3 (timing window), scored on one shared calibration/test split, with confident linkage, TPR at fixed FPR, signal estimate and AUC. Raw observations and score matrices are saved to `correlation_suite.npz`.
+- Closed-form model predictions for the lag-aware attacker, recorded with every suite run and available as `atl predict`.
+- `atl fidelity`, a check on host-induced timing noise, and wave mode (`sessions.max_concurrent`).
+- `atl paired`, paired-seed comparisons with a bootstrap interval and an equivalence margin.
+- `examples/mixnet_pool.yaml`.
+
+### Changed
+- TPR at a fixed FPR is conservative under tied scores (0 instead of 1), and the cut can be chosen on held-out calibration sessions (`adversary.evaluation.calibration_fraction`).
+
+## 0.3.0
+
+Real-process emulator with telescoping circuits, per-hop AEAD, selectable ciphers and curves, WAN link conditions, and the `global_observer`, `path_compromise`, `watermark` and `hop_depth` adversaries.

anontestlab/adversary/__init__.py

@@ -1,11 +1,13 @@
 from .base import Adversary, AdversaryResult, SessionObservation, SimulationContext
+from .correlation_suite import CorrelationSuite
 from .global_observer import GlobalPassiveObserver
 from .hop_depth import HopDepthAdversary
 from .path_compromise import PathCompromiseAdversary
 from .watermark import WatermarkAdversary
 
 ADVERSARIES: dict[str, type[Adversary]] = {
     "global_observer": GlobalPassiveObserver,
+    "correlation_suite": CorrelationSuite,
     "path_compromise": PathCompromiseAdversary,
     "watermark": WatermarkAdversary,
     "hop_depth": HopDepthAdversary,
@@ -25,6 +27,7 @@ def get_adversary(name: str, config) -> Adversary:
     "ADVERSARIES",
     "Adversary",
     "AdversaryResult",
+    "CorrelationSuite",
     "GlobalPassiveObserver",
     "HopDepthAdversary",
     "PathCompromiseAdversary",

anontestlab/adversary/attacks.py

@@ -0,0 +1,139 @@
+"""Score functions for the training-free correlation attackers. Each takes
+the per-session ingress (entry-side) and egress (exit-side) timestamp lists
+and returns an n x n score matrix: row i is ingress session i, column j is
+egress session j, and the diagonal holds the true pairs.
+
+- lag_correlation_scores: binned-count Pearson correlation, maximised over
+  non-negative lags (A1). With max_lag_bins=0 it is the zero-lag correlator
+  (A0).
+- likelihood_ratio_scores: point-process log-likelihood ratio of the egress
+  times given the ingress times and a known delay density (A2), after
+  Danezis, "The Traffic Analysis of Continuous-Time Mixes", PET 2004.
+- window_scores: share of ingress packets with an egress packet inside a
+  delay window, minus the share expected by chance (A3).
+"""
+from __future__ import annotations
+
+import numpy as np
+
+
+def bin_matrix(times_per_session: list[list[float]], bin_s: float, horizon_s: float) -> np.ndarray:
+    n_bins = max(1, int(np.ceil(horizon_s / bin_s)))
+    out = np.zeros((len(times_per_session), n_bins))
+    for row, times in enumerate(times_per_session):
+        if not times:
+            continue
+        idx = (np.asarray(times) // bin_s).astype(int)
+        idx = idx[(idx >= 0) & (idx < n_bins)]
+        np.add.at(out[row], idx, 1.0)
+    return out
+
+
+def _standardise_rows(m: np.ndarray) -> np.ndarray:
+    centred = m - m.mean(axis=1, keepdims=True)
+    norm = np.linalg.norm(centred, axis=1, keepdims=True)
+    norm[norm == 0] = 1.0
+    return centred / norm
+
+
+def lag_correlation_scores(
+    ingress: list[list[float]],
+    egress: list[list[float]],
+    bin_s: float,
+    horizon_s: float,
+    max_lag_bins: int,
+    ingress_horizon_s: float | None = None,
+) -> np.ndarray:
+    """Best standardised correlation over non-negative lags.
+
+    Without ingress_horizon_s both sides are binned over horizon_s and the
+    lagged windows shrink as the lag grows. With it, ingress is binned over
+    its own (shorter) window and every lag compares it against an egress
+    window of the same length, so ingress never carries the empty stretch
+    that egress needs for delayed packets. Empty bins shared by all sessions
+    would otherwise make impostor pairs correlate.
+    """
+    y = bin_matrix(egress, bin_s, horizon_s)
+    best = np.full((len(ingress), y.shape[0]), -np.inf)
+    if ingress_horizon_s is None:
+        x = bin_matrix(ingress, bin_s, horizon_s)
+        n_bins = x.shape[1]
+        for lag in range(0, min(max_lag_bins, n_bins - 2) + 1):
+            xs = _standardise_rows(x[:, : n_bins - lag])
+            ys = _standardise_rows(y[:, lag:])
+            np.maximum(best, xs @ ys.T, out=best)
+        return best
+    x = bin_matrix(ingress, bin_s, ingress_horizon_s)
+    n_x = x.shape[1]
+    xs = _standardise_rows(x)
+    for lag in range(0, min(max_lag_bins, y.shape[1] - n_x) + 1):
+        ys = _standardise_rows(y[:, lag : lag + n_x])
+        np.maximum(best, xs @ ys.T, out=best)
+    return best
+
+
+def likelihood_ratio_scores(
+    ingress: list[list[float]],
+    egress: list[list[float]],
+    density: np.ndarray,
+    step_s: float,
+    horizon_s: float,
+    floor_rate: float = 1e-3,
+) -> np.ndarray:
+    """Score(i, j) = sum over egress packets y of j of
+    log((c_ij + (f * X_i)(y)) / r_j), where f * X_i is ingress i's packet
+    train convolved with the delay density (packets per second), r_j is
+    egress j's mean rate, and c_ij = max(r_j - rate of X_i, floor_rate) is
+    the part of j's rate left unexplained by i. Under this model the
+    compensator terms of the two hypotheses cancel."""
+    n_bins = max(1, int(np.ceil(horizon_s / step_s)))
+    x = bin_matrix(ingress, step_s, horizon_s)
+    size = n_bins + len(density)
+    fft_len = 1 << (size - 1).bit_length()
+    kernel = np.fft.rfft(density, fft_len)
+    intensity = np.fft.irfft(np.fft.rfft(x, fft_len, axis=1) * kernel, fft_len, axis=1)[:, :n_bins]
+    np.maximum(intensity, 0.0, out=intensity)
+    x_rate = x.sum(axis=1) / horizon_s
+    scores = np.zeros((len(ingress), len(egress)))
+    for j, times in enumerate(egress):
+        if not times:
+            continue
+        idx = (np.asarray(times) // step_s).astype(int)
+        idx = idx[(idx >= 0) & (idx < n_bins)]
+        r_j = max(len(idx) / horizon_s, floor_rate)
+        c = np.maximum(r_j - x_rate, floor_rate)[:, None]
+        scores[:, j] = np.log((c + intensity[:, idx]) / r_j).sum(axis=1)
+    return scores
+
+
+def window_scores(
+    ingress: list[list[float]],
+    egress: list[list[float]],
+    window_s: float,
+    horizon_s: float,
+    grid_s: float = 0.001,
+) -> np.ndarray:
+    """Hit test on a grid of grid_s: an ingress packet at x counts as matched
+    if egress j has a packet in the grid cells from x to x + window_s. The
+    grid error is at most one cell at each end of the window."""
+    n_in = len(ingress)
+    lengths = np.array([len(x) for x in ingress])
+    owner = np.repeat(np.arange(n_in), lengths)
+    flat = np.concatenate([np.asarray(x, dtype=float) for x in ingress]) if lengths.sum() else np.zeros(0)
+    n_cells = int(np.ceil((horizon_s + window_s) / grid_s)) + 2
+    lo = np.clip((flat / grid_s).astype(np.int64), 0, n_cells - 1)
+    hi = np.clip(((flat + window_s) / grid_s).astype(np.int64) + 1, 0, n_cells - 1)
+    safe_len = np.maximum(lengths, 1)
+    scores = np.zeros((n_in, len(egress)))
+    for j, times in enumerate(egress):
+        chance = 1.0 - np.exp(-len(times) / horizon_s * window_s)
+        if len(times) == 0 or len(flat) == 0:
+            scores[:, j] = -chance
+            continue
+        cells = np.clip((np.asarray(times, dtype=float) / grid_s).astype(np.int64), 0, n_cells - 1)
+        cumulative = np.concatenate(([0], np.cumsum(np.bincount(cells, minlength=n_cells))))
+        hit = (cumulative[hi] - cumulative[lo]) > 0
+        share = np.bincount(owner, weights=hit, minlength=n_in) / safe_len
+        share[lengths == 0] = 0.0
+        scores[:, j] = share - chance
+    return scores

anontestlab/adversary/base.py

@@ -18,6 +18,7 @@ class SessionObservation:
     egress_seq: list[int] = field(default_factory=list)  # real_seq for each egress_times entry,
     # same index alignment; lets a position-sensitive adversary (watermark) key off true
     # send order instead of arrival order, which packet loss can desynchronize
+    leg_real_counts: list[int] = field(default_factory=list)  # real cells sent on each leg (path index)
 
 
 @dataclass
@@ -37,6 +38,7 @@ class AdversaryResult:
     name: str
     n_sessions: int
     metrics: dict[str, float]
+    artifacts: dict[str, object] = field(default_factory=dict)  # arrays saved next to the metrics (e.g. score matrices)
 
 
 class Adversary(ABC):

anontestlab/adversary/correlation_suite.py

@@ -0,0 +1,132 @@
+"""Runs the training-free correlation attackers side by side on the same
+observations and reports, for each one, confident linkage, TPR at fixed
+FPR and the signal estimate. All attackers share one calibration/test split
+of the sessions, so their results are paired.
+
+Attacker ids:
+  a0_w<bin>          zero-lag binned-count correlation
+  a1_w<bin>          lag-aware binned-count correlation (non-negative lags)
+  a2                 likelihood-ratio attacker with the known delay density
+  a3_q<quantile>     timing-window matcher, window = delay quantile + floor
+
+The score matrices and the split are returned as artifacts so later
+analysis (strongest-attacker selection, other threshold rules) does not
+need a rerun.
+"""
+from __future__ import annotations
+
+import math
+import random
+
+import numpy as np
+
+from ..metrics.stats import confident_linkage, roc_auc, signal_estimate, tpr_at_fpr_detail
+from . import attacks, delay_model
+from .base import Adversary, AdversaryResult, SimulationContext
+
+
+def _off_diagonal(m: np.ndarray) -> np.ndarray:
+    return m[~np.eye(m.shape[0], dtype=bool)]
+
+
+class CorrelationSuite(Adversary):
+    name = "correlation_suite"
+
+    def __init__(self, config):
+        self.config = config
+        self.bin_widths_s = list(config.suite_bin_widths_s)
+        self.attackers = set(config.suite_attackers)
+        self.psi = list(config.suite_psi)
+        self.base_rates = list(config.suite_base_rates)
+        self.fpr_targets = list(config.suite_fpr_targets)
+        self.calibration_fraction = config.observer_calibration_fraction or 0.5
+        self.window_quantiles = list(config.suite_window_quantiles)
+        self.latency_floor_s = config.suite_latency_floor_ms / 1000.0
+        self.lr_step_s = config.suite_lr_step_ms / 1000.0
+        if config.suite_lag_max_s is not None:
+            self.lag_max_s = config.suite_lag_max_s
+        else:
+            p99 = delay_model.delay_quantiles(config, (0.99,))[0.99]
+            self.lag_max_s = max(2.0, p99) + 0.5
+
+    @classmethod
+    def from_config(cls, config) -> CorrelationSuite:
+        return cls(config)
+
+    def _scores(self, ingress, egress, horizon) -> dict[str, np.ndarray]:
+        out = {}
+        flow = self.config.duration_s
+        for w in self.bin_widths_s:
+            if "a0" in self.attackers:
+                out[f"a0_w{w}"] = attacks.lag_correlation_scores(ingress, egress, w, horizon, 0, flow)
+            if "a1" in self.attackers:
+                lags = math.ceil(self.lag_max_s / w)
+                out[f"a1_w{w}"] = attacks.lag_correlation_scores(ingress, egress, w, horizon, lags, flow)
+        if "a2" in self.attackers:
+            density = delay_model.delay_density(self.config, self.lr_step_s)
+            out["a2"] = attacks.likelihood_ratio_scores(ingress, egress, density, self.lr_step_s, horizon)
+        if "a3" in self.attackers:
+            quantiles = delay_model.delay_quantiles(self.config, tuple(self.window_quantiles))
+            for q, value in quantiles.items():
+                out[f"a3_q{q}"] = attacks.window_scores(ingress, egress, value + self.latency_floor_s, horizon)
+        return out
+
+    def attack(self, ctx: SimulationContext, rng: random.Random) -> AdversaryResult:
+        sids = sorted(ctx.sessions)
+        n = len(sids)
+        if n < 4:
+            return AdversaryResult(self.name, n, {"suite_sessions": n})
+        # Every session is binned over the same configured window: the flow
+        # duration plus the lag range, since exit traffic trails by the added
+        # delay. A shared horizon taken from the latest timestamp would let one
+        # overrunning session pad every other series with trailing zeros, and
+        # that common on/off shape makes impostor pairs correlate. The binned
+        # attackers use only the flow duration on the ingress side for the
+        # same reason (see attacks.lag_correlation_scores).
+        horizon = self.config.duration_s + self.lag_max_s
+        raw_in = [ctx.sessions[s].ingress_times for s in sids]
+        raw_out = [ctx.sessions[s].egress_times for s in sids]
+        ingress = [[t for t in obs if 0.0 <= t < horizon] for obs in raw_in]
+        egress = [[t for t in obs if 0.0 <= t < horizon] for obs in raw_out]
+        dropped = sum(len(a) for a in raw_in + raw_out) - sum(len(a) for a in ingress + egress)
+        overrun = sum(
+            1 for a, b in zip(raw_in, raw_out) if any(t >= horizon for t in a) or any(t >= horizon for t in b)
+        )
+
+        order = np.random.default_rng(rng.randrange(2**32)).permutation(n)
+        n_cal = min(max(2, round(self.calibration_fraction * n)), n - 2)
+        cal, test = np.sort(order[:n_cal]), np.sort(order[n_cal:])
+
+        metrics: dict[str, float] = {"suite_calibration_sessions": len(cal), "suite_test_sessions": len(test),
+                                     "suite_lag_max_s": self.lag_max_s, "suite_horizon_s": horizon,
+                                     "suite_overrun_sessions": overrun, "suite_dropped_timestamps": dropped}
+        artifacts: dict[str, object] = {"session_ids": np.asarray(sids), "calibration_idx": cal, "test_idx": test}
+        # Raw observations as flat arrays plus offsets, so a run can be
+        # rescored offline without rerunning the relays.
+        for side, series in (("ingress", raw_in), ("egress", raw_out)):
+            artifacts[f"{side}_times"] = np.asarray([t for obs in series for t in obs], dtype=float)
+            artifacts[f"{side}_offsets"] = np.cumsum([0] + [len(obs) for obs in series])
+        for attacker, scores in self._scores(ingress, egress, horizon).items():
+            artifacts[f"scores_{attacker}"] = scores
+            cal_scores = scores[np.ix_(cal, cal)]
+            test_scores = scores[np.ix_(test, test)]
+            true_cal, imp_cal = np.diag(cal_scores), _off_diagonal(cal_scores)
+            true_test, imp_test = np.diag(test_scores), _off_diagonal(test_scores)
+            detail = tpr_at_fpr_detail(true_test, imp_cal, self.fpr_targets, test_impostor_scores=imp_test)
+            for fpr, point in detail.items():
+                metrics[f"{attacker}_tpr_at_fpr_{fpr}"] = point.tpr
+                metrics[f"{attacker}_realized_fpr_{fpr}"] = point.realized_fpr
+                metrics[f"{attacker}_fpr_support_{fpr}"] = point.expected_exceedances
+            for psi in self.psi:
+                for br in self.base_rates:
+                    metrics[f"{attacker}_lpsi_{psi}_br_{br}"] = confident_linkage(true_cal, imp_cal, true_test, psi, br)
+            metrics[f"{attacker}_mu_hat"] = signal_estimate(true_test, imp_test)
+            metrics[f"{attacker}_auc"] = roc_auc(list(true_test), list(imp_test))
+        if "a1" in self.attackers:
+            from ..model import predict  # imported here: model imports this package's delay model
+
+            psi = self.psi[0] if self.psi else 0.9
+            base_rate = self.base_rates[0] if self.base_rates else 1e-4
+            fpr = self.fpr_targets[0] if self.fpr_targets else 1e-3
+            metrics.update(predict(self.config, fpr=fpr, psi=psi, base_rate=base_rate))
+        return AdversaryResult(self.name, n, metrics, artifacts)