logos-blockchain/research
1
0
Fork 0
mirror of https://github.com/logos-blockchain/research.git synced 2026-08-07 03:33:33 +00:00
Code Issues Packages Projects Releases Wiki Activity

blend: minimum release interval, and jitter vs clock under a timing attack

Browse Source 
Two release designs at a MATCHED delay budget, so they differ only in how they
delay: clock (batch at free-running ticks, the existing model) and jitter (each
message waits its own exponential draw, mean set equal to the clock residual).
Plus min_blend_delay, which forbids intervals shorter than it.

Minimum interval -- a negative result, and provably so. A zero-length gap is
instantaneous, so it never covers an arrival and is never sampled by the residual
or by the size-biased interval. Excluding it therefore leaves the mean hold
exactly unchanged, and with it blending and linkability; what it does change is
E[S], the gap between release opportunities. Confirmed analytically and in
simulation: 1.168s vs 1.167s at M=3.

Timing attack -- the effective anonymity set of a release (perplexity of the
observer posterior over which arrival produced it), plus MAP success, the chance
its single best guess is right. The second matters because perplexity flatters a
heavy tail: an exponential never fully excludes an old arrival, so it can look
unlinkable while still being guessed correctly.

At the baseline rate BOTH designs fail almost completely -- MAP success 0.98-0.99,
effective set ~1. A relay handles so little traffic that in->out matching is
trivial, which follows directly from the mixing~0 result. Traffic, not delay, is
what buys timing protection.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
This commit is contained in:
Marcin Pawlowski 2026-08-06 12:41:39 +02:00 committed by Marcin Pawlowski
parent 3c7fef64d0
commit 57be929af7
No known key found for this signature in database
4 changed files with 235 additions and 22 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

16
tools/simulators/blend/src/blend/config.py
View File

@ -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:

44
tools/simulators/blend/src/blend/mixclock.py
View File

@ -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())

107
tools/simulators/blend/src/blend/traffic.py
View File

@ -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
}

90
tools/simulators/blend/tests/test_timing.py Normal file
View File

@ -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