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>
334 lines
14 KiB
Python
334 lines
14 KiB
Python
"""Confirmatory data-collection executor — the human-gated live data run.
|
||
|
||
This is the traffic-moving half of the RQ1+RQ2 battery: for every frozen §2 cell
|
||
in the randomized/interleaved schedule it stands up the cell's assembled,
|
||
condition-encoding circuit as an **isolated-docker** nested-SSH chain (via the
|
||
gate-item-1 :func:`forwarder.run_circuit_fixture`), pipes ``C`` seed-deterministic
|
||
**self-generated** flows through it, and **measures the real per-hop pcaps** to
|
||
derive the pre-registered DVs:
|
||
|
||
* **RQ1 — bridge linkability:** the correlator's ingress↔egress AUC computed on
|
||
per-bin byte-count series *read back out of the captured pcaps* (``scapy``),
|
||
not synthesized. The bridge-on+padding arm injects the R1 PADDING cover stream
|
||
so its egress timing genuinely diverges from ingress — a measured effect.
|
||
* **RQ2 — anonymity set:** the Shannon entropy (bits) of the *realized* entry-
|
||
node distribution over the ``C`` assembled circuits — a measurement of the
|
||
cell's selection over its consenting-node pool (single-house-N vs federated).
|
||
|
||
Containment (CLAUDE.md §Containment, load-bearing):
|
||
* every hop runs inside an isolated docker container — ``assert engine != local``
|
||
is re-checked per hop by :class:`forwarder.ForwarderPlan`; the host never
|
||
forwards. Only self-generated fixture bytes move, between our own containers.
|
||
* **no DV is ever fabricated.** The executor refuses to emit a confirmatory
|
||
metric that was not measured from a real delivered circuit + real pcap
|
||
(``ContainmentError`` / ``ExecutorError`` instead). It is the anti-fabrication
|
||
counterpart to the launcher guard.
|
||
|
||
The full frozen battery (R=30 × C=50 over the 6 cells) is the operator's explicit
|
||
GO (``confirmatory_run --operator-go`` + token). A reduced ``run_battery`` is used
|
||
for the live rehearsal on a non-confirmatory dir; it collects real measurements
|
||
but is not the pre-registered battery.
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import hashlib
|
||
import json
|
||
import shutil
|
||
from dataclasses import dataclass, field
|
||
from pathlib import Path
|
||
from typing import Dict, List, Optional, Sequence, Tuple
|
||
|
||
from cmd_chat.sor import battery as sor_battery
|
||
from cmd_chat.sor.analysis.detectors import bridge_correlation_auc, shannon_entropy_bits
|
||
from cmd_chat.sor.analysis.metrics import compute_metrics, write_metrics
|
||
from cmd_chat.sor.assembler import assemble
|
||
from cmd_chat.sor.config import Domain, SorRng
|
||
from cmd_chat.sor.forwarder import CircuitError, assert_isolated, run_circuit_fixture
|
||
from cmd_chat.sor.provenance import Node, RunManifest, write_manifest
|
||
from cmd_chat.sor.selector import SelectionResult
|
||
|
||
DEFAULT_BINS = 32
|
||
|
||
|
||
class ExecutorError(RuntimeError):
|
||
"""The executor could not collect a real measurement and refuses to emit a
|
||
(would-be fabricated) confirmatory DV in its place."""
|
||
|
||
|
||
# --------------------------------------------------------------------------- #
|
||
# Real pcap measurement (scapy) — never synthetic in the confirmatory path.
|
||
# --------------------------------------------------------------------------- #
|
||
def bin_pcap_bytes(pcap_path: Path, bins: int = DEFAULT_BINS) -> List[int]:
|
||
"""Read ``pcap_path`` and return a per-bin total-byte series: packet lengths
|
||
summed into ``bins`` equal time-bins across the capture window. This is a real
|
||
measurement of the captured (SSH-ciphertext) flow — the correlator input.
|
||
|
||
Raises :class:`ExecutorError` if the pcap cannot be read (we refuse to invent
|
||
a series). An empty capture yields an all-zero series (a real null observation)."""
|
||
from scapy.utils import rdpcap # local import: heavy dep, only for live runs
|
||
|
||
try:
|
||
packets = rdpcap(str(pcap_path))
|
||
except Exception as exc: # noqa: BLE001
|
||
raise ExecutorError(f"cannot read pcap {pcap_path}: {exc}") from exc
|
||
|
||
series = [0] * bins
|
||
if len(packets) == 0:
|
||
return series
|
||
times = [float(p.time) for p in packets]
|
||
t0, t1 = min(times), max(times)
|
||
span = (t1 - t0) or 1.0
|
||
for p, t in zip(packets, times):
|
||
idx = int((t - t0) / span * bins)
|
||
if idx >= bins:
|
||
idx = bins - 1
|
||
series[idx] += len(p)
|
||
return series
|
||
|
||
|
||
def _measure_flow(pcaps: Dict[int, Path], last_hop: int, bins: int) -> Tuple[List[int], List[int]]:
|
||
"""Ingress = entry-hop (hop0) pcap binned; egress = exit-hop (last) pcap binned.
|
||
Both read from real captures. Refuses if either pcap is missing."""
|
||
if 0 not in pcaps or last_hop not in pcaps:
|
||
raise ExecutorError(
|
||
f"missing pcap for ingress(0)/egress({last_hop}); have {sorted(pcaps)}"
|
||
)
|
||
return bin_pcap_bytes(pcaps[0], bins), bin_pcap_bytes(pcaps[last_hop], bins)
|
||
|
||
|
||
# --------------------------------------------------------------------------- #
|
||
# One (cell, run): C live circuits -> measured DVs -> provenance + metrics.
|
||
# --------------------------------------------------------------------------- #
|
||
@dataclass
|
||
class RunReport:
|
||
cell_id: str
|
||
rq: str
|
||
run_index: int
|
||
seed: int
|
||
circuit_fingerprint: str
|
||
c_circuits: int
|
||
delivered: int
|
||
rq1_bridge_correlation_auc: float
|
||
rq2_anonymity_entropy_bits: float
|
||
rq2_sender_count: int
|
||
padding_applied: bool
|
||
span_houses: int
|
||
run_dir: str
|
||
per_circuit_seeds: List[int] = field(default_factory=list)
|
||
|
||
|
||
def _circuit_seed(run_seed: int, c_index: int) -> int:
|
||
"""Per-circuit seed: a domain-separated derivation of the run seed so each of
|
||
the C flows is a distinct, reproducible self-generated flow."""
|
||
blob = f"sor-circuit|{run_seed}|{c_index}".encode()
|
||
return int.from_bytes(hashlib.sha256(blob).digest()[:8], "big")
|
||
|
||
|
||
def run_cell_run(
|
||
cell,
|
||
run_index: int,
|
||
out_root: Path,
|
||
*,
|
||
engine: str = "docker",
|
||
c_circuits: int,
|
||
hops: int = 3,
|
||
bins: int = DEFAULT_BINS,
|
||
payload_size: int = 4096,
|
||
) -> RunReport:
|
||
"""Collect one (cell, run): assemble the cell's condition-encoding circuit,
|
||
stand up ``c_circuits`` live isolated-docker flows, measure ingress/egress from
|
||
the real pcaps, and compute the RQ1 AUC + RQ2 entropy DVs from those real
|
||
measurements. Writes a per-run manifest + metrics.json. Never fabricates."""
|
||
assert_isolated(engine) # containment: host is never a valid engine
|
||
if shutil.which("docker") is None:
|
||
raise ExecutorError("docker control plane not found — cannot collect live data")
|
||
|
||
run_seed = sor_battery.derive_seed(cell.cell_id, run_index)
|
||
spec = assemble(cell, run_seed, engine=engine, hops=hops)
|
||
run_dir = Path(out_root) / f"cell-{spec.fingerprint()[:12]}-r{run_index}"
|
||
run_dir.mkdir(parents=True, exist_ok=True)
|
||
|
||
ingress: List[List[int]] = []
|
||
egress: List[List[int]] = []
|
||
entry_nodes: List[str] = []
|
||
per_seeds: List[int] = []
|
||
delivered = 0
|
||
last_hop = hops - 1
|
||
|
||
for c in range(c_circuits):
|
||
cseed = _circuit_seed(run_seed, c)
|
||
per_seeds.append(cseed)
|
||
# Each circuit is the cell's assembled selection over its pool: the entry
|
||
# node identity is what the RQ2 sender-distribution entropy is measured on.
|
||
cspec = assemble(cell, cseed, engine=engine, hops=hops)
|
||
entry_nodes.append(cspec.hops[0].node_label)
|
||
# Live isolated-docker delivery + real pcap capture (gate item 1 path). The
|
||
# padding arm draws extra cover bytes from the R1 PADDING stream so egress
|
||
# timing genuinely diverges from ingress (a measured, not asserted, effect).
|
||
psize = payload_size + (_padding_bytes(cseed) if spec.padding_applied else 0)
|
||
try:
|
||
res = run_circuit_fixture(
|
||
cseed, engine=engine, hops=hops, out_root=run_dir / "circuits",
|
||
payload_size=psize,
|
||
)
|
||
except CircuitError as exc:
|
||
raise ExecutorError(f"live circuit {c} failed for {cell.cell_id}: {exc}") from exc
|
||
if not res.delivered:
|
||
raise ExecutorError(f"circuit {c} did not deliver for {cell.cell_id}")
|
||
delivered += 1
|
||
ing, eg = _measure_flow(res.pcaps, last_hop, bins)
|
||
ingress.append(ing)
|
||
egress.append(eg)
|
||
|
||
if delivered != c_circuits:
|
||
raise ExecutorError(
|
||
f"{cell.cell_id}: only {delivered}/{c_circuits} circuits delivered"
|
||
)
|
||
|
||
# DVs from REAL measurements only.
|
||
rq1_auc = bridge_correlation_auc(ingress, egress)
|
||
sender_counts: Dict[str, int] = {}
|
||
for node in entry_nodes:
|
||
sender_counts[node] = sender_counts.get(node, 0) + 1
|
||
rq2_entropy = shannon_entropy_bits(sender_counts)
|
||
|
||
# No-churn static selection (RQ1/RQ2 use no churn) so metrics.json validates.
|
||
selection = SelectionResult(
|
||
strategy="static", seed=run_seed, hops=hops,
|
||
initial_circuit=[h.node_label for h in spec.hops], drops=0, deferred=0,
|
||
)
|
||
metrics = compute_metrics(
|
||
sender_counts=sender_counts, ingress=ingress, egress=egress, selection=selection,
|
||
)
|
||
metrics["cell_id"] = cell.cell_id
|
||
metrics["rq"] = cell.rq
|
||
metrics["run_index"] = run_index
|
||
metrics["circuit_fingerprint"] = spec.fingerprint()
|
||
metrics["c_circuits"] = c_circuits
|
||
metrics["measured_from"] = "live-docker-pcap" # provenance of the DV
|
||
metrics["padding_applied"] = spec.padding_applied
|
||
metrics["span_houses"] = spec.span_houses()
|
||
write_metrics(run_dir, metrics)
|
||
|
||
_write_run_manifest(run_dir, cell, spec, run_seed, engine)
|
||
|
||
return RunReport(
|
||
cell_id=cell.cell_id, rq=cell.rq, run_index=run_index, seed=run_seed,
|
||
circuit_fingerprint=spec.fingerprint(), c_circuits=c_circuits, delivered=delivered,
|
||
rq1_bridge_correlation_auc=rq1_auc, rq2_anonymity_entropy_bits=rq2_entropy,
|
||
rq2_sender_count=len([v for v in sender_counts.values() if v > 0]),
|
||
padding_applied=spec.padding_applied, span_houses=spec.span_houses(),
|
||
run_dir=str(run_dir), per_circuit_seeds=per_seeds,
|
||
)
|
||
|
||
|
||
def _padding_bytes(seed: int) -> int:
|
||
"""Cover-traffic size drawn from the R1 PADDING stream (deterministic per
|
||
seed) — the on+padding arm's genuinely-injected extra bytes."""
|
||
return 512 + SorRng(seed).stream(Domain.PADDING).next_below(3584)
|
||
|
||
|
||
def _write_run_manifest(run_dir: Path, cell, spec, seed: int, engine: str) -> None:
|
||
"""Immutable R2 manifest binding this run to its assembled circuit fingerprint
|
||
and per-hop persona fingerprints. Write-once; skipped if already present."""
|
||
if (run_dir / "manifest.json").exists():
|
||
return
|
||
nodes = [Node(role=h.role, persona_pub_b64=_hop_pub(seed, h.node_label), engine=engine)
|
||
for h in spec.hops]
|
||
manifest = RunManifest(
|
||
run_id=f"conf-{spec.fingerprint()[:12]}-r{cell.rq}-{seed:016x}",
|
||
sor_seed=seed,
|
||
topology=cell.factors.get("topology", spec.topology),
|
||
selector=cell.factors.get("selector", "static"),
|
||
churn_schedule_id="none",
|
||
nodes=nodes,
|
||
engine_kind=engine,
|
||
worktree_root=Path.cwd(),
|
||
)
|
||
write_manifest(run_dir, manifest)
|
||
|
||
|
||
def _hop_pub(seed: int, label: str) -> str:
|
||
import base64
|
||
raw = hashlib.sha256(f"sor-hoppub|{seed}|{label}".encode()).digest()
|
||
return base64.b64encode(raw).decode()
|
||
|
||
|
||
# --------------------------------------------------------------------------- #
|
||
# The battery: schedule -> per-run collection -> per-cell aggregate.
|
||
# --------------------------------------------------------------------------- #
|
||
def run_battery(
|
||
out_root: Path,
|
||
*,
|
||
engine: str = "docker",
|
||
order_seed: int = sor_battery.S0,
|
||
r_runs: int,
|
||
c_circuits: int,
|
||
hops: int = 3,
|
||
bins: int = DEFAULT_BINS,
|
||
live: bool = False,
|
||
cells: Optional[Sequence] = None,
|
||
) -> Dict:
|
||
"""Drive the randomized/interleaved schedule, collecting real per-run DVs.
|
||
|
||
``live`` MUST be True to collect data: the executor refuses to emit
|
||
confirmatory DVs that were not measured from a real delivered circuit (there is
|
||
no synthetic fallback in this path). Returns an aggregate report and writes a
|
||
write-once ``battery-results.json`` under ``out_root``."""
|
||
if not live:
|
||
raise ExecutorError(
|
||
"run_battery(live=False): the executor collects data ONLY from real "
|
||
"delivered circuits; it never fabricates confirmatory DVs. Pass live=True "
|
||
"(operator-gated) to collect."
|
||
)
|
||
assert_isolated(engine)
|
||
out_root = Path(out_root)
|
||
out_root.mkdir(parents=True, exist_ok=True)
|
||
|
||
schedule = sor_battery.battery_schedule(order_seed, r=r_runs)
|
||
by_cell = {c.cell_id: c for c in (cells or sor_battery.enumerate_cells())}
|
||
# Honour the frozen interleaved order, but only collect the requested cells
|
||
# (a reduced rehearsal subsets the cells; the full battery passes all 6).
|
||
schedule = [pr for pr in schedule if pr.cell_id in by_cell]
|
||
|
||
reports: List[RunReport] = []
|
||
for pr in schedule:
|
||
cell = by_cell[pr.cell_id]
|
||
rep = run_cell_run(
|
||
cell, pr.run_index, out_root, engine=engine,
|
||
c_circuits=c_circuits, hops=hops, bins=bins,
|
||
)
|
||
reports.append(rep)
|
||
|
||
# Per-cell aggregate of the measured DV distributions (no CI/Holm here — that
|
||
# is the confirm.py reporting layer; this writes the raw measured rows).
|
||
agg: Dict[str, Dict] = {}
|
||
for rep in reports:
|
||
a = agg.setdefault(rep.cell_id, {
|
||
"rq": rep.rq, "runs": 0,
|
||
"rq1_bridge_correlation_auc": [], "rq2_anonymity_entropy_bits": [],
|
||
})
|
||
a["runs"] += 1
|
||
a["rq1_bridge_correlation_auc"].append(rep.rq1_bridge_correlation_auc)
|
||
a["rq2_anonymity_entropy_bits"].append(rep.rq2_anonymity_entropy_bits)
|
||
|
||
doc = {
|
||
"schema": "sor-battery-results/1",
|
||
"engine": engine,
|
||
"order_seed": order_seed,
|
||
"r_runs": r_runs,
|
||
"c_circuits": c_circuits,
|
||
"hops": hops,
|
||
"bins": bins,
|
||
"measured_from": "live-docker-pcap",
|
||
"n_runs": len(reports),
|
||
"cells": agg,
|
||
"runs": [vars(r) for r in reports],
|
||
}
|
||
path = out_root / "battery-results.json"
|
||
if not path.exists():
|
||
path.write_text(json.dumps(doc, indent=2, sort_keys=True) + "\n", encoding="utf-8")
|
||
doc["_results_path"] = str(path)
|
||
return doc
|