|
| 1 | +"""Phase 3b neural benchmark runner. |
| 2 | +
|
| 3 | +Reads benchmarks/neural_matrix.toml, runs each config, writes a CSV + markdown |
| 4 | +summary to benchmark-results-neural/<timestamp>/. |
| 5 | +
|
| 6 | +kind="neural" rows use hybrid_attack (neural prefix peeling + SAT suffix). |
| 7 | +kind="sat" rows use the Phase 1 pure-SAT attack (no checkpoint needed). |
| 8 | +
|
| 9 | +If a neural checkpoint is absent the row is recorded with |
| 10 | +status="SKIP_MISSING_CHECKPOINT" and the runner continues — this allows |
| 11 | +infrastructure validation (CI) without trained weights. |
| 12 | +
|
| 13 | +Reproduce full matrix (requires GPU + trained checkpoints): |
| 14 | + uv run python -m benchmarks.bench_neural |
| 15 | +""" |
| 16 | + |
| 17 | +from __future__ import annotations |
| 18 | + |
| 19 | +import csv |
| 20 | +import time |
| 21 | +import tomllib |
| 22 | +from datetime import datetime |
| 23 | +from pathlib import Path |
| 24 | + |
| 25 | +from keeloq._types import bits_to_int |
| 26 | +from keeloq.attack import EncodeFn, SolveFn |
| 27 | +from keeloq.attack import attack as sat_attack |
| 28 | +from keeloq.cipher import encrypt |
| 29 | +from keeloq.encoders.cnf import encode as encode_cnf |
| 30 | +from keeloq.encoders.xor_aware import encode as encode_xor |
| 31 | +from keeloq.solvers.cryptominisat import solve as solve_cms |
| 32 | +from keeloq.solvers.dimacs_subprocess import solve as solve_subprocess |
| 33 | + |
| 34 | +# --------------------------------------------------------------------------- |
| 35 | +# Fixed KAT (same key + plaintexts as bench_attack.py for comparability) |
| 36 | +# --------------------------------------------------------------------------- |
| 37 | +KEY_BITS = "0011010011011111100101100001110000011101100111001000001101110100" |
| 38 | +PT1_BITS = "01100010100101110000101011100011" |
| 39 | +PT2_BITS = "11010011100101010000111100001010" |
| 40 | +PT3_BITS = "10101010101010101010101010101010" |
| 41 | +PT4_BITS = "01010101010101010101010101010101" |
| 42 | +PT5_BITS = "00000000000000000000000000000001" |
| 43 | +PT6_BITS = "11111111111111111111111111111110" |
| 44 | +PT7_BITS = "10110011001100110011001100110011" |
| 45 | +PT8_BITS = "01001100110011001100110011001100" |
| 46 | + |
| 47 | +_PTS = [ |
| 48 | + PT1_BITS, |
| 49 | + PT2_BITS, |
| 50 | + PT3_BITS, |
| 51 | + PT4_BITS, |
| 52 | + PT5_BITS, |
| 53 | + PT6_BITS, |
| 54 | + PT7_BITS, |
| 55 | + PT8_BITS, |
| 56 | +] |
| 57 | + |
| 58 | + |
| 59 | +def _encoder(name: str) -> EncodeFn: |
| 60 | + return {"cnf": encode_cnf, "xor": encode_xor}[name] # type: ignore[return-value] |
| 61 | + |
| 62 | + |
| 63 | +def _solver(name: str) -> SolveFn: |
| 64 | + if name == "cryptominisat": |
| 65 | + return solve_cms |
| 66 | + |
| 67 | + def _wrap(inst: object, timeout_s: float) -> object: |
| 68 | + return solve_subprocess(inst, solver_binary=name, timeout_s=timeout_s) # type: ignore[arg-type] |
| 69 | + |
| 70 | + return _wrap # type: ignore[return-value] |
| 71 | + |
| 72 | + |
| 73 | +def _build_pairs( |
| 74 | + pts: list[int], |
| 75 | + key: int, |
| 76 | + rounds: int, |
| 77 | + n_pairs: int, |
| 78 | +) -> list[tuple[int, int]]: |
| 79 | + """Build genuine (plaintext, ciphertext) pairs from the fixed KAT key.""" |
| 80 | + if n_pairs > len(pts): |
| 81 | + raise ValueError(f"num_pairs={n_pairs} exceeds available fixtures ({len(pts)})") |
| 82 | + return [(p, encrypt(p, key, rounds)) for p in pts[:n_pairs]] |
| 83 | + |
| 84 | + |
| 85 | +def _build_diff_pairs( |
| 86 | + pts: list[int], |
| 87 | + key: int, |
| 88 | + rounds: int, |
| 89 | + n_pairs: int, |
| 90 | + delta: int, |
| 91 | +) -> list[tuple[int, int]]: |
| 92 | + """Build (c0, c1) differential pairs: c0=E(pt,key), c1=E(pt^delta,key).""" |
| 93 | + if n_pairs > len(pts): |
| 94 | + raise ValueError(f"num_pairs={n_pairs} exceeds available fixtures ({len(pts)})") |
| 95 | + result = [] |
| 96 | + for pt in pts[:n_pairs]: |
| 97 | + # CUDA uint32 XOR unsupported — apply delta on CPU side. |
| 98 | + pt1 = (pt ^ delta) & 0xFFFFFFFF |
| 99 | + c0 = encrypt(pt, key, rounds) |
| 100 | + c1 = encrypt(pt1, key, rounds) |
| 101 | + result.append((c0, c1)) |
| 102 | + return result |
| 103 | + |
| 104 | + |
| 105 | +def _run_sat(run: dict[str, object], key: int, pts: list[int]) -> dict[str, object]: |
| 106 | + """Execute a pure-SAT benchmark row.""" |
| 107 | + rounds = int(run["rounds"]) # type: ignore[arg-type] |
| 108 | + n_pairs = int(run["num_pairs"]) # type: ignore[arg-type] |
| 109 | + hint_bits = int(run.get("hint_bits", 0)) # type: ignore[arg-type] |
| 110 | + timeout_s = float(run.get("timeout_s", 300.0)) # type: ignore[arg-type] |
| 111 | + |
| 112 | + pairs = _build_pairs(pts, key, rounds, n_pairs) |
| 113 | + hints: dict[int, int] | None = ( |
| 114 | + {i: (key >> (63 - i)) & 1 for i in range(64 - hint_bits, 64)} if hint_bits > 0 else None |
| 115 | + ) |
| 116 | + |
| 117 | + t0 = time.perf_counter() |
| 118 | + result = sat_attack( |
| 119 | + rounds=rounds, |
| 120 | + pairs=pairs, |
| 121 | + key_hints=hints, |
| 122 | + encoder=_encoder(str(run.get("encoder", "xor"))), |
| 123 | + solver_fn=_solver(str(run.get("solver", "cryptominisat"))), |
| 124 | + timeout_s=timeout_s, |
| 125 | + ) |
| 126 | + wall = time.perf_counter() - t0 |
| 127 | + |
| 128 | + return { |
| 129 | + "name": run["name"], |
| 130 | + "kind": "sat", |
| 131 | + "rounds": rounds, |
| 132 | + "num_pairs": n_pairs, |
| 133 | + "checkpoint": "", |
| 134 | + "status": result.status, |
| 135 | + "wall_time_s": f"{wall:.3f}", |
| 136 | + "bits_recovered_neurally": 0, |
| 137 | + "neural_wall_time_s": "0.000", |
| 138 | + "sat_wall_time_s": f"{result.solve_result.stats.wall_time_s:.3f}", |
| 139 | + "num_vars": result.solve_result.stats.num_vars, |
| 140 | + "num_clauses": result.solve_result.stats.num_clauses, |
| 141 | + "num_xors": result.solve_result.stats.num_xors, |
| 142 | + } |
| 143 | + |
| 144 | + |
| 145 | +def _run_neural(run: dict[str, object], key: int, pts: list[int]) -> dict[str, object]: |
| 146 | + """Execute a neural-hybrid benchmark row. |
| 147 | +
|
| 148 | + Deviations from plan (per task brief): |
| 149 | + - Uses two-arg hybrid_attack: pairs (differential c0:c1) + sat_pairs (pt:ct). |
| 150 | + - Applies extra_key_hints for cyclic-schedule bits when rounds < 64. |
| 151 | + - If the checkpoint doesn't exist, records SKIP_MISSING_CHECKPOINT instead of crashing. |
| 152 | + """ |
| 153 | + from keeloq.neural.distinguisher import load_checkpoint |
| 154 | + from keeloq.neural.hybrid import hybrid_attack |
| 155 | + |
| 156 | + rounds = int(run["rounds"]) # type: ignore[arg-type] |
| 157 | + n_pairs = int(run["num_pairs"]) # type: ignore[arg-type] |
| 158 | + ckpt_path = Path(str(run["checkpoint"])) |
| 159 | + beam_width = int(run.get("beam_width", 8)) # type: ignore[arg-type] |
| 160 | + neural_target_bits = run.get("neural_target_bits") |
| 161 | + neural_target_bits_int: int | None = ( |
| 162 | + int(neural_target_bits) if neural_target_bits is not None else None # type: ignore[arg-type] |
| 163 | + ) |
| 164 | + sat_timeout_s = float(run.get("sat_timeout_s", 60.0)) # type: ignore[arg-type] |
| 165 | + max_backtracks = int(run.get("max_backtracks", 8)) # type: ignore[arg-type] |
| 166 | + |
| 167 | + # Smoke-safe path: skip gracefully when checkpoint is absent. |
| 168 | + if not ckpt_path.exists(): |
| 169 | + print(f" [SKIP] checkpoint not found: {ckpt_path}", flush=True) |
| 170 | + return { |
| 171 | + "name": run["name"], |
| 172 | + "kind": "neural", |
| 173 | + "rounds": rounds, |
| 174 | + "num_pairs": n_pairs, |
| 175 | + "checkpoint": str(ckpt_path), |
| 176 | + "status": "SKIP_MISSING_CHECKPOINT", |
| 177 | + "wall_time_s": "0.000", |
| 178 | + "bits_recovered_neurally": 0, |
| 179 | + "neural_wall_time_s": "0.000", |
| 180 | + "sat_wall_time_s": "0.000", |
| 181 | + "num_vars": 0, |
| 182 | + "num_clauses": 0, |
| 183 | + "num_xors": 0, |
| 184 | + } |
| 185 | + |
| 186 | + # Load distinguisher checkpoint to discover the trained delta. |
| 187 | + model, train_result = load_checkpoint(ckpt_path) |
| 188 | + delta = train_result.config.delta |
| 189 | + |
| 190 | + # Build differential pairs (c0, c1) for the neural distinguisher. |
| 191 | + diff_pairs = _build_diff_pairs(pts, key, rounds, n_pairs, delta) |
| 192 | + |
| 193 | + # Build genuine (plaintext, ciphertext) pairs for SAT. |
| 194 | + sat_pairs = _build_pairs(pts, key, rounds, n_pairs) |
| 195 | + |
| 196 | + # Apply extra_key_hints for cyclic-schedule bits when rounds < 64. |
| 197 | + extra_hints: dict[int, int] | None = None |
| 198 | + if rounds < 64: |
| 199 | + extra_hints = {i: (key >> (63 - i)) & 1 for i in range(rounds, 64)} |
| 200 | + |
| 201 | + t0 = time.perf_counter() |
| 202 | + result = hybrid_attack( |
| 203 | + rounds=rounds, |
| 204 | + pairs=diff_pairs, |
| 205 | + sat_pairs=sat_pairs, |
| 206 | + distinguisher=model, |
| 207 | + beam_width=beam_width, |
| 208 | + neural_target_bits=neural_target_bits_int, |
| 209 | + sat_timeout_s=sat_timeout_s, |
| 210 | + max_backtracks=max_backtracks, |
| 211 | + extra_key_hints=extra_hints, |
| 212 | + ) |
| 213 | + wall = time.perf_counter() - t0 |
| 214 | + |
| 215 | + return { |
| 216 | + "name": run["name"], |
| 217 | + "kind": "neural", |
| 218 | + "rounds": rounds, |
| 219 | + "num_pairs": n_pairs, |
| 220 | + "checkpoint": str(ckpt_path), |
| 221 | + "status": result.status, |
| 222 | + "wall_time_s": f"{wall:.3f}", |
| 223 | + "bits_recovered_neurally": result.bits_recovered_neurally, |
| 224 | + "neural_wall_time_s": f"{result.neural_wall_time_s:.3f}", |
| 225 | + "sat_wall_time_s": f"{result.sat_wall_time_s:.3f}", |
| 226 | + "num_vars": 0, |
| 227 | + "num_clauses": 0, |
| 228 | + "num_xors": 0, |
| 229 | + } |
| 230 | + |
| 231 | + |
| 232 | +_FIELDS = [ |
| 233 | + "name", |
| 234 | + "kind", |
| 235 | + "rounds", |
| 236 | + "num_pairs", |
| 237 | + "checkpoint", |
| 238 | + "status", |
| 239 | + "wall_time_s", |
| 240 | + "bits_recovered_neurally", |
| 241 | + "neural_wall_time_s", |
| 242 | + "sat_wall_time_s", |
| 243 | + "num_vars", |
| 244 | + "num_clauses", |
| 245 | + "num_xors", |
| 246 | +] |
| 247 | + |
| 248 | + |
| 249 | +def run_matrix(matrix_path: Path, out_dir: Path) -> Path: |
| 250 | + """Run every row in the neural matrix, write CSV + markdown. Returns out_dir.""" |
| 251 | + out_dir.mkdir(parents=True, exist_ok=True) |
| 252 | + config = tomllib.loads(matrix_path.read_text()) |
| 253 | + |
| 254 | + key = bits_to_int(KEY_BITS) |
| 255 | + pts = [bits_to_int(p) for p in _PTS] |
| 256 | + |
| 257 | + rows: list[dict[str, object]] = [] |
| 258 | + for run in config["run"]: |
| 259 | + kind = str(run.get("kind", "sat")) |
| 260 | + print(f"[bench_neural] running {run['name']!r} (kind={kind})...", flush=True) |
| 261 | + |
| 262 | + try: |
| 263 | + row = _run_neural(run, key, pts) if kind == "neural" else _run_sat(run, key, pts) |
| 264 | + except Exception as exc: |
| 265 | + print(f" [ERROR] {exc}", flush=True) |
| 266 | + row = { |
| 267 | + "name": run["name"], |
| 268 | + "kind": kind, |
| 269 | + "rounds": run.get("rounds", 0), |
| 270 | + "num_pairs": run.get("num_pairs", 0), |
| 271 | + "checkpoint": str(run.get("checkpoint", "")), |
| 272 | + "status": f"CRASH: {exc}", |
| 273 | + "wall_time_s": "0.000", |
| 274 | + "bits_recovered_neurally": 0, |
| 275 | + "neural_wall_time_s": "0.000", |
| 276 | + "sat_wall_time_s": "0.000", |
| 277 | + "num_vars": 0, |
| 278 | + "num_clauses": 0, |
| 279 | + "num_xors": 0, |
| 280 | + } |
| 281 | + |
| 282 | + print(f" -> status={row['status']} wall_time_s={row['wall_time_s']}", flush=True) |
| 283 | + rows.append(row) |
| 284 | + |
| 285 | + # Write CSV |
| 286 | + csv_path = out_dir / "results.csv" |
| 287 | + with csv_path.open("w", newline="") as f: |
| 288 | + w = csv.DictWriter(f, fieldnames=_FIELDS) |
| 289 | + w.writeheader() |
| 290 | + w.writerows(rows) |
| 291 | + |
| 292 | + # Write markdown summary |
| 293 | + md_path = out_dir / "summary.md" |
| 294 | + with md_path.open("w") as f: |
| 295 | + f.write("# Phase 3b Neural Benchmark Results\n\n") |
| 296 | + f.write( |
| 297 | + "> Full matrix requires trained checkpoints. " |
| 298 | + "Reproduce: `uv run python -m benchmarks.bench_neural`\n\n" |
| 299 | + ) |
| 300 | + f.write("| " + " | ".join(_FIELDS) + " |\n") |
| 301 | + f.write("|" + "|".join(["---"] * len(_FIELDS)) + "|\n") |
| 302 | + for r in rows: |
| 303 | + f.write("| " + " | ".join(str(r[k]) for k in _FIELDS) + " |\n") |
| 304 | + |
| 305 | + print(f"[bench_neural] wrote {csv_path} and {md_path}", flush=True) |
| 306 | + return out_dir |
| 307 | + |
| 308 | + |
| 309 | +def main() -> None: |
| 310 | + matrix = Path(__file__).parent / "neural_matrix.toml" |
| 311 | + ts = datetime.now().strftime("%Y%m%d-%H%M%S") |
| 312 | + out = Path("benchmark-results-neural") / ts |
| 313 | + run_matrix(matrix, out) |
| 314 | + |
| 315 | + |
| 316 | +if __name__ == "__main__": |
| 317 | + main() |
0 commit comments