mirror of
https://git.k-space.ee/arti/vt1200.git
synced 2026-10-30 12:03:54 +02:00
448 lines
17 KiB
Python
448 lines
17 KiB
Python
#!/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))
|