leetcrypt
8224a38072
confirm_load.py reconstructs the RQ1 (entry,exit) circuit-pair set from the REAL per-hop pcaps (bin_pcap_bytes -> score_matrix; diagonal=linked, off-diagonal=unlinked) so confirm.rq1_p1_leak scores measured data — the frozen §4 unit. classify_run reads persisted metrics.json; collect_rq1_p1_pairs selects only bridge=on no-pad runs. Plumbing unit-tested on SYNTHETIC pcaps only (prereg §2 blinding — no inspecting intermediate confirmatory results before the battery completes). RQ1-P2 nopad/pad pairing + all RQ2 held for the operator ruling (NEEDS-OPERATOR: RQ2 per-circuit posterior underspecified). Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
81 lines
3.7 KiB
Python
81 lines
3.7 KiB
Python
"""SS3 RQ1 loader — pcap→pairs plumbing, on SYNTHETIC pcaps only.
|
|
|
|
These pin that the loader reconstructs the RQ1 (entry,exit) pair set from real
|
|
per-hop pcap bytes and classifies runs by their persisted metrics.json — WITHOUT
|
|
touching any confirmatory data (prereg §2 blinding). Detector accuracy itself is
|
|
covered by test_sor_analysis/test_sor_confirm; here we only prove the wiring.
|
|
"""
|
|
|
|
import json
|
|
import struct
|
|
|
|
from cmd_chat.sor.analysis import confirm_load
|
|
|
|
|
|
def _write_series_pcap(path, series):
|
|
"""Minimal LINKTYPE_RAW pcap placing one packet of size series[k] at t=k, so
|
|
bin_pcap_bytes(path, bins=len(series)) reproduces `series` bin-for-bin."""
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
with open(path, "wb") as f:
|
|
f.write(struct.pack("<IHHiIII", 0xA1B2C3D4, 2, 4, 0, 0, 65535, 101))
|
|
for k, nbytes in enumerate(series):
|
|
f.write(struct.pack("<IIII", k, 0, nbytes, nbytes))
|
|
f.write(b"x" * nbytes)
|
|
|
|
|
|
def _write_circuit(run_dir, cid, ingress_series, egress_series):
|
|
pdir = run_dir / "circuits" / cid / "pcap"
|
|
_write_series_pcap(pdir / "hop0.pcap", ingress_series) # ingress
|
|
_write_series_pcap(pdir / "hop2.pcap", egress_series) # egress (hops=3)
|
|
|
|
|
|
def _write_metrics(run_dir, cell_id, rq, padding):
|
|
run_dir.mkdir(parents=True, exist_ok=True)
|
|
(run_dir / "metrics.json").write_text(
|
|
json.dumps({"cell_id": cell_id, "rq": rq, "padding_applied": padding}),
|
|
encoding="utf-8",
|
|
)
|
|
|
|
|
|
def test_classify_run_reads_persisted_metrics(tmp_path):
|
|
rd = tmp_path / "cell-x-r0"
|
|
_write_metrics(rd, "RQ1/topo=1house/selector=static/bridge=on", "RQ1", False)
|
|
info = confirm_load.classify_run(rd)
|
|
assert info["cell_id"].endswith("bridge=on")
|
|
assert info["rq"] == "RQ1" and info["padding_applied"] is False
|
|
|
|
|
|
def test_reconstruct_run_pairs_shape_and_linkage(tmp_path):
|
|
rd = tmp_path / "cell-on-r0"
|
|
_write_metrics(rd, "RQ1/topo=1house/selector=static/bridge=on", "RQ1", False)
|
|
# Two circuits with anti-correlated temporal profiles; each hop0/hop2 share
|
|
# their own circuit's profile → linked (diagonal) pearson high, cross low.
|
|
_write_circuit(rd, "c0", [100, 10, 100, 10], [100, 10, 100, 10])
|
|
_write_circuit(rd, "c1", [10, 100, 10, 100], [10, 100, 10, 100])
|
|
pairs = confirm_load.reconstruct_run_pairs(rd, bins=4, hops=3)
|
|
assert len(pairs) == 4 # 2x2 score matrix
|
|
linked = [s for s, is_linked in pairs if is_linked]
|
|
unlinked = [s for s, is_linked in pairs if not is_linked]
|
|
assert len(linked) == 2 and len(unlinked) == 2
|
|
# Real measured signal carried through: same-circuit correlation beats cross.
|
|
assert min(linked) > max(unlinked)
|
|
|
|
|
|
def test_collect_rq1_p1_selects_only_bridge_on_nopad(tmp_path):
|
|
# bridge=on (leak arm) — should be collected.
|
|
on = tmp_path / "cell-on-r0"
|
|
_write_metrics(on, "RQ1/topo=1house/selector=static/bridge=on", "RQ1", False)
|
|
_write_circuit(on, "c0", [100, 10, 100, 10], [100, 10, 100, 10])
|
|
# bridge=on+padding — excluded (padding arm).
|
|
pad = tmp_path / "cell-pad-r0"
|
|
_write_metrics(pad, "RQ1/topo=1house/selector=static/bridge=on+padding", "RQ1", True)
|
|
_write_circuit(pad, "c0", [50, 50, 50, 50], [50, 50, 50, 50])
|
|
# RQ2 cell — excluded (wrong RQ).
|
|
rq2 = tmp_path / "cell-rq2-r0"
|
|
_write_metrics(rq2, "RQ2/topo=directory-federated/selector=static/bridge=off", "RQ2", False)
|
|
_write_circuit(rq2, "c0", [1, 2, 3, 4], [1, 2, 3, 4])
|
|
|
|
pairs = confirm_load.collect_rq1_p1_pairs(tmp_path, bins=4, hops=3)
|
|
assert len(pairs) == 1 # only the 1-circuit bridge=on run
|
|
assert pairs[0][1] is True # its single diagonal pair
|