#!/usr/bin/env python3 """Trace the gated path after the nearest selectionBuffer[0x20] writer.""" from __future__ import annotations import argparse import html import json import struct import sys from pathlib import Path from typing import Any sys.path.insert(0, str(Path(__file__).resolve().parent)) from probe_exe_scene_tables import find_cns_strings, read_sections, va_to_offset from summarize_script_handler_table import handler_for_opcode from summarize_save_selector_stream_traces import trace_stream ROOT = Path(__file__).resolve().parents[1] OUT = ROOT / "out" def load_json(path: Path, fallback: Any) -> Any: if not path.exists(): return fallback return json.loads(path.read_text(encoding="utf-8")) def parse_hex(value: str | None) -> int | None: return int(value, 16) if value else None def dword_at(exe: bytes, sections: list[dict], va: int) -> int | None: offset = va_to_offset(sections, va) if offset is None or offset + 4 > len(exe): return None return struct.unpack_from(" dict: value = dword_at(exe, sections, va) row = { "vaHex": f"0x{va:08x}", "valueHex": f"0x{value:08x}" if value is not None else None, } if value is None: row["kind"] = "unreadable" elif value in strings: row["kind"] = "cns" row["cns"] = strings[value] elif va_to_offset(sections, value) is not None: row["kind"] = "pointer" row["targetHex"] = f"0x{value:08x}" else: row["kind"] = "scalar" return row def trace_start(exe: bytes, sections: list[dict], strings: dict[int, str], start: int) -> list[dict]: rows = [] for row in trace_stream(exe, sections, strings, start, 20): rows.append({ "step": row.get("step"), "vaHex": row.get("vaHex"), "valueHex": row.get("valueHex"), "opcodeHex": row.get("opcodeHex"), "handlerVaHex": row.get("handlerVaHex"), "branchTargetHex": row.get("branchTargetHex"), "fallthroughVaHex": row.get("fallthroughVaHex"), "stopReason": row.get("stopReason"), "branchNote": row.get("branchNote"), }) return rows def build_rows(exe: bytes, writer_chain_rows: list[dict], leaf_streams: list[dict]) -> list[dict]: sections = read_sections(exe) strings = find_cns_strings(exe, sections) rows = [] for chain in writer_chain_rows: nearest = chain.get("nearestLinearWriter") or {} writer_va = parse_hex(nearest.get("writerVaHex")) reader_va = parse_hex(chain.get("frontierReaderVaHex")) if writer_va is None or reader_va is None: continue first_gate = writer_va + 8 gate_trace = trace_start(exe, sections, strings, first_gate) second_gate = parse_hex(gate_trace[0].get("fallthroughVaHex")) if gate_trace else None second_gate_trace = trace_start(exe, sections, strings, second_gate) if second_gate else [] post_gate = parse_hex(second_gate_trace[0].get("fallthroughVaHex")) if second_gate_trace else None post_gate_trace = trace_start(exe, sections, strings, post_gate) if post_gate else [] stop = post_gate_trace[-1] if post_gate_trace else {} stop_va = parse_hex(stop.get("vaHex")) stop_opcode = int((stop.get("opcodeHex") or "0x00"), 16) stop_handler = handler_for_opcode(exe, sections, stop_opcode) opcode24 = next((item for item in post_gate_trace if item.get("opcodeHex") == "0x24"), {}) opcode24_va = parse_hex(opcode24.get("vaHex")) opcode24_next = pointer_row(exe, sections, strings, opcode24_va + 4) if opcode24_va else {} opcode24_next_opcode = int((opcode24_next.get("valueHex") or "0x00000000")[-2:], 16) opcode24_next_handler = handler_for_opcode(exe, sections, opcode24_next_opcode) dispatch_context = [ pointer_row(exe, sections, strings, va) for va in range((stop_va or 0), (stop_va or 0) + 0x28, 4) ] if stop_va else [] frontier_leaf = next( ( stream for stream in leaf_streams if stream.get("leafPointerHex") == "0x00542ae8" and stream.get("source") == chain.get("source") and stream.get("target") == chain.get("target") ), {}, ) rows.append({ "source": chain.get("source"), "target": chain.get("target"), "nearestWriterVaHex": nearest.get("writerVaHex"), "frontierReaderVaHex": chain.get("frontierReaderVaHex"), "firstGateTrace": gate_trace, "secondGateTrace": second_gate_trace, "postGateTrace": post_gate_trace, "dispatchStopVaHex": stop.get("vaHex"), "dispatchStopOpcodeHex": stop.get("opcodeHex"), "dispatchStopHandlerVaHex": stop.get("handlerVaHex"), "dispatchStopHandlerSection": stop_handler.get("handlerSection"), "dispatchStopIsCodeHandler": stop_handler.get("isCodeHandler"), "dispatchContext": dispatch_context, "opcode24Boundary": { "opcodeVaHex": opcode24.get("vaHex"), "opcodeValueHex": opcode24.get("valueHex"), "handlerVaHex": opcode24.get("handlerVaHex"), "streamPlus1Hex": "0x01", "meaning": ( "Opcode 0x24 is an action/state handler. It reads stream+1 as a mode, checks " "runtime globals 0x0059e34d/0x0059e33e and object state fields, writes context+0x58 " "in several branches, and only has a simple +4 advance in the statically detected path." ), "nextDword": opcode24_next, "nextLowByteOpcodeHex": f"0x{opcode24_next_opcode:02x}", "nextLowByteHandlerVaHex": opcode24_next_handler.get("handlerVaHex"), "nextLowByteHandlerSection": opcode24_next_handler.get("handlerSection"), "nextLowByteIsCodeHandler": opcode24_next_handler.get("isCodeHandler"), }, "frontierLeafPointerHex": frontier_leaf.get("leafPointerHex"), "frontierLeafFieldRecords": frontier_leaf.get("fieldRecords") or [], "conclusion": ( "The nearest writer 0x005428bc does not linearly prove the frontier reader. " "It falls through two 0x11 gates with scalar branch targets 0x00000001/0x00000002, " "then reaches opcode 0x24 at 0x005428e4. Opcode 0x24 is a runtime action/state handler, " "so the bytes after 0x005428e4 are not proven linear bytecode by the simple stream tracer. " "The naive linear trace later reaches low byte 0xe8 at 0x005428f4, but its handler-table entry " "0x00440c28 is in .data rather than code. The next proof step is decoding that action/state " "payload boundary and showing it selects " "leaf 0x00542ae8 before the 0x00542b0c reader." ), }) return rows def markdown(rows: list[dict]) -> str: lines = [ "# Save Selector Gate Paths", "", "Control-flow boundary after the nearest `selectionBuffer[0x20]` writer.", "", ] for row in rows: lines.extend([ f"## {row['source']} -> {row['target']}", "", f"- nearest writer: `{row.get('nearestWriterVaHex')}`", f"- frontier reader: `{row.get('frontierReaderVaHex')}`", f"- dispatch stop: `{row.get('dispatchStopVaHex')}` opcode `{row.get('dispatchStopOpcodeHex')}` handler `{row.get('dispatchStopHandlerVaHex')}` in {row.get('dispatchStopHandlerSection')}", f"- frontier leaf candidate: `{row.get('frontierLeafPointerHex')}`", f"- opcode 0x24 boundary: `{(row.get('opcode24Boundary') or {}).get('opcodeVaHex')}` handler `{(row.get('opcode24Boundary') or {}).get('handlerVaHex')}`", f"- next low-byte handler: `{(row.get('opcode24Boundary') or {}).get('nextLowByteOpcodeHex')}` -> `{(row.get('opcode24Boundary') or {}).get('nextLowByteHandlerVaHex')}` in {(row.get('opcode24Boundary') or {}).get('nextLowByteHandlerSection')}", f"- conclusion: {row.get('conclusion')}", "", "| segment | step | va | value | opcode | handler | branch target | fallthrough | stop |", "| --- | ---: | --- | --- | --- | --- | --- | --- | --- |", ]) for segment, trace in [ ("first gate", row.get("firstGateTrace") or []), ("second gate", row.get("secondGateTrace") or []), ("post gate", row.get("postGateTrace") or []), ]: for item in trace: lines.append( f"| {segment} | {item.get('step')} | `{item.get('vaHex')}` | `{item.get('valueHex')}` | " f"`{item.get('opcodeHex')}` | `{item.get('handlerVaHex')}` | " f"`{item.get('branchTargetHex') or '-'}` | `{item.get('fallthroughVaHex') or '-'}` | " f"{item.get('stopReason') or '-'} |" ) lines.extend(["", "### Dispatch Context", "", "| va | value | kind | target |", "| --- | --- | --- | --- |"]) for item in row.get("dispatchContext") or []: target = item.get("targetHex") or item.get("cns") or "-" lines.append(f"| `{item.get('vaHex')}` | `{item.get('valueHex')}` | {item.get('kind')} | {target} |") boundary = row.get("opcode24Boundary") or {} lines.extend([ "", "### Opcode 0x24 Boundary", "", boundary.get("meaning") or "-", "", f"- next dword: `{(boundary.get('nextDword') or {}).get('valueHex')}` at `{(boundary.get('nextDword') or {}).get('vaHex')}`", f"- next low-byte handler: `{boundary.get('nextLowByteOpcodeHex')}` -> `{boundary.get('nextLowByteHandlerVaHex')}` in {boundary.get('nextLowByteHandlerSection')}", ]) lines.append("") return "\n".join(lines) def html_page(rows: list[dict]) -> str: parts = [ "Save Selector Gate Paths", "", "

Save Selector Gate Paths

", "

Control-flow boundary after the nearest selectionBuffer[0x20] writer.

", ] for row in rows: parts.extend([ f"

{html.escape(str(row.get('source')))} -> {html.escape(str(row.get('target')))}

", "", "", ]) for segment, trace in [ ("first gate", row.get("firstGateTrace") or []), ("second gate", row.get("secondGateTrace") or []), ("post gate", row.get("postGateTrace") or []), ]: for item in trace: parts.append( "" f"" f"" f"" f"" f"" f"" f"" f"" f"" "" ) parts.append("
segmentstepvavalueopcodehandlerbranch targetfallthroughstop
{html.escape(segment)}{item.get('step')}{html.escape(str(item.get('vaHex')))}{html.escape(str(item.get('valueHex')))}{html.escape(str(item.get('opcodeHex')))}{html.escape(str(item.get('handlerVaHex')))}{html.escape(str(item.get('branchTargetHex') or '-'))}{html.escape(str(item.get('fallthroughVaHex') or '-'))}{html.escape(str(item.get('stopReason') or '-'))}

Dispatch Context

") for item in row.get("dispatchContext") or []: target = item.get("targetHex") or item.get("cns") or "-" parts.append( "" f"" f"" f"" f"" "" ) boundary = row.get("opcode24Boundary") or {} parts.append("
vavaluekindtarget
{html.escape(str(item.get('vaHex')))}{html.escape(str(item.get('valueHex')))}{html.escape(str(item.get('kind')))}{html.escape(str(target))}
") parts.append("

Opcode 0x24 Boundary

") parts.append(f"

{html.escape(str(boundary.get('meaning') or '-'))}

") parts.append( f"

next dword: {html.escape(str((boundary.get('nextDword') or {}).get('valueHex')))} " f"at {html.escape(str((boundary.get('nextDword') or {}).get('vaHex')))}; " f"next low-byte handler: {html.escape(str(boundary.get('nextLowByteOpcodeHex')))} " f"-> {html.escape(str(boundary.get('nextLowByteHandlerVaHex')))} " f"in {html.escape(str(boundary.get('nextLowByteHandlerSection')))}

" ) return "\n".join(parts) def write_outputs(rows: list[dict], out_dir: Path) -> None: out_dir.mkdir(parents=True, exist_ok=True) (out_dir / "save_selector_gate_paths.json").write_text( json.dumps(rows, ensure_ascii=False, indent=2) + "\n", encoding="utf-8", ) def main() -> None: parser = argparse.ArgumentParser() parser.add_argument("--exe", type=Path, default=ROOT / "Hwanse2.exe") parser.add_argument("--out-dir", type=Path, default=OUT) args = parser.parse_args() rows = build_rows( args.exe.read_bytes(), load_json(args.out_dir / "save_selector_writer_chain.json", []), load_json(args.out_dir / "save_selector_leaf_streams.json", []), ) write_outputs(rows, args.out_dir) print(f"wrote {len(rows)} save selector gate path rows -> {args.out_dir / 'save_selector_gate_paths.json'}") if __name__ == "__main__": main()