diff --git a/tools/simulators/blend/src/blend/config.py b/tools/simulators/blend/src/blend/config.py index fdbdebc..5d77489 100644 --- a/tools/simulators/blend/src/blend/config.py +++ b/tools/simulators/blend/src/blend/config.py @@ -12,6 +12,8 @@ AdversaryMode = Literal[ "random", "worstcase_coverage", "worstcase_eclipse", "worstcase_degree" ] ChurnMode = Literal["uniform", "regional"] +ReleaseMode = Literal["clock", "jitter"] +_RELEASE_MODES = ("clock", "jitter") StakeDist = Literal["uniform", "zipf"] _STAKE_DISTS = ("uniform", "zipf") _DISTS = ("geo", "fixed", "uniform", "exp") @@ -33,6 +35,8 @@ class SimConfig: # --- propagation (Blend cascade, delays in ms) --- blend_hops: int = 3 # relay-path length (swept) max_blend_delay: int = 3 # free-running release-clock max interval, whole SECONDS + min_blend_delay: int = 0 # shortest allowed interval; 1 forbids instant re-release + release_mode: ReleaseMode = "clock" # "clock" = batch at ticks; "jitter" = per-message delay unresponsive_frac: float = 0.0 # ratio of nodes that do NOT relay any messages (swept) churn_mode: ChurnMode = "uniform" # "uniform" = independent nodes; "regional" = whole regions redundancy: int = 1 # copies per emission via R independent cascades (swept) @@ -73,6 +77,11 @@ class SimConfig: raise ValueError(f"need 1 <= blend_hops < n_nodes, got {self.blend_hops}") if self.max_blend_delay < 0: raise ValueError("max_blend_delay must be >= 0 (whole seconds)") + if not (0 <= self.min_blend_delay <= max(self.max_blend_delay, 0)): + raise ValueError( + f"need 0 <= min_blend_delay <= max_blend_delay, got {self.min_blend_delay}") + if self.release_mode not in _RELEASE_MODES: + raise ValueError(f"release_mode must be one of {_RELEASE_MODES}") if not (0.0 <= self.unresponsive_frac < 1.0): raise ValueError(f"need 0 <= unresponsive_frac < 1, got {self.unresponsive_frac}") if self.redundancy < 1: @@ -137,7 +146,7 @@ class SimConfig: """Hashable identity used to seed RNGs deterministically.""" return ( self.n_nodes, self.degree, self.n_regions, self.region_locality, - self.blend_hops, self.max_blend_delay, + self.blend_hops, self.max_blend_delay, self.min_blend_delay, self.release_mode, self.unresponsive_frac, self.churn_mode, self.redundancy, self.cover_rate_mult, self.block_interval_slots, self.slots_per_epoch, self.stake_inference_ratio, self.traffic_window_slots, @@ -163,6 +172,8 @@ class SweepConfig: max_blend_delay: list[int] = field(default_factory=lambda: [3]) unresponsive_frac: list[float] = field(default_factory=lambda: [0.0]) churn_mode: list[str] = field(default_factory=lambda: ["uniform"]) + min_blend_delay: list[int] = field(default_factory=lambda: [0]) + release_mode: list[str] = field(default_factory=lambda: ["clock"]) redundancy: list[int] = field(default_factory=lambda: [1]) cover_rate_mult: list[float] = field(default_factory=list) # empty = no cover-traffic study f_adv: list[float] = field(default_factory=lambda: [0.1, 0.2, 0.33, 0.5]) @@ -202,7 +213,8 @@ class SweepConfig: d = dict(d) base = d.pop("base", {}) known = {"n_nodes", "degree", "blend_hops", "max_blend_delay", "unresponsive_frac", - "churn_mode", "redundancy", "cover_rate_mult", "f_adv", "adversary_mode", + "churn_mode", "min_blend_delay", "release_mode", "redundancy", "cover_rate_mult", + "f_adv", "adversary_mode", "seeds"} unknown = set(d) - known if unknown: diff --git a/tools/simulators/blend/src/blend/mixclock.py b/tools/simulators/blend/src/blend/mixclock.py index d480af1..245792b 100644 --- a/tools/simulators/blend/src/blend/mixclock.py +++ b/tools/simulators/blend/src/blend/mixclock.py @@ -16,21 +16,43 @@ from __future__ import annotations import numpy as np -def mix_wait(rng: np.random.Generator, max_blend_delay: int, size: int) -> np.ndarray: - """``size`` i.i.d. mixing-delay residuals (ms) for a Uniform{0..max_blend_delay}-sec clock.""" - m = int(max_blend_delay) - if m <= 0 or size <= 0: +def mix_wait(rng: np.random.Generator, max_blend_delay: int, size: int, + min_blend_delay: int = 0) -> np.ndarray: + """``size`` i.i.d. mixing-delay residuals (ms) for a Uniform{min..max}-second clock. + + ``min_blend_delay`` excludes short intervals. Note it cannot change the *mean* residual: a + zero-length interval is instantaneous, so it never covers a message arrival and is never + sampled by the size-biased draw. Dropping mass that was never sampled leaves the mean where it + was -- what a minimum actually removes is the chance of an interval too short to mix in. + """ + m, lo = int(max_blend_delay), max(int(min_blend_delay), 0) + if m <= 0 or size <= 0 or lo > m: return np.zeros(max(size, 0), dtype=float) - s = np.arange(1, m + 1, dtype=float) - probs = s / s.sum() # size-biased over positive intervals + s = np.arange(max(lo, 1), m + 1, dtype=float) # zero-length intervals never cover an arrival + probs = s / s.sum() # size-biased over the eligible intervals covering = rng.choice(s, size=size, p=probs) residual_seconds = rng.uniform(0.0, covering) # uniform phase within the covering interval return residual_seconds * 1000.0 # -> milliseconds -def mean_residual_ms(max_blend_delay: int) -> float: - """Analytic mean mixing delay (ms): (2M+1)/6 seconds.""" - m = int(max_blend_delay) - if m <= 0: +def mean_residual_ms(max_blend_delay: int, min_blend_delay: int = 0) -> float: + """Analytic mean mixing delay (ms), the renewal residual ``E[S^2]/(2E[S])``. + + For ``Uniform{0..M}`` this is ``(2M+1)/6`` seconds, and excluding zero-length intervals leaves + it unchanged -- see :func:`mix_wait`. + """ + m, lo = int(max_blend_delay), max(int(min_blend_delay), 0) + if m <= 0 or lo > m: return 0.0 - return (2.0 * m + 1.0) / 6.0 * 1000.0 + s = np.arange(lo, m + 1, dtype=float) + if s.sum() <= 0: + return 0.0 + return float((s * s).sum() / s.sum() / 2.0) * 1000.0 + + +def mean_interval_s(max_blend_delay: int, min_blend_delay: int = 0) -> float: + """Mean gap between releases, ``E[S]`` seconds -- this one *does* move with the minimum.""" + m, lo = int(max_blend_delay), max(int(min_blend_delay), 0) + if m <= 0 or lo > m: + return 0.0 + return float(np.arange(lo, m + 1, dtype=float).mean()) diff --git a/tools/simulators/blend/src/blend/traffic.py b/tools/simulators/blend/src/blend/traffic.py index 84c0168..2eacd8d 100644 --- a/tools/simulators/blend/src/blend/traffic.py +++ b/tools/simulators/blend/src/blend/traffic.py @@ -33,7 +33,7 @@ from scipy.sparse.csgraph import dijkstra from .config import SimConfig from .graph import Graph -from .mixclock import mix_wait +from .mixclock import mean_residual_ms, mix_wait class ReleaseClock: @@ -44,12 +44,14 @@ class ReleaseClock: only the relays a message actually visits ever grow a clock. """ - __slots__ = ("_m", "_rng", "_ticks") + __slots__ = ("_lo", "_m", "_rng", "_ticks") - def __init__(self, max_blend_delay: int, rng: np.random.Generator) -> None: + def __init__(self, max_blend_delay: int, rng: np.random.Generator, + min_blend_delay: int = 0) -> None: self._m = int(max_blend_delay) + self._lo = max(int(min_blend_delay), 0) self._rng = rng - first = 0.0 if self._m <= 0 else float(mix_wait(rng, self._m, 1)[0]) / 1000.0 + first = 0.0 if self._m <= 0 else float(mix_wait(rng, self._m, 1, self._lo)[0]) / 1000.0 self._ticks: list[float] = [first] def next_tick_at_or_after(self, t: float) -> float: @@ -57,7 +59,7 @@ class ReleaseClock: if self._m <= 0: return t while self._ticks[-1] < t: - step = float(self._rng.integers(0, self._m + 1)) + step = float(self._rng.integers(self._lo, self._m + 1)) self._ticks.append(self._ticks[-1] + step) for tick in self._ticks: # tick lists stay short (window / mean interval) if tick >= t: @@ -93,17 +95,18 @@ class TrafficWindow: def _clock(clocks: dict[int, ReleaseClock], node: int, max_blend_delay: int, - rng: np.random.Generator) -> ReleaseClock: + rng: np.random.Generator, min_blend_delay: int = 0) -> ReleaseClock: c = clocks.get(node) if c is None: - c = ReleaseClock(max_blend_delay, rng) + c = ReleaseClock(max_blend_delay, rng, min_blend_delay) clocks[node] = c return c def simulate_window(graph: Graph, config: SimConfig, rng: np.random.Generator, window_slots: int, max_blend_delay: int | None = None, - blend_hops: int | None = None) -> TrafficWindow: + blend_hops: int | None = None, release_mode: str | None = None, + min_blend_delay: int | None = None) -> TrafficWindow: """Play ``window_slots`` seconds of network traffic and record every hold and broadcast. Each slot, the network emits once per ``cover_rate_mult`` on average -- the emitter is uniform @@ -113,6 +116,11 @@ def simulate_window(graph: Graph, config: SimConfig, rng: np.random.Generator, n = graph.n k = int(config.blend_hops if blend_hops is None else blend_hops) m = int(config.max_blend_delay if max_blend_delay is None else max_blend_delay) + lo = int(config.min_blend_delay if min_blend_delay is None else min_blend_delay) + mode = release_mode or config.release_mode + # Matched delay budget: exponential jitter with the same mean hold as the clock, so the two + # designs are compared at equal latency cost and differ only in HOW they delay. + jitter_mean_s = mean_residual_ms(m, lo) / 1000.0 f = 1.0 / config.block_interval_slots win = TrafficWindow(window_seconds=float(window_slots)) clocks = win.clocks @@ -154,7 +162,12 @@ def simulate_window(graph: Graph, config: SimConfig, rng: np.random.Generator, dropped = True break arrived = t + float(leg) / 1000.0 - released = _clock(clocks, int(relays[hop]), m, rng).next_tick_at_or_after(arrived) + if mode == "jitter": + # each message waits its own independent draw -- no batching, no quantisation + released = arrived + float(rng.exponential(jitter_mean_s)) if m > 0 else arrived + else: + released = _clock(clocks, int(relays[hop]), m, rng, + lo).next_tick_at_or_after(arrived) win.holds.append(Hold(int(relays[hop]), arrived, released)) t = released if not dropped: @@ -243,3 +256,79 @@ def traffic_metrics(win: TrafficWindow, config: SimConfig, out.update(blending_mean=float(arr.mean()), blending_p50=float(np.percentile(arr, 50)), blending_p90=float(np.percentile(arr, 90)), blending_max=float(arr.max())) return out + + +def timing_linkability(win: TrafficWindow, config: SimConfig, + max_blend_delay: int | None = None, + min_blend_delay: int | None = None, + release_mode: str | None = None) -> dict: + """Can an observer match a relay's outgoing message to the incoming one, from timing alone? + + This is the attack that distinguishes a *blended* message from a merely *relayed* one: a relay + that holds and re-emits leaves a timing signature, and if only one arrival can explain a given + release then the two are linked and the relay's role in that cascade is exposed. + + The measure is the **effective anonymity set** of each release -- the perplexity of the + observer's posterior over which arrival produced it. It is comparable across designs: + + * ``clock`` -- a release at a tick could be any arrival since the previous tick, all equally + likely, so the effective set is simply the batch size. + * ``jitter`` -- each message waits an independent draw, so every earlier arrival is a candidate + weighted by the delay density (exponential here). Nothing is quantised, so the weights decay + smoothly and the posterior concentrates on whichever arrival is closest to the expected lag. + + ``linked_frac`` is the share of releases whose set collapses to one candidate: the message is + then linked with certainty, whatever the nominal delay was. + + The effective set alone flatters a heavy-tailed delay, because a long thin tail keeps old + arrivals nominally "possible" while contributing almost nothing. ``map_success`` therefore also + reports how often the adversary's single best guess is right -- for an exponential delay the + most likely source is always the most recent arrival, so a design can look unlinkable by + perplexity and still be guessed correctly most of the time. + """ + m = int(config.max_blend_delay if max_blend_delay is None else max_blend_delay) + lo = int(config.min_blend_delay if min_blend_delay is None else min_blend_delay) + mode = release_mode or config.release_mode + if not win.holds or m <= 0: + return {"timing_set_mean": 1.0, "timing_set_p90": 1.0, "timing_linked_frac": 1.0} + mean_hold = mean_residual_ms(m, lo) / 1000.0 + by_node: dict[int, list[Hold]] = {} + for h in win.holds: + by_node.setdefault(h.node, []).append(h) + + sets: list[float] = [] + hits: list[float] = [] + for node, holds in by_node.items(): + arrivals = np.sort(np.array([h.arrived for h in holds])) + true_src = {h.released: h.arrived for h in holds} # the arrival that really produced it + for r in sorted({h.released for h in holds}): + if mode == "clock": + clock = win.clocks.get(node) + prev = 0.0 + if clock is not None: + earlier = [t for t in clock.ticks_in(-1e18, r) if t < r] + if earlier: + prev = earlier[-1] + cand = arrivals[(arrivals > prev) & (arrivals <= r + 1e-12)] + n_c = max(len(cand), 1) + sets.append(float(n_c)) # uniform posterior -> perplexity = n + hits.append(1.0 / n_c) # MAP is a coin flip among the batch + else: + earlier = arrivals[arrivals <= r + 1e-12] + if earlier.size == 0: + sets.append(1.0) + hits.append(1.0) + continue + w = np.exp(-(r - earlier) / mean_hold) # exponential delay density + p = w / w.sum() + ent = float(-np.sum(p * np.log(np.clip(p, 1e-300, None)))) + sets.append(float(np.exp(ent))) # perplexity = effective set size + # MAP for an exponential is the most recent arrival; is that the true source? + hits.append(float(abs(earlier[int(np.argmax(p))] - true_src[r]) < 1e-12)) + arr = np.asarray(sets) + return { + "timing_set_mean": float(arr.mean()), + "timing_set_p90": float(np.percentile(arr, 90)), + "timing_linked_frac": float(np.mean(arr < 1.5)), # effectively a forced match + "map_success": float(np.mean(hits)) if hits else 1.0, # best single guess is correct + } diff --git a/tools/simulators/blend/tests/test_timing.py b/tools/simulators/blend/tests/test_timing.py new file mode 100644 index 0000000..2a20244 --- /dev/null +++ b/tools/simulators/blend/tests/test_timing.py @@ -0,0 +1,90 @@ +"""Release designs: a minimum interval, and jitter vs clock-tick release under a timing attack.""" + +import numpy as np + +from blend.config import SimConfig +from blend.graph import build_graph +from blend.mixclock import mean_interval_s, mean_residual_ms, mix_wait +from blend.traffic import ReleaseClock, simulate_window, timing_linkability, traffic_metrics + + +def _run(mode="clock", M=30, lo=0, rate=1.0, slots=120, n=2000, seed=3): + cfg = SimConfig(n_nodes=n, degree=8, blend_hops=3, max_blend_delay=M, min_blend_delay=lo, + release_mode=mode, cover_rate_mult=rate) + g = build_graph(cfg) + w = simulate_window(g, cfg, np.random.default_rng(seed), slots) + return traffic_metrics(w, cfg), timing_linkability(w, cfg) + + +# --- the minimum interval ------------------------------------------------------------------------- + +def test_a_minimum_interval_does_not_change_the_mean_hold(): + """A zero-length gap is instantaneous, so it never covers an arrival and is never sampled. + Excluding it removes mass the residual never saw -- the mean hold is identical.""" + for M in (3, 10, 30): + assert abs(mean_residual_ms(M, 0) - mean_residual_ms(M, 1)) < 1e-9 + + +def test_a_minimum_interval_does_lengthen_the_gap_between_releases(): + """What it does change is E[S]: release opportunities become rarer.""" + for M in (3, 10, 30): + assert mean_interval_s(M, 1) > mean_interval_s(M, 0) + + +def test_sampled_holds_match_the_analytic_mean_with_and_without_a_minimum(): + rng = np.random.default_rng(0) + for M in (3, 30): + for lo in (0, 1): + got = float(np.mean(mix_wait(rng, M, 60_000, lo))) + assert abs(got - mean_residual_ms(M, lo)) < 0.05 * mean_residual_ms(M, lo) + + +def test_clock_respects_the_minimum_interval(): + c = ReleaseClock(5, np.random.default_rng(0), min_blend_delay=2) + c.next_tick_at_or_after(200.0) + gaps = [b - a for a, b in zip(c._ticks, c._ticks[1:], strict=False)] + assert all(2 - 1e-9 <= g <= 5 + 1e-9 for g in gaps) + + +def test_the_minimum_does_not_measurably_change_anonymity(): + """Follows from the mean hold being unchanged: blending and linkability track it.""" + a, ta = _run(lo=0) + b, tb = _run(lo=1) + assert abs(a["hold_seconds_mean"] - b["hold_seconds_mean"]) < 0.5 + assert abs(ta["timing_linked_frac"] - tb["timing_linked_frac"]) < 0.05 + + +# --- jitter vs clock ------------------------------------------------------------------------------ + +def test_both_designs_cost_the_same_delay(): + """The comparison is only meaningful at a matched latency budget.""" + c, _ = _run("clock") + j, _ = _run("jitter") + assert abs(c["hold_seconds_mean"] - j["hold_seconds_mean"]) < 1.0 + + +def test_timing_linkage_is_near_total_at_the_baseline_rate(): + """The headline: a relay handles so little traffic that in->out matching is trivial under + EITHER design, so neither provides timing protection at one message per second.""" + for mode in ("clock", "jitter"): + _, t = _run(mode, rate=1.0) + assert t["map_success"] > 0.9 + assert t["timing_set_mean"] < 1.3 + + +def test_more_traffic_is_what_buys_timing_protection(): + _, lo_rate = _run("clock", rate=1.0) + _, hi_rate = _run("clock", rate=64.0, slots=60) + assert hi_rate["timing_set_mean"] > lo_rate["timing_set_mean"] + assert hi_rate["map_success"] < lo_rate["map_success"] + + +def test_perplexity_flatters_jitter_more_than_the_best_guess_does(): + """A heavy tail keeps old arrivals nominally possible while contributing almost nothing, so + the effective-set advantage of jitter overstates its real advantage under a MAP attack.""" + _, c = _run("clock", rate=64.0, slots=60) + _, j = _run("jitter", rate=64.0, slots=60) + set_gain = j["timing_set_mean"] / c["timing_set_mean"] + map_gain = (1 - j["map_success"]) / (1 - c["map_success"]) + assert set_gain > 1.0 and map_gain > 1.0 # jitter wins on both + assert set_gain > map_gain # but the set measure overstates by how much