#!/usr/bin/env python3 """Host-side driver for the dwt_prf challenge-transform oracle. Generates a deterministic vector set, runs the firmware's transform twice (two independent fresh-NVRAM boots) through dwt/oracle.lua, checks the results against the live TCP captures in dwt/evidence/run-live/, writes dwt/challenge_vectors.json and a characterisation report. Usage: python3 dwt/oracle.py # full run (two boots), ~2-4 minutes python3 dwt/oracle.py --quick # one boot, used while developing python3 dwt/oracle.py --no-run # regenerate inputs and JSON from the # outputs already in dwt/evidence/oracle/ Conventions (important for the reverser): input_hex = the 16-byte buffer exactly as the transform sees it, i.e. the 16 bytes received from the wire with each 32-bit word byte-swapped (dwt_prf's ntohl step, ROM 0x3FB2C140). output_hex = the in-place transformed buffer that dwt_prf memcmp's against the host-order challenge words. """ from __future__ import annotations import argparse import json import os import random import shutil import subprocess import sys import time from pathlib import Path PROJECT = Path(__file__).resolve().parent.parent MAME = os.environ.get("VT1200_MAME", "/home/arti/code/k-space/mame0289/vt1200") ROMPATH = PROJECT / "harness/mame/roms" EVID = PROJECT / "dwt/evidence/oracle" LUA = PROJECT / "dwt/oracle.lua" STAGE_LUA = PROJECT / "dwt/evidence/scripts/oracle_stage.lua" VECTORS_JSON = PROJECT / "dwt/challenge_vectors.json" LIVE_PROBES_PATH = PROJECT / "dwt/evidence/run-live/render-probes.txt" START_FRAME = int(os.environ.get("VT1200_ORACLE_START", "3000")) # -------------------------------------------------------------------------- # vector generation # -------------------------------------------------------------------------- def swap_words(b: bytes) -> bytes: """Byte-swap each 32-bit word (the firmware's ntohl, applied in place).""" out = bytearray() for i in range(0, len(b), 4): out += b[i:i + 4][::-1] return bytes(out) # The 16-byte challenge a live terminal sent over TCP (dwt/evidence/ # run-live/remote-attempt.txt, dwt/notes/EXPLOIT_ATTEMPT.md section 2). LIVE_WIRE_CHALLENGE = bytes.fromhex( "f32cec8d0000ead00263a8c0f324ec8e") # (name, wire reply bytes, expected first 8 transformed bytes) from # dwt/evidence/run-live/render-probes.txt. The transform input is the # wire reply with each word byte-swapped. LIVE_PROBES = [ ("live_zero", bytes(16), "90de09b472855817"), ("live_p0_01", bytes.fromhex("01000000") + bytes(12), "b1975ff1d78d032a"), ("live_p0_02", bytes.fromhex("02000000") + bytes(12), "5865800e5d8c3c55"), ("live_p0_03", bytes.fromhex("03000000") + bytes(12), "ee590df03c2bdbdd"), ("live_p0_10", bytes.fromhex("10000000") + bytes(12), "0b9ab22e400a068f"), ("live_p0_20", bytes.fromhex("20000000") + bytes(12), "362cf798ec8ae918"), ("live_p0_30", bytes.fromhex("30000000") + bytes(12), "9ecae2e539a16357"), ("live_p0_80", bytes.fromhex("80000000") + bytes(12), "57d8297c38f7783b"), ("live_p0_ff", bytes.fromhex("ff000000") + bytes(12), "fc00338c744b2d1b"), ("live_p1_01", bytes.fromhex("00010000") + bytes(12), "21105daed5e13ec5"), ("live_p7_01", bytes.fromhex("0000000000000001") + bytes(8), "46ba798d00e10cdb"), ] def build_vectors(): """Return [(hex_input, kind, note_or_None), ...] deterministically.""" items = [] def add(b: bytes, kind: str, note: str | None = None): assert len(b) == 16, (kind, len(b)) items.append((b.hex(), kind, note)) # --- structured specials ------------------------------------------- add(bytes(16), "zero") add(b"\xff" * 16, "ones") add(bytes(range(16)), "ascending") add(bytes(range(15, -1, -1)), "descending") add(bytes([0xAA, 0x55] * 8), "alternating_aa55") add(bytes([0x55, 0xAA] * 8), "alternating_55aa") add(bytes.fromhex("00" * 8 + "ff" * 8), "half_zero_half_ones") add(bytes.fromhex("ff" * 8 + "00" * 8), "half_ones_half_zero") add(bytes.fromhex("8000000000000000" + "00" * 8), "single_msb_first_half") add(bytes.fromhex("00" * 8 + "8000000000000000"), "single_msb_second_half") add(bytes.fromhex("7f" + "00" * 15), "byte0_7f") add(bytes.fromhex("00" * 8 + "01" + "00" * 7), "last_byte_01") add(bytes.fromhex("0011223344556677") * 2, "repeated_8byte_record") # --- the live challenge, in both orders ---------------------------- add(LIVE_WIRE_CHALLENGE, "live_wire_challenge", "the 16 bytes the terminal sent over TCP (wire order)") add(swap_words(LIVE_WIRE_CHALLENGE), "live_challenge_post_ntohl", "same challenge after dwt_prf's per-word ntohl; this is f's input") # --- live probe corpus (validates the direct call) ----------------- for name, wire, _expected in LIVE_PROBES: add(swap_words(wire), name, "live TCP probe, post-ntohl input") # --- single-bit flips of a baseline -------------------------------- base = bytes(range(16)) for byte_i in range(8): for bit in range(8): v = bytearray(base) v[byte_i] ^= 1 << bit add(bytes(v), "flip_first8", f"ascending baseline, byte {byte_i} bit {bit}") for byte_i in range(8, 16): v = bytearray(base) v[byte_i] ^= 0x01 add(bytes(v), "flip_second8", f"ascending baseline, byte {byte_i} bit 0") # --- per-record independence grid ---------------------------------- rng = random.Random(0x771) halves_a = [bytes(rng.randrange(256) for _ in range(8)) for _ in range(8)] halves_b = [bytes(rng.randrange(256) for _ in range(8)) for _ in range(8)] for h in halves_a: add(h + bytes(8), "indep_a_zero") add(h + b"\xff" * 8, "indep_a_ones") for h in halves_b: add(bytes(8) + h, "indep_b_zero") add(b"\xff" * 8 + h, "indep_b_ones") # --- random inputs -------------------------------------------------- for _ in range(48): add(bytes(rng.randrange(256) for _ in range(16)), "random") # --- two deliberate duplicates (in-run determinism) ----------------- add(bytes(16), "zero_duplicate", "repeat of the first vector") add(bytes(range(16)), "ascending_duplicate", "repeat of ascending") return items # -------------------------------------------------------------------------- # emulator run # -------------------------------------------------------------------------- def run_once(tag: str, inputs: list[str]) -> tuple[dict[str, str], Path]: run_dir = EVID / tag if run_dir.exists(): shutil.rmtree(run_dir) (run_dir / "nvram").mkdir(parents=True) in_path = run_dir / "in.txt" out_path = run_dir / "out.txt" in_path.write_text("\n".join(inputs) + "\n") env = dict(os.environ) env.update({ "VT1200_SKIP_LEVEL1": "1", "VT1200_ORACLE_IN": str(in_path), "VT1200_ORACLE_OUT": str(out_path), "VT1200_ORACLE_START": str(START_FRAME), "VT1200_ORACLE_TAG": tag, }) cmd = [ MAME, "vt1200", "-rompath", str(ROMPATH), "-video", "none", "-sound", "none", "-nothrottle", "-debug", "-debugger", "none", "-debuglog", "-nvram_directory", str(run_dir / "nvram"), "-autoboot_script", str(LUA), "-autoboot_delay", "0", ] t0 = time.time() with open(run_dir / "stdout.log", "w") as log: proc = subprocess.run(cmd, cwd=run_dir, env=env, stdout=log, stderr=subprocess.STDOUT, timeout=1800) dt = time.time() - t0 print(f"[{tag}] mame exit={proc.returncode} wall={dt:.1f}s") if not out_path.exists(): raise SystemExit(f"[{tag}] no output file; see {run_dir}/stdout.log") result = {} for line in out_path.read_text().splitlines(): parts = line.split() if len(parts) == 3: result[parts[1].lower()] = parts[2].lower() print(f"[{tag}] {len(result)} outputs") return result, run_dir # -------------------------------------------------------------------------- # intermediate-state capture (one 8-byte record at a time) # -------------------------------------------------------------------------- STAGE_RECORDS = [ "0000000000000000", "ffffffffffffffff", "0001020304050607", "8dec2cf3d0ea0000", "c0a863028eec24f3", "0011223344556677", ] def run_stages() -> Path | None: """Run the per-record stage capture; returns the combined output path.""" run_dir = EVID / "stages" if run_dir.exists(): shutil.rmtree(run_dir) (run_dir / "nvram").mkdir(parents=True) in_path = run_dir / "in.txt" out_path = run_dir / "out.txt" in_path.write_text("\n".join(STAGE_RECORDS) + "\n") env = dict(os.environ) env.update({ "VT1200_SKIP_LEVEL1": "1", "VT1200_ORACLE_IN": str(in_path), "VT1200_ORACLE_OUT": str(out_path), "VT1200_ORACLE_START": str(START_FRAME), "VT1200_ORACLE_TAG": "stages", }) cmd = [ MAME, "vt1200", "-rompath", str(ROMPATH), "-video", "none", "-sound", "none", "-nothrottle", "-debug", "-debugger", "none", "-debuglog", "-nvram_directory", str(run_dir / "nvram"), "-autoboot_script", str(STAGE_LUA), "-autoboot_delay", "0", ] with open(run_dir / "stdout.log", "w") as log: subprocess.run(cmd, cwd=run_dir, env=env, stdout=log, stderr=subprocess.STDOUT, timeout=900) if not out_path.exists(): print(f"[stages] no output; see {run_dir}/stdout.log") return None combined = EVID / "stages.txt" combined.write_text(out_path.read_text()) print(f"[stages] wrote {combined}") return combined # -------------------------------------------------------------------------- # checks and characterisation # -------------------------------------------------------------------------- def hamming(a: bytes, b: bytes) -> int: return sum(bin(x ^ y).count("1") for x, y in zip(a, b)) def main(argv): ap = argparse.ArgumentParser() ap.add_argument("--quick", action="store_true", help="one boot only") ap.add_argument("--no-run", action="store_true", help="reuse dwt/evidence/oracle/run*/out.txt") ap.add_argument("--skip-stages", action="store_true", help="do not run the intermediate-state capture") args = ap.parse_args(argv[1:]) EVID.mkdir(parents=True, exist_ok=True) items = build_vectors() inputs = [h for h, _k, _n in items] print(f"{len(items)} vectors") if args.no_run: runs = {} for tag in ("runA", "runB"): p = EVID / tag / "out.txt" if not p.exists(): raise SystemExit(f"missing {p}") runs[tag] = {} for line in p.read_text().splitlines(): parts = line.split() if len(parts) == 3: runs[tag][parts[1].lower()] = parts[2].lower() run_dirs = {"runA": EVID / "runA", "runB": EVID / "runB"} else: runA, dirA = run_once("runA", inputs) run_dirs = {"runA": dirA} if args.quick: runs = {"runA": runA} else: runB, dirB = run_once("runB", inputs) run_dirs["runB"] = dirB runs = {"runA": runA, "runB": runB} # Assemble vectors; an input is only emitted once both boots agree. vectors = [] missing = [] for i, (h, kind, note) in enumerate(items): outs = [r.get(h) for r in runs.values()] if any(o is None for o in outs): missing.append((i, h, kind)) continue if len(set(outs)) != 1: print(f"MISMATCH boot outputs for {h}: {outs}") continue meta = {"stage": "full", "boot": len(runs), "kind": kind, "index": i} if note: meta["note"] = note vectors.append({"input_hex": h, "output_hex": outs[0], "meta": meta}) if missing: print(f"WARNING: {len(missing)} vectors missing from a run: {missing[:5]}") VECTORS_JSON.write_text(json.dumps(vectors, indent=1) + "\n") print(f"wrote {VECTORS_JSON} ({len(vectors)} vectors)") # ---------------- intermediate states ------------------------------ stage_path = None if not args.no_run and not args.skip_stages: stage_path = run_stages() if stage_path is None: stage_path = EVID / "stages.txt" if (EVID / "stages.txt").exists() else None # ---------------- cross-check against the live TCP probes ----------- by_input = {v["input_hex"]: v["output_hex"] for v in vectors} live_ok, live_bad = 0, [] for name, wire, expected in LIVE_PROBES: h = swap_words(wire).hex() got = by_input.get(h) if got is not None and got[:16] == expected: live_ok += 1 else: live_bad.append((name, h, expected, got)) print(f"live-probe agreement: {live_ok}/{len(LIVE_PROBES)}") for row in live_bad: print(" MISMATCH", row) # ---------------- characterisation --------------------------------- report = {} # determinism across boots common = set(runs["runA"]) & set(runs.get("runB", runs["runA"])) diffs = [h for h in common if len(runs) > 1 and runs["runA"][h] != runs["runB"][h]] report["determinism"] = { "boots": len(runs), "inputs_compared": len(common), "differences": len(diffs), "first_differences": diffs[:5], } # per-record independence from the dedicated grid def out(h): return by_input.get(h) indep_errors = [] for h in [v["input_hex"] for v in vectors if v["meta"]["kind"] == "indep_a_zero"]: first = bytes.fromhex(h)[:8] o_zero = out((first + bytes(8)).hex()) o_ones = out((first + b"\xff" * 8).hex()) if o_zero and o_ones and o_zero[:16] != o_ones[:16]: indep_errors.append(("a", h)) for h in [v["input_hex"] for v in vectors if v["meta"]["kind"] == "indep_b_zero"]: second = bytes.fromhex(h)[8:] o_zero = out((bytes(8) + second).hex()) o_ones = out((b"\xff" * 8 + second).hex()) if o_zero and o_ones and o_zero[16:] != o_ones[16:]: indep_errors.append(("b", h)) report["per_record_independence"] = { "errors": len(indep_errors), "examples": indep_errors[:5], } # permutation / duplicates / fixed points distinct_inputs = {} for v in vectors: if v["meta"]["kind"].startswith(("random", "flip", "indep")): distinct_inputs.setdefault(v["input_hex"], v["output_hex"]) out_vals = list(distinct_inputs.values()) report["permutation"] = { "distinct_inputs": len(distinct_inputs), "distinct_outputs": len(set(out_vals)), "fixed_points": [v["input_hex"] for v in vectors if v["input_hex"] == v["output_hex"]], } # avalanche: single-bit flips of the ascending baseline base = bytes(range(16)) base_out = out(base.hex()) aval = [] record_leaks = 0 if base_out: base_out_b = bytes.fromhex(base_out) for v in vectors: if v["meta"]["kind"] != "flip_first8": continue ob = bytes.fromhex(v["output_hex"]) aval.append(hamming(ob[:8], base_out_b[:8])) if ob[8:] != base_out_b[8:]: record_leaks += 1 report["avalanche"] = { "samples": len(aval), "mean_output_bits_changed_first_record": ( round(sum(aval) / len(aval), 3) if aval else None), "min": min(aval) if aval else None, "max": max(aval) if aval else None, "second_record_changed": record_leaks, } # first/second half outputs are identical for a doubled record doubled = out((bytes.fromhex("0011223344556677") * 2).hex()) report["record_is_repeated"] = bool( doubled and doubled[:16] == doubled[16:]) # intermediate states: with dwt_prf's INIT(1) the two "font render" calls # are memcpy (their tables @1C0220/@1C0240 are NULL), so feistel_in == # input and pre_final_render == output. if stage_path and stage_path.exists(): rows = [] for line in stage_path.read_text().splitlines(): p = line.split() if len(p) == 5: rows.append(p) report["intermediate"] = { "captured": len(rows), "feistel_in_equals_input": all(r[2] == r[1] for r in rows), "pre_final_render_equals_output": all(r[3] == r[4] for r in rows), "file": str(stage_path.relative_to(PROJECT)), } # direction / convention notes are static; record the evidence sources report["direction"] = { "wire_challenge": LIVE_WIRE_CHALLENGE.hex(), "challenge_post_ntohl": swap_words(LIVE_WIRE_CHALLENGE).hex(), "before_transform": "buffer = received 16 bytes with each 32-bit word byte-swapped (dwt_prf ntohl, in place)", "after_transform": "buffer = f(before) in place; dwt_prf memcmp's it against the host-order challenge words", "pass_condition": "f(ntohl(reply_on_wire)) == host_order_challenge; reply = ntohl(f^-1(host_order_challenge))", } report["live_probe_agreement"] = { "matched": live_ok, "total": len(LIVE_PROBES), "failures": live_bad, } report["vector_count"] = len(vectors) (EVID / "characterisation.json").write_text(json.dumps(report, indent=2) + "\n") print(json.dumps(report, indent=2)) return 0 if (not live_bad and not diffs and not missing) else 1 if __name__ == "__main__": sys.exit(main(sys.argv))