#!/usr/bin/env python3 """Run bounded Wine runtime probes for the EXE trace environment.""" from __future__ import annotations import argparse import json import os import signal import socket import shutil import subprocess import time from pathlib import Path ROOT = Path(__file__).resolve().parents[1] OUT = ROOT / "out" DEFAULT_PREFIX = Path.home() / ".cache" / "hwanse-wine32" NULL_ALSA_CONFIG = ROOT / "tools" / "runtime_trace_null_alsa.conf" PE_IMAGE_BASE = 0x00400000 TRACE_STATIC_POINTS = { "mode1SourceWriteWatch": 0x0059E348, "opcode24Mode1Read": 0x0040C675, "currentFrontierReader": 0x00542B0C, "gateTimeContextBase": 0x005428C4, } CRASH_MARKERS = [ "Segmentation fault", "uncaught target signal 11", "Protocol error", "Assertion `!status' failed", "could not exec the wine loader", "Exec format error", "Can't attach process", "No process loaded", "WINEDBG_STATUS=255", "Could not insert hardware watchpoint", "Could not insert hardware breakpoints", "Remote connection closed", "A fatal error internal to GDB", ] FAILED_RUNTIME_TRACE_EXECUTION_GATE_IDS = [ "qemu-i386-wine-trace-capture", "gdbstub-breakpoint-watchpoint-stability", "route-watchpoint-trace-proof", ] RUNTIME_TRACE_EXECUTION_MISSING_EVIDENCE = [ "stable qemu-i386/Wine trace capture session for Hwanse2.exe", "breakpoint/watchpoint session that survives selected trace points", "route execution trace reaching selected-root/opcode24 producer watchpoints", ] RUNTIME_TRACE_EXECUTION_EVIDENCE_REFS = [ {"path": "tools/probe_runtime_trace_execution.py", "description": "bounded Wine/qemu/gdbstub execution probes"}, {"path": "tools/runtime_trace_null_alsa.conf", "description": "headless ALSA config used for Wine runtime probes"}, {"path": "Hwanse2.exe", "description": "target executable and static trace point addresses"}, ] def truncate(text: str, limit: int = 5000) -> str: if len(text) <= limit: return text return text[:limit] + f"\n... truncated {len(text) - limit} bytes ..." def text_part(value: str | bytes | None) -> str: if value is None: return "" if isinstance(value, bytes): return value.decode("utf-8", errors="replace") return value def run_probe(name: str, cmd: list[str], timeout: int, prefix: Path, extra_env: dict[str, str] | None = None) -> dict: env = os.environ.copy() env.update({ "WINEPREFIX": str(prefix), "WINEARCH": "win32", "WINEDEBUG": "-all", }) if NULL_ALSA_CONFIG.exists(): env["ALSA_CONFIG_PATH"] = str(NULL_ALSA_CONFIG) if extra_env: env.update(extra_env) timed_out = False try: result = subprocess.run( cmd, cwd=ROOT, env=env, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, text=True, errors="replace", timeout=timeout, ) status = result.returncode output = result.stdout except subprocess.TimeoutExpired as exc: timed_out = True status = 124 output = text_part(exc.stdout) + text_part(exc.stderr) text = truncate(output) crashed = any(marker in text for marker in CRASH_MARKERS) return { "name": name, "command": cmd, "timeoutSeconds": timeout, "status": status, "timedOut": timed_out, "crashed": crashed, "output": text, } def find_free_port() -> int: with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as sock: sock.bind(("127.0.0.1", 0)) return int(sock.getsockname()[1]) def run_qemu_gdbstub_probe( name: str, gdb_commands: list[str], timeout: int, prefix: Path, wine_args: list[str] | None = None, ) -> dict: env = os.environ.copy() env.update({ "WINEPREFIX": str(prefix), "WINEARCH": "win32", "WINEDEBUG": "-all", }) if NULL_ALSA_CONFIG.exists(): env["ALSA_CONFIG_PATH"] = str(NULL_ALSA_CONFIG) port = find_free_port() wine_args = wine_args or ["Hwanse2.exe"] qemu_cmd = ["xvfb-run", "-a", "qemu-i386", "-g", str(port), "/usr/lib/wine/wine", *wine_args] gdb_cmd = [ "gdb-multiarch", "-q", "-ex", "set debuginfod enabled off", "-ex", "set pagination off", "-ex", "set architecture i386", "-ex", f"target remote :{port}", ] for command in gdb_commands: gdb_cmd.extend(["-ex", command]) qemu = subprocess.Popen( qemu_cmd, cwd=ROOT, env=env, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, text=True, errors="replace", start_new_session=True, ) time.sleep(2) timed_out = False try: result = subprocess.run( gdb_cmd, cwd=ROOT, env=env, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, text=True, errors="replace", timeout=timeout, ) status = result.returncode gdb_output = result.stdout except subprocess.TimeoutExpired as exc: timed_out = True status = 124 gdb_output = text_part(exc.stdout) + text_part(exc.stderr) def stop_qemu() -> None: try: os.killpg(qemu.pid, signal.SIGTERM) except ProcessLookupError: return except OSError: qemu.terminate() def kill_qemu() -> None: try: os.killpg(qemu.pid, signal.SIGKILL) except ProcessLookupError: return except OSError: qemu.kill() stop_qemu() try: qemu_output, _ = qemu.communicate(timeout=5) except subprocess.TimeoutExpired: kill_qemu() qemu_output, _ = qemu.communicate(timeout=5) subprocess.run( ["wineserver", "-k"], cwd=ROOT, env=env, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, timeout=5, ) text = truncate("## gdb\n" + gdb_output + "\n## qemu\n" + (qemu_output or "")) crashed = any(marker in text for marker in CRASH_MARKERS) return { "name": name, "command": [ " ".join(qemu_cmd), " ".join(gdb_cmd), ], "timeoutSeconds": timeout, "status": status, "timedOut": timed_out, "crashed": crashed, "output": text, } def probe_success(row: dict, allow_timeout: bool = False) -> bool: if row.get("crashed"): return False if allow_timeout and row.get("timedOut"): return True return row.get("status") == 0 def read_binfmt() -> dict: path = Path("/proc/sys/fs/binfmt_misc/qemu-i386") if not path.exists(): return { "registered": False, "path": str(path), "raw": "", } raw = path.read_text(encoding="utf-8", errors="replace") return { "registered": "enabled" in raw, "path": str(path), "raw": raw, } def read_wine_audio_driver(prefix: Path) -> str | None: user_reg = prefix / "user.reg" if not user_reg.exists(): return None in_drivers = False for line in user_reg.read_text(encoding="utf-8", errors="replace").splitlines(): if line.startswith("["): in_drivers = line.startswith("[Software\\\\Wine\\\\Drivers]") continue if in_drivers and line.startswith('"Audio"='): return line.split("=", 1)[1].strip().strip('"') return None def parse_hex(value: str | None) -> int | None: if not value: return None try: return int(value, 16) except (TypeError, ValueError): return None def hex32(value: int) -> str: return f"0x{value:08x}" def signed_hex32(value: int) -> str: sign = "-" if value < 0 else "" return f"{sign}0x{abs(value):08x}" def load_relocation_context(out_dir: Path = OUT) -> dict: snapshot_path = out_dir / "runtime_memory_snapshot.json" if not snapshot_path.exists(): return { "available": False, "source": str(snapshot_path), "reason": "runtime memory snapshot is missing", } try: snapshot = json.loads(snapshot_path.read_text(encoding="utf-8")) except (OSError, json.JSONDecodeError) as exc: return { "available": False, "source": str(snapshot_path), "reason": f"runtime memory snapshot could not be read: {exc}", } image_base = parse_hex(snapshot.get("imageBaseHex")) or PE_IMAGE_BASE loaded_base = parse_hex(snapshot.get("loadedBaseHex")) if loaded_base is None: return { "available": False, "source": str(snapshot_path), "imageBaseHex": hex32(image_base), "reason": "runtime memory snapshot has no loadedBaseHex", } trace_points = [] for name, static_va in TRACE_STATIC_POINTS.items(): runtime_va = loaded_base + (static_va - image_base) trace_points.append({ "name": name, "staticVaHex": hex32(static_va), "runtimeVaHex": hex32(runtime_va), }) return { "available": True, "source": str(snapshot_path), "imageBaseHex": hex32(image_base), "loadedBaseHex": hex32(loaded_base), "deltaHex": signed_hex32(loaded_base - image_base), "tracePoints": trace_points, } def relocated_point(relocation_context: dict, name: str) -> str | None: for point in relocation_context.get("tracePoints") or []: if point.get("name") == name: return point.get("runtimeVaHex") return None def build_summary(prefix: Path = DEFAULT_PREFIX) -> dict: prefix.mkdir(parents=True, exist_ok=True) binfmt = read_binfmt() gdb_multiarch = shutil.which("gdb-multiarch") relocation_context = load_relocation_context() notepad_attach_script = ( "wine notepad >/tmp/hwanse-notepad.log 2>&1 & " "sleep 5; " "pid=$(wine tasklist | awk '/notepad.exe/ {print $2; exit}'); " "echo NOTEPAD_PID=$pid; " "printf 'info proc\\nattach %s\\ninfo threads\\ndetach\\nquit\\n' \"$pid\" > /tmp/hwanse-winedbg-attach.cmd; " "if [ -n \"$pid\" ]; then " "timeout 15 winedbg --file /tmp/hwanse-winedbg-attach.cmd; " "echo WINEDBG_STATUS=$?; " "fi; " "wineserver -k" ) notepad_launch_script = ( "printf 'info proc\\ninfo threads\\nquit\\n' > /tmp/hwanse-winedbg-launch.cmd; " "timeout 15 winedbg --file /tmp/hwanse-winedbg-launch.cmd notepad; " "echo WINEDBG_STATUS=$?; " "wineserver -k" ) virtual_desktop_attach_script = ( "wine explorer /desktop=hwanse,640x480 Hwanse2.exe & " "sleep 5; " "pid=$(wine tasklist | awk '/Hwanse2.exe/ {print $2; exit}'); " "echo HWANSE_PID=$pid; " "printf 'info proc\\nwatch *0x0059e348\\nbreak *0x0040c675\\ncont\\nquit\\n' > /tmp/hwanse-winedbg-watch.cmd; " "if [ -n \"$pid\" ]; then " "timeout 18 winedbg --file /tmp/hwanse-winedbg-watch.cmd \"$pid\"; " "echo WINEDBG_STATUS=$?; " "fi; " "wineserver -k" ) native_watch_script = ( "printf 'watch *0x0059e348\\nbreak *0x0040c675\\ncont\\nquit\\n' > /tmp/hwanse-winedbg-native.cmd; " "timeout 18 winedbg --file /tmp/hwanse-winedbg-native.cmd Hwanse2.exe; " "echo WINEDBG_STATUS=$?; " "wineserver -k" ) probes = [ run_probe("wine version", ["wine", "--version"], 15, prefix), run_probe("wineboot", ["xvfb-run", "-a", "wineboot", "-u"], 90, prefix), run_probe( "notepad winedbg attach control", ["xvfb-run", "-a", "bash", "-lc", notepad_attach_script], 25, prefix, ), run_probe( "notepad winedbg launch control", ["xvfb-run", "-a", "bash", "-lc", notepad_launch_script], 25, prefix, ), run_probe("Hwanse2.exe startup", ["xvfb-run", "-a", "wine", "Hwanse2.exe"], 15, prefix), run_probe( "Hwanse2.exe virtual desktop startup", ["xvfb-run", "-a", "wine", "explorer", "/desktop=hwanse,640x480", "Hwanse2.exe"], 15, prefix, ), run_probe( "Hwanse2.exe virtual desktop attach", ["xvfb-run", "-a", "bash", "-lc", virtual_desktop_attach_script], 30, prefix, ), run_probe("winedbg gdb startup", ["xvfb-run", "-a", "winedbg", "--gdb", "Hwanse2.exe"], 20, prefix), run_probe( "winedbg native watchpoint startup", ["xvfb-run", "-a", "bash", "-lc", native_watch_script], 20, prefix, ), ] if gdb_multiarch: probes.append(run_probe( "winedbg gdb-multiarch startup", ["xvfb-run", "-a", "winedbg", "--gdb", "Hwanse2.exe"], 20, prefix, {"WINE_GDB": gdb_multiarch}, )) probes.append(run_qemu_gdbstub_probe( "qemu gdbstub connect control", ["info registers eip", "detach", "quit"], 25, prefix, )) probes.append(run_qemu_gdbstub_probe( "qemu gdbstub virtual desktop connect control", ["info registers eip", "detach", "quit"], 25, prefix, ["explorer", "/desktop=hwanse,640x480", "Hwanse2.exe"], )) probes.append(run_qemu_gdbstub_probe( "qemu gdbstub hardware watchpoint startup", [ "watch *(unsigned char*)0x0059e348", "hbreak *0x0040c675", "info breakpoints", "continue", "detach", "quit", ], 35, prefix, )) probes.append(run_qemu_gdbstub_probe( "qemu gdbstub software breakpoint startup", [ "break *0x0040c675", "break *0x00542b0c", "info breakpoints", "continue", "detach", "quit", ], 35, prefix, )) if relocation_context.get("available"): mode1_source = relocated_point(relocation_context, "mode1SourceWriteWatch") opcode24_read = relocated_point(relocation_context, "opcode24Mode1Read") frontier_reader = relocated_point(relocation_context, "currentFrontierReader") if mode1_source and opcode24_read and frontier_reader: probes.append(run_qemu_gdbstub_probe( "qemu gdbstub relocated software breakpoint startup", [ f"x/4wx {opcode24_read}", f"x/4wx {frontier_reader}", f"break *{opcode24_read}", f"break *{frontier_reader}", "info breakpoints", "continue", "detach", "quit", ], 35, prefix, )) probes.append(run_qemu_gdbstub_probe( "qemu gdbstub virtual desktop relocated software breakpoint startup", [ f"x/4wx {opcode24_read}", f"x/4wx {frontier_reader}", f"break *{opcode24_read}", f"break *{frontier_reader}", "info breakpoints", "continue", "detach", "quit", ], 35, prefix, ["explorer", "/desktop=hwanse,640x480", "Hwanse2.exe"], )) probes.append(run_qemu_gdbstub_probe( "qemu gdbstub relocated watchpoint startup", [ f"x/bx {mode1_source}", f"watch *(unsigned char*){mode1_source}", f"break *{opcode24_read}", "info breakpoints", "continue", "detach", "quit", ], 35, prefix, )) probes.append(run_qemu_gdbstub_probe( "qemu gdbstub virtual desktop relocated watchpoint startup", [ f"x/bx {mode1_source}", f"watch *(unsigned char*){mode1_source}", f"break *{opcode24_read}", "info breakpoints", "continue", "detach", "quit", ], 35, prefix, ["explorer", "/desktop=hwanse,640x480", "Hwanse2.exe"], )) probe_map = {row["name"]: row for row in probes} wine_ok = probe_success(probe_map["wine version"]) # wineboot can leave qemu/X shutdown noise in stdout while still completing # the prefix update. Treat the process status as the control signal here. wineboot_ok = probe_map["wineboot"].get("status") == 0 and not probe_map["wineboot"].get("timedOut") notepad_attach_ok = probe_success(probe_map["notepad winedbg attach control"], allow_timeout=True) notepad_launch_ok = probe_success(probe_map["notepad winedbg launch control"], allow_timeout=True) exe_ok = probe_success(probe_map["Hwanse2.exe startup"], allow_timeout=True) virtual_desktop_ok = probe_success(probe_map["Hwanse2.exe virtual desktop startup"], allow_timeout=True) virtual_attach_ok = probe_success(probe_map["Hwanse2.exe virtual desktop attach"], allow_timeout=True) gdb_rows = [ row for row in probes if row["name"] in { "winedbg gdb startup", "winedbg gdb-multiarch startup", "winedbg native watchpoint startup", "qemu gdbstub hardware watchpoint startup", "qemu gdbstub software breakpoint startup", } or row["name"].startswith("qemu gdbstub relocated ") or row["name"].startswith("qemu gdbstub virtual desktop relocated ") ] qemu_gdbstub_connect = probe_map.get("qemu gdbstub connect control") gdb_ok = any(probe_success(row, allow_timeout=True) for row in gdb_rows) blockers = [] if not binfmt["registered"]: blockers.append("qemu-i386 binfmt is not registered, so this VM cannot exec i386 Wine transparently") if not wine_ok: blockers.append("wine --version does not execute successfully") if not wineboot_ok: blockers.append("wineboot cannot initialize the 32-bit Wine prefix") if not notepad_attach_ok: blockers.append("WineDbg attach fails even for a notepad control process under qemu-i386") if not notepad_launch_ok: blockers.append("WineDbg launch crashes even for a notepad control process under qemu-i386") if not exe_ok: blockers.append("Hwanse2.exe startup crashes or exits before a useful route trace can be captured") if not virtual_desktop_ok: blockers.append("Hwanse2.exe does not stay alive in Wine virtual desktop mode") if not virtual_attach_ok: blockers.append("Hwanse2.exe stays alive in virtual desktop mode, but WineDbg attach/watchpoint still fails") if not gdb_ok: blockers.append("no Wine debugger path reaches a stable watchpoint-capable session") if qemu_gdbstub_connect and probe_success(qemu_gdbstub_connect) and not any( probe_success(row, allow_timeout=True) for row in gdb_rows if row["name"].startswith("qemu gdbstub ") ): blockers.append("QEMU gdbstub connects, but route watchpoint/breakpoint insertion is not stable") can_capture = not blockers return { "objective": "bounded runtime execution probe for Hwanse2.exe trace capture", "winePrefix": str(prefix), "alsaConfigPath": str(NULL_ALSA_CONFIG) if NULL_ALSA_CONFIG.exists() else None, "wineAudioDriver": read_wine_audio_driver(prefix), "gdbMultiarchPath": gdb_multiarch, "qemuI386Binfmt": binfmt, "relocationContext": relocation_context, "probes": probes, "canCaptureTraceNow": can_capture, "promotionStatus": "blocked" if not can_capture else "trace-capture-ready-proof-missing", "blockers": blockers, "proofFound": False, "runtimeTraceExecutionProofFound": False, "failedRuntimeTraceExecutionGateIds": FAILED_RUNTIME_TRACE_EXECUTION_GATE_IDS, "missingEvidence": RUNTIME_TRACE_EXECUTION_MISSING_EVIDENCE, "evidenceRefs": RUNTIME_TRACE_EXECUTION_EVIDENCE_REFS, "evidenceRefCount": len(RUNTIME_TRACE_EXECUTION_EVIDENCE_REFS), "conclusion": ( "The VM can start the required Wine debugger path." if can_capture else "The Wine packages are installed, but the current VM still cannot capture the route trace reliably." ), } def write_outputs(summary: dict, out_dir: Path = OUT) -> None: out_dir.mkdir(parents=True, exist_ok=True) (out_dir / "runtime_trace_execution_probe.json").write_text( json.dumps(summary, ensure_ascii=False, separators=(",", ":")) + "\n", encoding="utf-8", ) def main() -> None: parser = argparse.ArgumentParser(description=__doc__) 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.prefix) write_outputs(summary, args.out_dir) print( "wrote runtime trace execution probe -> " f"{args.out_dir / 'runtime_trace_execution_probe.json'} " f"(can capture={summary['canCaptureTraceNow']})" ) if __name__ == "__main__": main()