run_tor_timing_study.py

#!/usr/bin/env python3
"""Run the Tor guard-fingerprint timing attack study.

Scores the residual anonymity of a WabiSabi round under the guard-fingerprint
timing attack, on a real per-guard latency dataset measured against the live Tor
network (``data/tor_latency_dataset.json``), falling back to a synthetic model if
no dataset is present. Shows that a single request barely beats chance, but the
many guard-pinned requests of a round amplify the fixed guard offset into a
material anonymity loss, that the pre-established-circuit-pool fix does not remove
the residual guard-RTT fingerprint, and that timing randomization does (at a cost).

Writes ``tor_timing_results.json``.

Usage::

    # measure a fresh real dataset first (needs a running tor + control port):
    python -m coinjoin_simulator.tor_timing.tor_probe --control-port 9051 \
        --guards 8 --builds 12 --rtts 12 --out data/tor_latency_dataset.json

    python run_tor_timing_study.py
    python run_tor_timing_study.py --quick
"""

from __future__ import annotations

import argparse
import json
import time
from pathlib import Path

from coinjoin_simulator.tor_timing.latency import GuardLatency, TorLatencyDataset
from coinjoin_simulator.tor_timing.study import TimingStudyConfig, build_study

OUTPUT_PATH = Path("tor_timing_results.json")
DATASET_PATH = Path("data/tor_latency_dataset.json")


def _synthetic_dataset() -> TorLatencyDataset:
    """A calibrated synthetic fallback when no real measurement is available."""
    import random

    rng = random.Random(0)
    guards = []
    for i in range(8):
        offset = rng.gauss(0.0, 167.0)
        rtt_offset = rng.gauss(0.0, 23.0)
        build = tuple(max(1.0, 1150 + offset + rng.gauss(0, 461)) for _ in range(12))
        rtt = tuple(max(1.0, 410 + rtt_offset + rng.gauss(0, 56)) for _ in range(12))
        guards.append(
            GuardLatency(
                guard_fingerprint=f"SYNTH{i:039d}",
                guard_nickname=f"synth{i}",
                build_ms=build,
                guard_rtt_ms=rtt,
            )
        )
    return TorLatencyDataset(source="synthetic", tor_version="n/a", guards=tuple(guards))


def main() -> None:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--quick", action="store_true", help="fewer seeds / smaller sweeps")
    parser.add_argument("--dataset", default=str(DATASET_PATH), help="latency dataset path")
    args = parser.parse_args()

    dataset_path = Path(args.dataset)
    if dataset_path.exists():
        dataset = TorLatencyDataset.load(dataset_path)
        print(f"Loaded {dataset.source} dataset: {len(dataset.guards)} guards "
              f"(tor {dataset.tor_version})")
    else:
        dataset = _synthetic_dataset()
        print(f"No dataset at {dataset_path}; using synthetic fallback "
              f"({len(dataset.guards)} guards)")

    cfg = (
        TimingStudyConfig(seeds=15, reject_seeds=80, size_sweep=(8, 20, 50))
        if args.quick
        else TimingStudyConfig()
    )

    start = time.time()
    payload = build_study(dataset, cfg)
    payload["generated_at"] = time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime())
    OUTPUT_PATH.write_text(json.dumps(payload, indent=2) + "\n")

    _print_summary(payload)
    print(f"\nWrote {OUTPUT_PATH} in {time.time() - start:.1f}s")


def _print_summary(payload: dict) -> None:
    h = payload["headline"]
    print(
        f"\nSource: {h['source']} (tor {h['tor_version']}), {h['n_guards']} guards; "
        f"build within/across ratio {h['build_variance_ratio']:.2f}, "
        f"guard_rtt {h['guard_rtt_variance_ratio']:.2f}"
    )

    def amp(title: str, rows: list[dict]) -> None:
        print(f"\n{title}")
        print(f"  {'q':>4} {'eff/N':>7} {'degree':>7} {'deanon':>7} {'conf':>6}")
        for r in rows:
            print(f"  {int(r['q']):>4} {r['effective_set']:>7.2f} {r['degree']:>7.3f} "
                  f"{r['deanonymized_fraction']:>7.2f} {r['confidence']:>6.2f}")

    amp("P0 amplification, no-defense build latency (real guards)",
        payload["p0_amplification_build"])
    amp("P0 amplification, pool-defense residual guard RTT (real guards)",
        payload["p0_amplification_guard_rtt"])

    print("\nP1 round size (synthetic, calibrated to real; q="
          f"{payload['config']['size_sweep']} at fixed amplification)")
    print(f"  {'N':>5} {'eff':>7} {'eff/N':>7} {'degree':>7} {'deanon':>7}")
    for r in payload["p1_round_size"]:
        print(f"  {int(r['n_clients']):>5} {r['effective_set']:>7.2f} "
              f"{r['normalized_effective_set']:>7.3f} {r['degree']:>7.3f} "
              f"{r['deanonymized_fraction']:>7.2f}")

    print("\nP2 timing-randomization defense (build, high amplification)")
    print(f"  {'extra_ms':>9} {'eff/N':>7} {'degree':>7} {'deanon':>7}")
    for r in payload["p2_jitter_defense"]:
        print(f"  {r['extra_jitter_ms']:>9.0f} {r['effective_set']:>7.2f} "
              f"{r['degree']:>7.3f} {r['deanonymized_fraction']:>7.2f}")

    print("\nP3 active amplification (reject low-confidence rounds)")
    print(f"  {'thresh':>7} {'accept':>7} {'deanon_all':>11} {'deanon_kept':>12}")
    for r in payload["p3_active_rejection"]:
        print(f"  {r['threshold']:>7.2f} {r['accept_rate']:>7.2f} "
              f"{r['deanonymized_all']:>11.2f} {r['deanonymized_accepted']:>12.2f}")

    print("\nP4 retry intersection (same coin re-registered across failed rounds)")
    print(f"  {'rounds':>6} {'per-round':>10} {'any-round':>10} {'best-conf':>10} "
          f"{'active':>7} {'commit':>7}")
    for r in payload["p4_retry_intersection"]:
        print(f"  {int(r['n_rounds']):>6} {r['single_round_deanon']:>10.2f} "
              f"{r['any_round_deanon']:>10.2f} {r['best_confidence_deanon']:>10.2f} "
              f"{r['active_disruption_deanon']:>7.2f} {r['active_commit_rate']:>7.2f}")

    cfg = payload["config"]
    print(f"\nP5 access-latency sensitivity ({cfg['access_observable']}, "
          f"q={cfg['access_q']}, per-client last-mile offset)")
    print(f"  {'spread_ms':>9} {'eff/N':>7} {'deanon':>7} {'conf':>6}")
    for r in payload["p5_access_latency"]:
        print(f"  {r['access_spread_ms']:>9.0f} {r['effective_set']:>7.2f} "
              f"{r['deanonymized_fraction']:>7.2f} {r['confidence']:>6.2f}")


if __name__ == "__main__":
    main()