#!/usr/bin/env python3 """Summarize selection-buffer use around save-selector frontier branches.""" from __future__ import annotations import argparse import html import json from pathlib import Path ROOT = Path(__file__).resolve().parents[1] OUT = ROOT / "out" def byte_at_word(value_hex: str, index: int) -> int: value = int(value_hex, 16) return (value >> (index * 8)) & 0xFF def stream_fields(step: dict) -> dict: value_hex = step.get("valueHex") or "0x00000000" return { "opcodeHex": step.get("opcodeHex"), "valueHex": value_hex, "streamPlus1": byte_at_word(value_hex, 1), "streamPlus1Hex": f"0x{byte_at_word(value_hex, 1):02x}", "streamPlus2": byte_at_word(value_hex, 2), "streamPlus2Hex": f"0x{byte_at_word(value_hex, 2):02x}", } def build_rows(frontier_branches: list[dict], stream_traces: list[dict]) -> list[dict]: traces_by_stream = {row.get("streamVaHex"): row for row in stream_traces} rows = [] for edge in frontier_branches: for branch in edge.get("branchSteps") or []: trace = traces_by_stream.get(branch.get("leafPointerHex")) or traces_by_stream.get(branch.get("streamVaHex")) or {} steps = trace.get("trace") or [] branch_step = next( (step for step in steps if step.get("vaHex") == branch.get("streamVaHex")), None, ) if not branch_step: branch_step = next((step for step in steps if step.get("opcodeHex") == "0x11"), {}) branch_fields = stream_fields(branch_step) setup_steps = [] for step in steps: if step.get("vaHex") == branch_step.get("vaHex"): break if step.get("opcodeHex") not in {"0x10", "0x12"}: continue fields = stream_fields(step) setup_steps.append({ "vaHex": step.get("vaHex"), "opcodeHex": step.get("opcodeHex"), "handlerVaHex": step.get("handlerVaHex"), "valueHex": step.get("valueHex"), "selectionBufferOffsetHex": fields["streamPlus2Hex"], "selectionBufferOffset": fields["streamPlus2"], "stateTable": ( "primaryBranchState" if step.get("opcodeHex") == "0x10" and fields["streamPlus1"] == 0 else "secondaryBranchState" if step.get("opcodeHex") == "0x10" else None ), "meaning": ( "fill branch-state table from helper argument stream+2" if step.get("opcodeHex") == "0x10" else "select/store active state slot into selectionBuffer[stream+2]" ), }) branch_offset = branch.get("selectionBufferOffset") local_writes = [ step for step in setup_steps if step.get("opcodeHex") == "0x12" and step.get("selectionBufferOffset") == branch_offset ] rows.append({ "source": edge.get("source"), "target": edge.get("target"), "leafPointerHex": branch.get("leafPointerHex"), "branchVaHex": branch.get("streamVaHex"), "condition": branch.get("condition"), "branchSelectionBufferOffset": branch_offset, "branchSelectionBufferOffsetHex": branch.get("selectionBufferOffsetHex"), "branchStateTable": branch.get("stateTable"), "setupSteps": setup_steps, "localWriterCount": len(local_writes), "hasLocalWriterForBranchOffset": bool(local_writes), "conclusion": ( "The branch reads selectionBuffer[0x20], but this frontier stream does not write " "that offset before the branch. The value is inherited from wider runtime state; " "the next reverse-engineering step is tracing writers to context+0xa8+0x20." if not local_writes else "The branch offset is written earlier in the same stream." ), }) return rows def markdown(rows: list[dict]) -> str: lines = [ "# Save Selector Frontier Selection Flow", "", "Selection-buffer offsets used before save-selector frontier branches.", "", "| source | target | branch | setup offsets | local writer? | conclusion |", "| --- | --- | --- | --- | --- | --- |", ] for row in rows: setup = ", ".join( f"{step['vaHex']} {step['opcodeHex']}->{step['selectionBufferOffsetHex']}" for step in row.get("setupSteps") or [] ) or "-" lines.append( f"| {row.get('source')} | {row.get('target')} | " f"{row.get('branchVaHex')} `{row.get('condition')}` | {setup} | " f"{'yes' if row.get('hasLocalWriterForBranchOffset') else 'no'} | {row.get('conclusion')} |" ) if not rows: lines.append("| - | - | - | - | - | - |") lines.append("") return "\n".join(lines) def html_page(rows: list[dict]) -> str: body = [] for row in rows: setup = "
".join( html.escape( f"{step['vaHex']} {step['opcodeHex']} selectionBuffer[{step['selectionBufferOffsetHex']}] " f"{step.get('meaning')}" ) for step in row.get("setupSteps") or [] ) or "-" body.append( "" f"{html.escape(row.get('source') or '-')}" f"{html.escape(row.get('target') or '-')}" f"{html.escape(row.get('branchVaHex') or '-')}
{html.escape(row.get('condition') or '-')}" f"{setup}" f"{'yes' if row.get('hasLocalWriterForBranchOffset') else 'no'}" f"{html.escape(row.get('conclusion') or '')}" "" ) return "\n".join([ "", '', "", ' ', ' ', " Save Selector Frontier Selection Flow", " ", "", "", "

Save Selector Frontier Selection Flow

", "

Selection-buffer offsets used before save-selector frontier branches.

", " ", f" {''.join(body) or ''}
sourcetargetbranchsetup offsetslocal writer?conclusion
No rows.
", "", "", "", ]) def write_outputs(rows: list[dict], out_dir: Path = OUT) -> None: out_dir.mkdir(parents=True, exist_ok=True) (out_dir / "save_selector_frontier_selection_flow.json").write_text( json.dumps(rows, ensure_ascii=False, indent=2) + "\n", encoding="utf-8", ) def main() -> None: parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--frontier-branches", type=Path, default=OUT / "save_selector_frontier_branches.json") parser.add_argument("--stream-traces", type=Path, default=OUT / "save_selector_stream_traces.json") parser.add_argument("--out-dir", type=Path, default=OUT) args = parser.parse_args() rows = build_rows( json.loads(args.frontier_branches.read_text(encoding="utf-8")), json.loads(args.stream_traces.read_text(encoding="utf-8")), ) write_outputs(rows, args.out_dir) print(f"wrote {len(rows)} selection-flow rows -> {args.out_dir / 'save_selector_frontier_selection_flow.json'}") if __name__ == "__main__": main()