#!/usr/bin/env python3 """Summarize storage context for opcode 0x24 mode 1 source byte 0x0059e348.""" from __future__ import annotations import argparse import html import json 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 read_sections, va_to_offset ROOT = Path(__file__).resolve().parents[1] OUT = ROOT / "out" MODE1_SOURCE = 0x0059E348 DIAGNOSTIC_WATCH_POLL = "runtime_selected_pointer_patched_public_selector_2_0_active_order_poll.json" DIAGNOSTIC_WATCH_NAMES = [ "opcode24Mode1Source", "opcode24RuntimeFlag", "opcode24CurrentObjectIndex", ] OPCODE24_RUNTIME_PRODUCER_MISSING_EVIDENCE_BY_GATE = { "indirect-runtime-producer": ( "indirect runtime producer of 0x0059e348 on a normal route path" ), "opcode24-route-stream-selection": ( "opcode 0x24 mode 1 object+0x61 writes select a current-route stream" ), "strict-hotspot": "strict map1_01a source coordinate or non-coordinate trigger", } def hex32(value: int) -> str: return f"0x{value:08x}" def parse_hex(value: str | None) -> int: return int(value or "0", 16) def load_json(path: Path, fallback: Any) -> Any: if not path.exists(): return fallback return json.loads(path.read_text(encoding="utf-8")) def list_text(values: list[Any] | None) -> str: return ",".join(str(value) for value in (values or [])) or "-" def section_for_va(sections: list[dict], va: int) -> dict | None: for section in sections: start = section["va"] end = start + max(section["size"], section["raw_size"]) if start <= va < end: return section return None def section_summary(exe: bytes, sections: list[dict]) -> dict: section = section_for_va(sections, MODE1_SOURCE) if not section: return { "sectionName": None, "mode1SourceInKnownSection": False, "mode1SourceRawOffsetHex": None, "mode1SourceHasRawByte": False, "staticInitialValue": None, "staticInitialValueKind": "unknown-section", } raw_end_va = section["va"] + section["raw_size"] virtual_end_va = section["va"] + section["size"] raw_offset = va_to_offset(sections, MODE1_SOURCE) has_raw_byte = raw_offset is not None and raw_offset < len(exe) static_value = exe[raw_offset] if has_raw_byte else 0 static_kind = "raw-file-byte" if has_raw_byte else "pe-zero-filled-section-tail" return { "sectionName": section["name"], "sectionVaHex": hex32(section["va"]), "sectionRawSizeHex": hex32(section["raw_size"]), "sectionVirtualSizeHex": hex32(section["size"]), "sectionRawEndVaHex": hex32(raw_end_va), "sectionVirtualEndVaHex": hex32(virtual_end_va), "mode1SourceInKnownSection": True, "mode1SourceRawOffsetHex": hex32(raw_offset) if raw_offset is not None else None, "mode1SourceHasRawByte": has_raw_byte, "mode1SourceAfterRawEndBytes": max(0, MODE1_SOURCE - raw_end_va), "staticInitialValue": static_value, "staticInitialValueHex": f"0x{static_value:02x}", "staticInitialValueKind": static_kind, "zeroInitializedByPeLoader": not has_raw_byte, } def save_block_rows(save_loader_trace: dict) -> list[dict]: rows = [] for block in save_loader_trace.get("saveReadBlocks") or []: target = parse_hex(block.get("targetVaHex")) size = parse_hex(block.get("sizeHex")) end = target + size contains = target <= MODE1_SOURCE < end rows.append({ "callVaHex": block.get("callVaHex"), "targetVaHex": block.get("targetVaHex"), "endVaHex": hex32(end), "saveOffsetHex": block.get("saveOffsetHex"), "sizeHex": block.get("sizeHex"), "description": block.get("description"), "containsMode1Source": contains, "distanceAfterEnd": max(0, MODE1_SOURCE - end), "distanceAfterEndHex": hex32(max(0, MODE1_SOURCE - end)), }) return rows def aggregate_watch_values(rows: list[dict]) -> dict[str, list[dict[str, Any]]]: aggregate: dict[str, dict[str, int]] = {} for row in rows: for name, values in (row.get("uniqueWatchValues") or {}).items(): if name not in DIAGNOSTIC_WATCH_NAMES: continue target = aggregate.setdefault(name, {}) for value_row in values or []: value_hex = value_row.get("valueHex") if value_hex is None: continue target[value_hex] = target.get(value_hex, 0) + int(value_row.get("count") or 0) return { name: [ {"valueHex": value_hex, "count": count} for value_hex, count in sorted(value_rows.items(), key=lambda item: (-item[1], item[0])) ] for name, value_rows in sorted(aggregate.items()) } def stable_watch_hex(watch_values: dict[str, list[dict[str, Any]]], sample_count: int, name: str) -> str | None: rows = watch_values.get(name) or [] if len(rows) != 1: return None if rows[0].get("count") != sample_count: return None return rows[0].get("valueHex") def build_diagnostic_runtime_watch(active_order_poll: dict | None) -> dict: active_order_poll = active_order_poll or {} if not active_order_poll: return { "available": False, "sourcePoll": DIAGNOSTIC_WATCH_POLL, "diagnosticOnly": True, } all_rows = active_order_poll.get("rows") or [] route_rows = [row for row in all_rows if row.get("reachedRouteSelectorContext")] if not route_rows: return { "available": False, "sourcePoll": DIAGNOSTIC_WATCH_POLL, "diagnosticOnly": True, "totalSampleCount": active_order_poll.get("sampleCount"), "totalSequenceCount": active_order_poll.get("sequenceCount"), "routeSampleCount": 0, "routeSequenceCount": 0, "promotionStatus": "missing-route-row", } route_sample_count = sum(int(row.get("sampleCount") or 0) for row in route_rows) watch_values = aggregate_watch_values(route_rows) mode1_source = stable_watch_hex(watch_values, route_sample_count, "opcode24Mode1Source") runtime_flag = stable_watch_hex(watch_values, route_sample_count, "opcode24RuntimeFlag") current_object_index = stable_watch_hex(watch_values, route_sample_count, "opcode24CurrentObjectIndex") return { "available": True, "sourcePoll": DIAGNOSTIC_WATCH_POLL, "diagnosticOnly": True, "totalSampleCount": active_order_poll.get("sampleCount"), "totalSequenceCount": active_order_poll.get("sequenceCount"), "routeSampleCount": route_sample_count, "routeSequenceCount": len(route_rows), "routeSequenceNames": [row.get("name") for row in route_rows if row.get("name")], "observedSelectors": active_order_poll.get("observedSelectors") or [], "reachedCurrentRoot": active_order_poll.get("anyReachedCurrentRoot"), "reachedRouteSelector": active_order_poll.get("anyReachedRouteSelectorContext"), "mode1SourceValueHex": mode1_source, "runtimeFlagValueHex": runtime_flag, "currentObjectIndexValueHex": current_object_index, "allWatchedValuesStable": all( stable_watch_hex(watch_values, route_sample_count, name) is not None for name in DIAGNOSTIC_WATCH_NAMES ), "runtimeFlagStaysDisabledInDiagnostic": runtime_flag == "0x00", "mode1SourceStaysZeroInDiagnostic": mode1_source == "0x00", "currentObjectIndexStaysZeroInDiagnostic": current_object_index == "0x00", "normalRouteProof": False, "notRoutePromotionProof": True, "promotionStatus": "diagnostic-only", } def build_summary( exe: bytes, save_loader_trace: dict, mode1_source_writes: dict, active_order_poll: dict | None = None, ) -> dict: sections = read_sections(exe) storage = section_summary(exe, sections) blocks = save_block_rows(save_loader_trace) containing_blocks = [row for row in blocks if row["containsMode1Source"]] direct = { "exactMode1RefCount": mode1_source_writes.get("exactMode1RefCount", 0), "coveringWriteCount": mode1_source_writes.get("coveringWriteCount", 0), "indexedWriteCandidateCount": mode1_source_writes.get("indexedWriteCandidateCount", 0), "addressProducerCandidateCount": mode1_source_writes.get("addressProducerCandidateCount", 0), "staticProducerCandidateCount": mode1_source_writes.get("staticProducerCandidateCount", 0), } not_savedata_backed = not containing_blocks no_static_producer = ( direct["coveringWriteCount"] == 0 and direct["indexedWriteCandidateCount"] == 0 and direct["addressProducerCandidateCount"] == 0 ) diagnostic_watch = build_diagnostic_runtime_watch(active_order_poll) conclusion = ( "Opcode 0x24 mode 1 source byte 0x0059e348 lives in the zero-filled tail of the .data section: " "there is no raw file byte for it, so its static process-start value is 0. It is also outside all " "three save-loader read blocks, and the direct/static producer scan found no writer. This rules out " "treating 0x0059e348 as a savedata-backed selector or static leaf chooser for map1_01a->map2_02d; " "in the current static evidence it is not a savedata-backed selector. The patched selector 2:0 diagnostic " "route row also keeps 0x0059e348 at 0x00, so that constructed run does not reveal an indirect producer. " "It may still be mutated indirectly on a normal route path, so promotion remains blocked without a runtime " "producer trace or equivalent proof." ) evidence_refs = [ { "path": "out/save_loader_trace.json", "fields": [ "saveReadBlocks", ], }, { "path": "out/save_selector_opcode24_mode1_source_writes.json", "fields": [ "rows", "exactMode1RefCount", "coveringWriteCount", "indexedWriteCandidateCount", "addressProducerCandidateCount", "staticProducerCandidateCount", "promotionStatus", ], }, { "path": f"out/{DIAGNOSTIC_WATCH_POLL}", "fields": [ "rows", "sampleCount", "sequenceCount", "observedSelectors", "anyReachedCurrentRoot", "anyReachedRouteSelectorContext", "promotionStatus", ], }, { "path": "out/save_selector_opcode24_mode1_indirect_context.json", "fields": [ "mode1DirectRefs", "globalBaseDirectRefs", "basePlusMode1OffsetCandidateCount", "baseWindowMode1WriteCandidateCount", "nearbyBaseWindowMode1WriteCandidateCount", "noStaticBaseIndirectCandidate", "promotionStatus", ], }, { "path": "out/save_selector_opcode24_mode1_file_read_context.json", "fields": [ "globalDestinationReadFileRows", "mode1FileReadCandidates", "readFileCallCount", "mode1FileReadCandidateCount", "mode1FileReadProducerFound", ], }, { "path": "out/save_selector_opcode24_mode1_block_writes.json", "fields": [ "rows", "addressLikeCoveringBases", "directCoveringWrites", "blockWriteCandidates", "blockWriteCandidateCount", ], }, { "path": "out/save_selector_opcode24_mode1_default_effect.json", "fields": [ "object61ConsumerGroups", "branchOperand", "remainingProofs", "routeOperandRowCount", "staticDefaultPromotesRoute", "runtimeProducerRequired", "promotionStatus", ], }, { "path": "out/save_selector_opcode24_runtime_enabled_context.json", "fields": [ "runtimeEnabledFlagHex", "modeDispatchRequiresRuntimeFlagOne", "directWriteCount", "staticEvidenceProvesModeDispatch", "remainingProofs", "promotionStatus", ], }, { "path": "out/runtime_opcode24_flag_context.json", "fields": [ "classification", "pollCount", "runtimeFlagOneCount", "mode1SourceNonzeroObserved", "selectedRootExecutionProofFound", "routePromotionEvidenceFound", ], }, { "path": "out/runtime_trace_execution_probe.json", "fields": [ "canCaptureTraceNow", "blockers", "probes", "qemuI386Binfmt", "relocationContext", "promotionStatus", ], }, ] failed_gate_ids = list(OPCODE24_RUNTIME_PRODUCER_MISSING_EVIDENCE_BY_GATE) missing_evidence = [ OPCODE24_RUNTIME_PRODUCER_MISSING_EVIDENCE_BY_GATE[gate_id] for gate_id in failed_gate_ids ] return { "mode1SourceHex": hex32(MODE1_SOURCE), "storage": storage, "saveReadBlocks": blocks, "saveReadBlockContainsMode1Source": bool(containing_blocks), "directProducerSummary": direct, "notSavedataBacked": not_savedata_backed, "noStaticProducer": no_static_producer, "staticInitialZero": storage.get("staticInitialValue") == 0, "diagnosticRuntimeWatch": diagnostic_watch, "proofFound": False, "opcode24RuntimeProducerProofFound": False, "opcode24RouteStreamSelectionProofFound": False, "strictHotspotFound": False, "failedOpcode24RuntimeProducerGateIds": failed_gate_ids, "missingEvidence": missing_evidence, "promotionStatus": "blocked", "remainingProofs": [ "find an indirect runtime producer of 0x0059e348, if one exists", "prove opcode 0x24 mode 1 object+0x61 writes select a current-route stream, not only object state", "find a strict map1_01a source coordinate or non-coordinate trigger", ], "evidenceRefs": evidence_refs, "evidenceRefCount": len(evidence_refs), "conclusion": conclusion, } def markdown(summary: dict) -> str: storage = summary["storage"] direct = summary["directProducerSummary"] diagnostic = summary.get("diagnosticRuntimeWatch") or {} observed_selectors = ", ".join(diagnostic.get("observedSelectors") or []) or "-" route_sequences = ", ".join(diagnostic.get("routeSequenceNames") or []) or "-" lines = [ "# Save Selector Opcode 0x24 Mode1 Runtime Context", "", f"- mode1 source: `{summary['mode1SourceHex']}`", f"- section: `{storage.get('sectionName')}`", f"- section raw end: `{storage.get('sectionRawEndVaHex')}`", f"- section virtual end: `{storage.get('sectionVirtualEndVaHex')}`", f"- has raw file byte: {storage.get('mode1SourceHasRawByte')}", f"- static initial value: `{storage.get('staticInitialValueHex')}` ({storage.get('staticInitialValueKind')})", f"- save loader block contains mode1 source: {summary['saveReadBlockContainsMode1Source']}", f"- direct/static producers: exact refs={direct['exactMode1RefCount']}, covering writes={direct['coveringWriteCount']}, indexed writes={direct['indexedWriteCandidateCount']}, address producers={direct['addressProducerCandidateCount']}", f"- evidence refs: {summary['evidenceRefCount']}", f"- proofFound: {summary.get('proofFound')}", f"- opcode24 runtime producer proof found: {summary.get('opcode24RuntimeProducerProofFound')}", f"- opcode24 route-stream selection proof found: {summary.get('opcode24RouteStreamSelectionProofFound')}", f"- strict hotspot found: {summary.get('strictHotspotFound')}", f"- failed opcode24 runtime producer gates: `{list_text(summary.get('failedOpcode24RuntimeProducerGateIds'))}`", f"- missing evidence count: {len(summary.get('missingEvidence') or [])}", f"- promotion status: `{summary['promotionStatus']}`", "", summary["conclusion"], "", "## Missing Evidence", "", *[f"- {item}" for item in summary.get("missingEvidence") or []], "", "## Evidence Refs", "", "| path | fields |", "| --- | --- |", ] for ref in summary.get("evidenceRefs") or []: lines.append( f"| `{ref.get('path')}` | `{list_text(ref.get('fields'))}` |" ) lines.extend([ "", "## Diagnostic Runtime Watch", "", "| field | value |", "| --- | --- |", f"| source poll | `{diagnostic.get('sourcePoll')}` |", f"| available | `{diagnostic.get('available')}` |", f"| diagnostic only | `{diagnostic.get('diagnosticOnly')}` |", f"| route samples | `{diagnostic.get('routeSampleCount')}` across `{diagnostic.get('routeSequenceCount')}` route-reaching sequence(s) |", f"| total samples | `{diagnostic.get('totalSampleCount')}` across `{diagnostic.get('totalSequenceCount')}` sequence(s) |", f"| route sequence names | `{route_sequences}` |", f"| observed selectors | `{observed_selectors}` |", f"| mode1 source | `{diagnostic.get('mode1SourceValueHex')}` |", f"| runtime flag | `{diagnostic.get('runtimeFlagValueHex')}` |", f"| current object index | `{diagnostic.get('currentObjectIndexValueHex')}` |", f"| all watched values stable | `{diagnostic.get('allWatchedValuesStable')}` |", f"| normal route proof | `{diagnostic.get('normalRouteProof')}` |", f"| promotion status | `{diagnostic.get('promotionStatus')}` |", "", "## Save Read Blocks", "", "| call | target | end | save offset | size | contains | distance after end | description |", "| --- | --- | --- | --- | --- | --- | ---: | --- |", ]) for row in summary["saveReadBlocks"]: lines.append( f"| `{row.get('callVaHex')}` | `{row.get('targetVaHex')}` | `{row.get('endVaHex')}` | " f"`{row.get('saveOffsetHex')}` | `{row.get('sizeHex')}` | {row.get('containsMode1Source')} | " f"`{row.get('distanceAfterEndHex')}` | {row.get('description') or '-'} |" ) lines.extend(["", "## Remaining Proofs", ""]) lines.extend(f"- {item}" for item in summary["remainingProofs"]) lines.append("") return "\n".join(lines) def html_page(summary: dict) -> str: storage = summary["storage"] direct = summary["directProducerSummary"] diagnostic = summary.get("diagnosticRuntimeWatch") or {} observed_selectors = ", ".join(diagnostic.get("observedSelectors") or []) or "-" route_sequences = ", ".join(diagnostic.get("routeSequenceNames") or []) or "-" block_rows = "\n".join( "" f"{html.escape(str(row.get('callVaHex')))}" f"{html.escape(str(row.get('targetVaHex')))}" f"{html.escape(str(row.get('endVaHex')))}" f"{html.escape(str(row.get('saveOffsetHex')))}" f"{html.escape(str(row.get('sizeHex')))}" f"{row.get('containsMode1Source')}" f"{html.escape(str(row.get('distanceAfterEndHex')))}" f"{html.escape(str(row.get('description') or '-'))}" "" for row in summary["saveReadBlocks"] ) proofs = "\n".join(f"
  • {html.escape(item)}
    • " for item in summary["remainingProofs"]) missing_items = "\n".join( f"
  • {html.escape(item)}
    • " for item in summary.get("missingEvidence") or [] ) or "
  • -
    • " evidence_ref_rows = "\n".join( "" f"{html.escape(ref.get('path') or '-')}" f"{html.escape(list_text(ref.get('fields')))}" "" for ref in summary.get("evidenceRefs") or [] ) or 'No evidence refs recorded.' diagnostic_rows = "\n".join([ f"source poll{html.escape(str(diagnostic.get('sourcePoll')))}", f"available{html.escape(str(diagnostic.get('available')))}", f"diagnostic only{html.escape(str(diagnostic.get('diagnosticOnly')))}", f"route samples{html.escape(str(diagnostic.get('routeSampleCount')))} across {html.escape(str(diagnostic.get('routeSequenceCount')))} route-reaching sequence(s)", f"total samples{html.escape(str(diagnostic.get('totalSampleCount')))} across {html.escape(str(diagnostic.get('totalSequenceCount')))} sequence(s)", f"route sequence names{html.escape(route_sequences)}", f"observed selectors{html.escape(observed_selectors)}", f"mode1 source{html.escape(str(diagnostic.get('mode1SourceValueHex')))}", f"runtime flag{html.escape(str(diagnostic.get('runtimeFlagValueHex')))}", f"current object index{html.escape(str(diagnostic.get('currentObjectIndexValueHex')))}", f"all watched values stable{html.escape(str(diagnostic.get('allWatchedValuesStable')))}", f"normal route proof{html.escape(str(diagnostic.get('normalRouteProof')))}", f"promotion status{html.escape(str(diagnostic.get('promotionStatus')))}", ]) return "\n".join([ "Save Selector Opcode 0x24 Mode1 Runtime Context", "", "

    Save Selector Opcode 0x24 Mode1 Runtime Context

    ", "", f"

    {html.escape(summary['conclusion'])}

    ", "

    Missing Evidence

    ", "

    Evidence Refs

    ", "", evidence_ref_rows, "
    pathfields
    ", "

    Diagnostic Runtime Watch

    ", "", diagnostic_rows, "
    ", "

    Save Read Blocks

    ", "", block_rows, "
    calltargetendsave offsetsizecontainsdistance after enddescription
    ", "

    Remaining Proofs

    ", ]) def write_outputs(summary: dict, out_dir: Path = OUT, html_out: Path | None = None) -> Path: out_dir.mkdir(parents=True, exist_ok=True) json_out = out_dir / "save_selector_opcode24_mode1_runtime_context.json" json_out.write_text( json.dumps(summary, ensure_ascii=False, indent=2) + "\n", encoding="utf-8", ) if html_out is not None: html_out.parent.mkdir(parents=True, exist_ok=True) html_out.write_text(html_page(summary), encoding="utf-8") return json_out def main() -> None: parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--exe", type=Path, default=ROOT / "Hwanse2.exe") parser.add_argument("--save-loader-trace", type=Path, default=OUT / "save_loader_trace.json") parser.add_argument("--mode1-source-writes", type=Path, default=OUT / "save_selector_opcode24_mode1_source_writes.json") parser.add_argument("--diagnostic-poll", type=Path, default=OUT / DIAGNOSTIC_WATCH_POLL) parser.add_argument("--out-dir", type=Path, default=OUT) parser.add_argument("--html-out", type=Path) args = parser.parse_args() summary = build_summary( args.exe.read_bytes(), load_json(args.save_loader_trace, {}), load_json(args.mode1_source_writes, {}), load_json(args.diagnostic_poll, {}), ) json_out = write_outputs(summary, args.out_dir, args.html_out) print(f"wrote opcode24 mode1 runtime context -> {json_out}") if __name__ == "__main__": main()