leetcrypt
ce68d41348
Measures RQ1 bridge-correlation AUC + RQ2 Shannon entropy from REAL per-hop pcaps of live isolated-docker circuits. run_battery(live=False) hard-raises and the executor refuses (ExecutorError) any DV it did not measure — the executor-side twin of the launcher's "never fabricate cells" guard. Wired into confirmatory_run's triple-locked tokened GO (executor.run_battery(live=True) behind operator token + verified frozen prereg SHA + engine!=local + green preflight + full grid). Containment intact: self-fixture bytes only, isolated engine only, no external target. Frozen prereg untouched (SHA f22331a72e…). Verify: test_sor_executor.py 7 + confirmatory_run + full SOR suite = 163 passed; preflight green (grid 3/3, READY); no confirmatory data collected. Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
66 lines
2.3 KiB
Python
66 lines
2.3 KiB
Python
"""Confirmatory data-collection executor — anti-fabrication guard + real pcap
|
|
measurement. The heavy live-docker path is exercised by the operator rehearsal;
|
|
here we pin the offline invariants that keep the confirmatory DVs honest.
|
|
"""
|
|
|
|
import struct
|
|
|
|
import pytest
|
|
|
|
from cmd_chat.sor import executor
|
|
|
|
|
|
def test_run_battery_refuses_non_live():
|
|
# The executor collects DVs ONLY from real delivered circuits — never a
|
|
# synthetic/dry confirmatory path.
|
|
with pytest.raises(executor.ExecutorError):
|
|
executor.run_battery("output/never", r_runs=1, c_circuits=1, live=False)
|
|
|
|
|
|
def test_measure_flow_refuses_missing_pcap():
|
|
with pytest.raises(executor.ExecutorError):
|
|
executor._measure_flow({0: "only-ingress"}, last_hop=2, bins=8)
|
|
|
|
|
|
def test_bin_pcap_bytes_refuses_unreadable(tmp_path):
|
|
bad = tmp_path / "not.pcap"
|
|
bad.write_bytes(b"not a pcap file")
|
|
with pytest.raises(executor.ExecutorError):
|
|
executor.bin_pcap_bytes(bad, bins=8)
|
|
|
|
|
|
def _write_minimal_pcap(path, packets):
|
|
"""A minimal little-endian pcap (LINKTYPE_RAW=101) with the given (ts, payload)
|
|
packets, so we can exercise real binning without docker."""
|
|
with open(path, "wb") as f:
|
|
# global header: magic, ver 2.4, tz/sig 0, snaplen, network=101 (RAW)
|
|
f.write(struct.pack("<IHHiIII", 0xA1B2C3D4, 2, 4, 0, 0, 65535, 101))
|
|
for ts, payload in packets:
|
|
sec = int(ts)
|
|
usec = int((ts - sec) * 1_000_000)
|
|
f.write(struct.pack("<IIII", sec, usec, len(payload), len(payload)))
|
|
f.write(payload)
|
|
|
|
|
|
def test_bin_pcap_bytes_bins_real_bytes(tmp_path):
|
|
p = tmp_path / "cap.pcap"
|
|
# Two packets at t=0 and t=1, 100 bytes each, into 2 bins → [100, 100].
|
|
_write_minimal_pcap(p, [(0.0, b"x" * 100), (1.0, b"y" * 100)])
|
|
series = executor.bin_pcap_bytes(p, bins=2)
|
|
assert len(series) == 2
|
|
assert sum(series) == 200
|
|
assert series[0] == 100 and series[-1] == 100
|
|
|
|
|
|
def test_bin_pcap_bytes_empty_is_zero_series(tmp_path):
|
|
p = tmp_path / "empty.pcap"
|
|
_write_minimal_pcap(p, [])
|
|
assert executor.bin_pcap_bytes(p, bins=4) == [0, 0, 0, 0]
|
|
|
|
|
|
def test_circuit_seed_is_deterministic_and_distinct():
|
|
a = executor._circuit_seed(12345, 0)
|
|
b = executor._circuit_seed(12345, 0)
|
|
c = executor._circuit_seed(12345, 1)
|
|
assert a == b and a != c
|