research/tools/simulators/blend/pd/tests/test_adversary.py
Marcin Pawlowski 6ad63ce2f3
Add pd: peering-degree Blend Monte-Carlo graph simulator
Static-graph simulator quantifying how a node's peering degree trades off
propagation speed, adversary exposure, deanonymization, and reliability in the
Blend network. Scales to 1e6 nodes (sparse CSR + sampled Dijkstra); the
adversary and deanonymization metrics are exact at every N.

Model (ms): seeded d-regular peer graph (matching-union), Blend cascade
(sender -> blend_hops timed-release mix relays -> final flood), geographic link
base + exponential transport jitter, per-node processing lag, free-running
release-clock mixing.

Metrics:
- propagation: full-delay mean/p50/p90/p99, path/broadcast split, coverage times
- reliability: message success-delivery-rate ~ (1-unresponsive_frac)^blend_hops
  and flood coverage, with unresponsive nodes modelled as routing holes
- adversary (exact): observed/eclipsed fractions, random + worst-case placement
- deanonymization (exact): P(whole blend path adversarial) ~ f_adv^blend_hops,
  and full deanonymization (path adversarial AND honest sender peered with an
  adversary) = deanon_rate * observed_frac

Deterministic blake2b seed streams, three parquet tables, joblib parallelism,
memguard, an analytic verify harness, 50 unit tests, and an auto-installing
Makefile.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
2026-08-06 17:59:55 +02:00

63 lines
2.3 KiB
Python

from itertools import combinations
import numpy as np
from pd.adversary import _greedy_coverage, adversary_metrics, place_adversary
from pd.config import SimConfig
from pd.graph import Graph, build_graph
def _cycle4():
indptr = np.array([0, 2, 4, 6, 8], dtype=np.int64)
indices = np.array([1, 3, 0, 2, 1, 3, 0, 2], dtype=np.int64)
return Graph(n=4, degree=2, indptr=indptr, indices=indices,
base=np.ones(8), src=np.array([0, 0, 1, 1, 2, 2, 3, 3]), p=np.zeros(4))
def test_coverage_eclipse_hand_checked():
g = _cycle4() # 0-1-2-3-0
m = adversary_metrics(g, np.array([False, True, False, True])) # adv {1,3}
assert m["observed_count"] == 2 and m["eclipsed_count"] == 2 # honest {0,2} fully surrounded
m = adversary_metrics(g, np.array([False, True, False, False])) # adv {1}
assert m["observed_count"] == 2 and m["eclipsed_count"] == 0 # {0,2} observed, none eclipsed
def test_random_closed_form():
g = build_graph(SimConfig(n_nodes=5000, degree=6, graph_seed=0))
rng = np.random.default_rng(0)
def _obs():
return adversary_metrics(g, place_adversary(g, 0.2, "random", rng, 10**9))["observed_frac"]
obs = np.mean([_obs() for _ in range(5)])
assert abs(obs - (1 - 0.8 ** 6)) < 0.02
def test_worstcase_coverage_is_an_envelope():
g = build_graph(SimConfig(n_nodes=400, degree=4, graph_seed=0))
rng = np.random.default_rng(0)
rand = adversary_metrics(g, place_adversary(g, 0.2, "random", rng, 10**9))["observed_frac"]
wc = adversary_metrics(
g, place_adversary(g, 0.2, "worstcase_coverage", rng, 10**9))["observed_frac"]
assert wc >= rand - 1e-9
def test_greedy_coverage_near_optimal():
g = build_graph(SimConfig(n_nodes=10, degree=3, graph_seed=0))
best = max(adversary_metrics(g, _mask(10, c))["observed_count"]
for c in combinations(range(10), 2))
idx = _greedy_coverage(g, 2, np.random.default_rng(0))
got = adversary_metrics(g, _mask(10, idx))["observed_count"]
assert got >= (1 - 1 / np.e) * best - 1e-9
def _mask(n, idx):
m = np.zeros(n, dtype=bool)
m[list(idx)] = True
return m
def test_worstcase_cap_raises():
import pytest
g = build_graph(SimConfig(n_nodes=200, degree=4))
with pytest.raises(ValueError):
place_adversary(g, 0.2, "worstcase_coverage", np.random.default_rng(0), worstcase_max_n=100)