#!/usr/bin/env python3 """Run the original EXE without input and poll opening/runtime script pointers.""" from __future__ import annotations import argparse import html import json import os import shutil import struct import subprocess import time from pathlib import Path from typing import Any from probe_runtime_input_path import ( DEFAULT_PREFIX, IMAGE_BASE, IMAGE_SIZE, NULL_ALSA_CONFIG, OUT, find_free_display, find_hwanse_pid, hex32, load_selector_contexts, loaded_base, selected_pointer_context, static_from_runtime, ) ROOT = Path(__file__).resolve().parents[1] OPENING_STREAM_START = 0x004A2D38 OPENING_STREAM_END = 0x004A3DEC SELECTED_POINTER_GLOBAL = 0x0059DE30 GENERIC_STOP_FLAG = 0x0055A1B8 ACTIVE_ACTOR_COUNT = 0x004576E8 ACTIVE_ACTOR_POINTER_TABLE = 0x0059DD70 CURRENT_INPUT_MASK = 0x0059E310 PRESSED_EDGE_MASK = 0x0059E312 def static_to_runtime(static_va: int, base: int) -> int: return base + (static_va - IMAGE_BASE) def truncate(text: str, limit: int = 4000) -> str: if len(text) <= limit: return text return text[:limit] + f"\n... truncated {len(text) - limit} bytes ..." def process_maps(pid: int) -> list[tuple[int, int]]: rows: list[tuple[int, int]] = [] maps_path = Path("/proc") / str(pid) / "maps" try: lines = maps_path.read_text(encoding="utf-8", errors="replace").splitlines() except OSError: return rows for line in lines: parts = line.split(maxsplit=5) if not parts: continue try: start_text, end_text = parts[0].split("-", 1) rows.append((int(start_text, 16), int(end_text, 16))) except ValueError: continue return rows def is_mapped(ranges: list[tuple[int, int]], address: int, size: int = 1) -> bool: return any(start <= address and address + size <= end for start, end in ranges) def read_bytes(mem, ranges: list[tuple[int, int]], address: int, size: int) -> bytes | None: if not is_mapped(ranges, address, size): return None try: mem.seek(address) data = mem.read(size) except OSError: return None return data if len(data) == size else None def read_u8(mem, ranges: list[tuple[int, int]], address: int) -> int | None: data = read_bytes(mem, ranges, address, 1) return data[0] if data else None def read_u16(mem, ranges: list[tuple[int, int]], address: int) -> int | None: data = read_bytes(mem, ranges, address, 2) return struct.unpack(" int | None: data = read_bytes(mem, ranges, address, 4) return struct.unpack(" bool: return static_va is not None and OPENING_STREAM_START <= static_va < OPENING_STREAM_END def sample_runtime(pid: int, base: int, t0: float) -> dict[str, Any]: ranges = process_maps(pid) sample: dict[str, Any] = { "timeMs": int((time.time() - t0) * 1000), "readOk": False, } try: with (Path("/proc") / str(pid) / "mem").open("rb", buffering=0) as mem: selected_runtime = read_u32(mem, ranges, static_to_runtime(SELECTED_POINTER_GLOBAL, base)) selected_static = static_from_runtime(selected_runtime, base) stop_flag = read_u8(mem, ranges, static_to_runtime(GENERIC_STOP_FLAG, base)) active_count = read_u8(mem, ranges, static_to_runtime(ACTIVE_ACTOR_COUNT, base)) input_mask = read_u16(mem, ranges, static_to_runtime(CURRENT_INPUT_MASK, base)) edge_mask = read_u16(mem, ranges, static_to_runtime(PRESSED_EDGE_MASK, base)) actors = [] table_runtime = static_to_runtime(ACTIVE_ACTOR_POINTER_TABLE, base) for index in range(12): object_runtime = read_u32(mem, ranges, table_runtime + index * 4) if not object_runtime: continue script_runtime = read_u32(mem, ranges, object_runtime + 0x40) frame_runtime = read_u32(mem, ranges, object_runtime + 0x64) script_static = static_from_runtime(script_runtime, base) frame_static = static_from_runtime(frame_runtime, base) actors.append( { "slot": index, "objectRuntimeHex": hex32(object_runtime), "scriptRuntimeHex": hex32(script_runtime) if script_runtime is not None else None, "scriptStaticHex": hex32(script_static) if script_static is not None else None, "scriptInOpeningRange": in_opening_range(script_static), "frameScriptRuntimeHex": hex32(frame_runtime) if frame_runtime is not None else None, "frameScriptStaticHex": hex32(frame_static) if frame_static is not None else None, } ) opening_actors = [row for row in actors if row.get("scriptInOpeningRange")] sample.update( { "readOk": True, "selectedPointerRuntimeHex": hex32(selected_runtime) if selected_runtime is not None else None, "selectedPointerStaticHex": hex32(selected_static) if selected_static is not None else None, "selectedPointerInOpeningRange": in_opening_range(selected_static), "genericStopFlag": stop_flag, "activeActorCount": active_count, "currentInputMaskHex": f"0x{input_mask:04x}" if input_mask is not None else None, "pressedEdgeMaskHex": f"0x{edge_mask:04x}" if edge_mask is not None else None, "actorScriptRows": actors, "openingActorScriptRows": opening_actors, "openingActorScriptCount": len(opening_actors), } ) except OSError as exc: sample["error"] = str(exc) return sample def capture_screenshot(display: str, path: Path) -> dict[str, Any]: import_path = shutil.which("import") if not import_path: return {"path": str(path), "status": 127, "error": "ImageMagick import not found"} env = os.environ.copy() env["DISPLAY"] = display try: result = subprocess.run( [import_path, "-window", "root", str(path)], stdout=subprocess.PIPE, stderr=subprocess.STDOUT, text=True, errors="replace", timeout=5, env=env, ) return { "path": str(path), "status": result.returncode, "output": truncate(result.stdout), "exists": path.exists(), } except subprocess.TimeoutExpired as exc: return { "path": str(path), "status": 124, "output": truncate((exc.stdout or "") + (exc.stderr or "")), "exists": path.exists(), } def make_env(prefix: Path, display: str) -> dict[str, str]: env = os.environ.copy() runtime_dir = Path("/tmp") / f"hwanse-opening-noinput-{os.getuid()}" runtime_dir.mkdir(mode=0o700, exist_ok=True) try: runtime_dir.chmod(0o700) except OSError: pass env.update( { "DISPLAY": display, "WINEPREFIX": str(prefix), "WINEARCH": "win32", "WINEDEBUG": "-all", "XDG_RUNTIME_DIR": str(runtime_dir), } ) if NULL_ALSA_CONFIG.exists(): env["ALSA_CONFIG_PATH"] = str(NULL_ALSA_CONFIG) return env def build_summary( duration: float, interval: float, screenshot_interval: float, prefix: Path, out_dir: Path, capture_screenshots: bool = False, ) -> dict[str, Any]: display = find_free_display() prefix.mkdir(parents=True, exist_ok=True) screen_dir = out_dir / "runtime_opening_noinput_screens" screen_dir.mkdir(parents=True, exist_ok=True) env = make_env(prefix, display) subprocess.run(["wineserver", "-k"], cwd=ROOT, env=env, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, timeout=8) xvfb = subprocess.Popen( ["Xvfb", display, "-screen", "0", "1280x1024x24"], stdout=subprocess.PIPE, stderr=subprocess.STDOUT, text=True, errors="replace", ) time.sleep(1) wine = subprocess.Popen( ["wine", "explorer", "/desktop=hwanse,640x480", "Hwanse2.exe"], cwd=ROOT, env=env, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, text=True, errors="replace", ) samples = [] screenshots = [] t0 = time.time() next_capture = 0.0 pid = None base = None try: while time.time() - t0 < duration: pid = find_hwanse_pid() base = loaded_base(pid) if pid else None if pid and base: samples.append(sample_runtime(pid, base, t0)) else: samples.append({"timeMs": int((time.time() - t0) * 1000), "readOk": False, "pid": pid, "loadedBaseHex": None}) elapsed = time.time() - t0 if capture_screenshots and elapsed >= next_capture: shot = screen_dir / f"shot_{int(elapsed * 1000):05d}.png" screenshots.append(capture_screenshot(display, shot)) next_capture += screenshot_interval time.sleep(interval) finally: subprocess.run(["wineserver", "-k"], cwd=ROOT, env=env, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, timeout=8) try: wine_output, _ = wine.communicate(timeout=5) except subprocess.TimeoutExpired: wine.kill() wine_output, _ = wine.communicate(timeout=5) xvfb.terminate() try: xvfb_output, _ = xvfb.communicate(timeout=5) except subprocess.TimeoutExpired: xvfb.kill() xvfb_output, _ = xvfb.communicate(timeout=5) roots, contexts = load_selector_contexts() selected_values = sorted({row.get("selectedPointerStaticHex") for row in samples if row.get("selectedPointerStaticHex")}) opening_selected_count = sum(1 for row in samples if row.get("selectedPointerInOpeningRange")) opening_actor_count = sum(1 for row in samples if row.get("openingActorScriptCount")) return { "objective": "no-input Wine runtime poll from process start through the opening/title prelude", "durationSeconds": duration, "intervalSeconds": interval, "screenshotIntervalSeconds": screenshot_interval, "screenshotCaptureEnabled": capture_screenshots, "display": display, "winePrefix": str(prefix), "linuxPidLast": pid, "loadedBaseHexLast": hex32(base) if base else None, "openingRangeHex": f"{hex32(OPENING_STREAM_START)}..{hex32(OPENING_STREAM_END)}", "selectedPointerGlobalHex": hex32(SELECTED_POINTER_GLOBAL), "activeActorPointerTableHex": hex32(ACTIVE_ACTOR_POINTER_TABLE), "samples": samples, "sampleCount": len(samples), "selectedPointerStaticValues": selected_values, "selectedPointerContexts": [ {"valueHex": value, "context": selected_pointer_context(value, roots, contexts)} for value in selected_values ], "openingSelectedSampleCount": opening_selected_count, "openingActorScriptSampleCount": opening_actor_count, "openingSelectedObserved": opening_selected_count > 0, "openingActorScriptObserved": opening_actor_count > 0, "screenshots": screenshots, "successfulScreenshots": [row for row in screenshots if row.get("exists")], "startupOutput": truncate(wine_output or ""), "xvfbOutput": truncate(xvfb_output or ""), "promotionStatus": "runtime-opening-selected-pointer-observed" if opening_selected_count else "runtime-opening-pointer-not-observed", "conclusion": ( "The no-input runtime poll observed the original EXE holding selected-pointer global 0x0059de30 inside the opening stream range." if opening_selected_count else "The no-input runtime poll did not observe selected-pointer global 0x0059de30 inside the opening stream range." ), } def markdown(summary: dict[str, Any]) -> str: lines = [ "# Runtime Opening No-Input Poll", "", f"- duration: {summary['durationSeconds']}s", f"- loaded base last: `{summary.get('loadedBaseHexLast') or '-'}`", f"- opening range: `{summary['openingRangeHex']}`", f"- selected pointer global: `{summary['selectedPointerGlobalHex']}`", f"- selected values: `{', '.join(summary['selectedPointerStaticValues']) or '-'}`", f"- opening selected observed: {summary['openingSelectedObserved']} ({summary['openingSelectedSampleCount']}/{summary['sampleCount']})", f"- opening actor script observed: {summary['openingActorScriptObserved']} ({summary['openingActorScriptSampleCount']}/{summary['sampleCount']})", f"- screenshots: {len(summary['successfulScreenshots'])}", f"- promotion status: `{summary['promotionStatus']}`", "", summary["conclusion"], "", "## Timeline", "", "| t | selected static | selected opening | actor opening scripts | stop flag | input |", "|---:|---|---:|---:|---:|---|", ] for row in summary["samples"]: lines.append( f"| {row.get('timeMs')}ms | `{row.get('selectedPointerStaticHex') or '-'}` | " f"{row.get('selectedPointerInOpeningRange')} | {row.get('openingActorScriptCount') or 0} | " f"{row.get('genericStopFlag') if row.get('genericStopFlag') is not None else '-'} | " f"`{row.get('currentInputMaskHex') or '-'} / {row.get('pressedEdgeMaskHex') or '-'}` |" ) lines.append("") return "\n".join(lines) def html_page(summary: dict[str, Any]) -> str: timeline_rows = "\n".join( "" f"{row.get('timeMs')}ms" f"{html.escape(row.get('selectedPointerStaticHex') or '-')}" f"{row.get('selectedPointerInOpeningRange')}" f"{row.get('openingActorScriptCount') or 0}" f"{row.get('genericStopFlag') if row.get('genericStopFlag') is not None else '-'}" f"{html.escape(row.get('currentInputMaskHex') or '-')} / {html.escape(row.get('pressedEdgeMaskHex') or '-')}" "" for row in summary["samples"] ) screenshot_rows = "\n".join( "
" f"
{html.escape(Path(row['path']).name)} status {row.get('status')}
" "
" for row in summary["successfulScreenshots"] ) context_items = "".join( f"
  • {html.escape(row['valueHex'])} " f"{html.escape(str((row.get('context') or {}).get('selector') or '-'))}
    • " for row in summary["selectedPointerContexts"] ) return f""" Runtime Opening No-Input Poll

    Runtime Opening No-Input Poll

    Home ยท opening consumer trace

    • duration: {summary['durationSeconds']}s
    • opening range: {html.escape(summary['openingRangeHex'])}
    • selected pointer global: {summary['selectedPointerGlobalHex']}
    • opening selected observed: {summary['openingSelectedObserved']} ({summary['openingSelectedSampleCount']}/{summary['sampleCount']})
    • opening actor script observed: {summary['openingActorScriptObserved']} ({summary['openingActorScriptSampleCount']}/{summary['sampleCount']})
    • promotion status: {summary['promotionStatus']}

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

    Selected Pointer Contexts

      {context_items or '
    • -
      • '}

    Screenshot Samples

    {screenshot_rows or '

    Screenshot capture is disabled for active reports. Use --capture-screenshots for a disposable visual probe.

    '}

    Timeline

    {timeline_rows}
    tselected staticselected openingactor opening scriptsstop flaginput
    """ def write_outputs(summary: dict[str, Any], out_dir: Path, md_out: Path | None = None) -> None: out_dir.mkdir(parents=True, exist_ok=True) (out_dir / "runtime_opening_noinput_poll.json").write_text( json.dumps(summary, ensure_ascii=False, indent=2) + "\n", encoding="utf-8", ) if md_out is not None: md_out.write_text(markdown(summary), encoding="utf-8") (out_dir / "runtime_opening_noinput_poll.html").write_text(html_page(summary), encoding="utf-8") def main() -> None: parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--duration", type=float, default=35.0) parser.add_argument("--interval", type=float, default=0.5) parser.add_argument("--screenshot-interval", type=float, default=5.0) parser.add_argument("--capture-screenshots", action="store_true") parser.add_argument("--prefix", type=Path, default=DEFAULT_PREFIX) parser.add_argument("--out-dir", type=Path, default=OUT) args = parser.parse_args() summary = build_summary( args.duration, args.interval, args.screenshot_interval, args.prefix, args.out_dir, capture_screenshots=args.capture_screenshots, ) write_outputs(summary, args.out_dir) print(f"runtime opening no-input poll -> {args.out_dir / 'runtime_opening_noinput_poll.html'}") print(summary["conclusion"]) if __name__ == "__main__": main()