From 9360d1ea7a293f8081e5dcab8cad3558d95c0640 Mon Sep 17 00:00:00 2001 From: pyr0ball Date: Sat, 4 Jul 2026 09:17:36 -0700 Subject: [PATCH] feat: add Win32 debug agent with breakpoint/register/memory support Implements DebugAgent class (Task 2) with launch, set/remove breakpoint, get_registers, read/write_memory, continue_execution, and get_status. Handles WOW64 WX86 exception codes (0x4000001F/0x4000001E), keeps debuggee suspended via _pending_event until caller resumes, and tracks stopped thread for correct register reads. Smoke-tested against live HAVOC_NOCD.EXE: stopped at breakpoint@0x40C8E0 with eip=0x40c8e0 confirmed. Co-Authored-By: Claude Sonnet 4.6 Claude-Session: https://claude.ai/code/session_01TN4Ytn3gdWRonNmHpisWQv --- tools/havoc_debug_agent.py | 275 ++++++++++++++++++++++++++ tools/havoc_debug_agent_smoke_test.py | 31 +++ 2 files changed, 306 insertions(+) create mode 100644 tools/havoc_debug_agent.py create mode 100644 tools/havoc_debug_agent_smoke_test.py diff --git a/tools/havoc_debug_agent.py b/tools/havoc_debug_agent.py new file mode 100644 index 0000000..941972b --- /dev/null +++ b/tools/havoc_debug_agent.py @@ -0,0 +1,275 @@ +"""Win32 debug agent for HAVOC_NOCD.EXE. Extends the ctypes debug-loop approach proven +in tools/dbg_path.py into a reusable, addressable class. Windows-only -- requires +ctypes.windll, must run on Caspian (or, later, under Wine) against the real process.""" + +from __future__ import annotations + +import ctypes +import time +from ctypes import wintypes, byref, sizeof, c_char +from typing import Optional + +from tools.havoc_debug_protocol import evaluate_condition, WatchpointSlots + +k = ctypes.windll.kernel32 + +DEBUG_ONLY_THIS_PROCESS = 0x00000002 +DBG_CONTINUE = 0x00010002 +DBG_EXCEPTION_NOT_HANDLED = 0x80010001 +EXCEPTION_BREAKPOINT = 0x80000003 +EXCEPTION_SINGLE_STEP = 0x80000004 +# In a WOW64 (32-bit) process debugged by a 64-bit debugger, INT3 breakpoints and trap-flag +# single-steps in the 32-bit code arrive as these WX86 status codes, NOT the native ones +# above. This only shows up against the real process (confirmed on HAVOC_NOCD.EXE: a 0xCC at +# 0x40C8E0 raised 0x4000001F, not 0x80000003). Match both everywhere a trap is handled. +STATUS_WX86_SINGLE_STEP = 0x4000001E +STATUS_WX86_BREAKPOINT = 0x4000001F +_BREAKPOINT_CODES = (EXCEPTION_BREAKPOINT, STATUS_WX86_BREAKPOINT) +_SINGLE_STEP_CODES = (EXCEPTION_SINGLE_STEP, STATUS_WX86_SINGLE_STEP) +_DEBUG_TRAP_CODES = _BREAKPOINT_CODES + _SINGLE_STEP_CODES +EXCEPTION_ACCESS_VIOLATION = 0xC0000005 +CREATE_PROCESS_DEBUG_EVENT = 3 +CREATE_THREAD_DEBUG_EVENT = 2 +EXIT_PROCESS_DEBUG_EVENT = 5 +LOAD_DLL_DEBUG_EVENT = 6 +EXCEPTION_DEBUG_EVENT = 1 +CONTEXT_FULL = 0x00010007 +TRAP_FLAG = 0x100 # EFlags bit 8: single-step after the next instruction + + +class STARTUPINFO(ctypes.Structure): + _fields_ = [("cb", wintypes.DWORD), ("lpReserved", wintypes.LPWSTR), + ("lpDesktop", wintypes.LPWSTR), ("lpTitle", wintypes.LPWSTR), + ("dwX", wintypes.DWORD), ("dwY", wintypes.DWORD), ("dwXSize", wintypes.DWORD), + ("dwYSize", wintypes.DWORD), ("dwXCountChars", wintypes.DWORD), + ("dwYCountChars", wintypes.DWORD), ("dwFillAttribute", wintypes.DWORD), + ("dwFlags", wintypes.DWORD), ("wShowWindow", wintypes.WORD), + ("cbReserved2", wintypes.WORD), ("lpReserved2", ctypes.c_void_p), + ("hStdInput", wintypes.HANDLE), ("hStdOutput", wintypes.HANDLE), + ("hStdError", wintypes.HANDLE)] + + +class PROCESS_INFORMATION(ctypes.Structure): + _fields_ = [("hProcess", wintypes.HANDLE), ("hThread", wintypes.HANDLE), + ("dwProcessId", wintypes.DWORD), ("dwThreadId", wintypes.DWORD)] + + +class EXCEPTION_RECORD(ctypes.Structure): + _fields_ = [("ExceptionCode", wintypes.DWORD), ("ExceptionFlags", wintypes.DWORD), + ("ExceptionRecord", ctypes.c_void_p), ("ExceptionAddress", ctypes.c_void_p), + ("NumberParameters", wintypes.DWORD), ("ExceptionInformation", ctypes.c_void_p * 15)] + + +class EXCEPTION_DEBUG_INFO(ctypes.Structure): + _fields_ = [("ExceptionRecord", EXCEPTION_RECORD), ("dwFirstChance", wintypes.DWORD)] + + +class DEBUG_EVENT(ctypes.Structure): + class _U(ctypes.Union): + _fields_ = [("Exception", EXCEPTION_DEBUG_INFO), ("raw", c_char * 160)] + _fields_ = [("dwDebugEventCode", wintypes.DWORD), ("dwProcessId", wintypes.DWORD), + ("dwThreadId", wintypes.DWORD), ("u", _U)] + + +class WOW64_FLOATING_SAVE_AREA(ctypes.Structure): + _fields_ = [("ControlWord", wintypes.DWORD), ("StatusWord", wintypes.DWORD), + ("TagWord", wintypes.DWORD), ("ErrorOffset", wintypes.DWORD), + ("ErrorSelector", wintypes.DWORD), ("DataOffset", wintypes.DWORD), + ("DataSelector", wintypes.DWORD), ("RegisterArea", c_char * 80), + ("Cr0NpxState", wintypes.DWORD)] + + +class WOW64_CONTEXT(ctypes.Structure): + _fields_ = [("ContextFlags", wintypes.DWORD), ("Dr0", wintypes.DWORD), + ("Dr1", wintypes.DWORD), ("Dr2", wintypes.DWORD), ("Dr3", wintypes.DWORD), + ("Dr6", wintypes.DWORD), ("Dr7", wintypes.DWORD), + ("FloatSave", WOW64_FLOATING_SAVE_AREA), ("SegGs", wintypes.DWORD), + ("SegFs", wintypes.DWORD), ("SegEs", wintypes.DWORD), ("SegDs", wintypes.DWORD), + ("Edi", wintypes.DWORD), ("Esi", wintypes.DWORD), ("Ebx", wintypes.DWORD), + ("Edx", wintypes.DWORD), ("Ecx", wintypes.DWORD), ("Eax", wintypes.DWORD), + ("Ebp", wintypes.DWORD), ("Eip", wintypes.DWORD), ("SegCs", wintypes.DWORD), + ("EFlags", wintypes.DWORD), ("Esp", wintypes.DWORD), ("SegSs", wintypes.DWORD), + ("ExtendedRegisters", c_char * 512)] + + +_REGISTER_NAMES = ["Eax", "Ebx", "Ecx", "Edx", "Esi", "Edi", "Ebp", "Esp", "Eip", "EFlags"] + + +class DebugAgent: + """Owns one debugged process. One DebugAgent instance per running game session.""" + + def __init__(self) -> None: + self._process_info: Optional[PROCESS_INFORMATION] = None + self._threads: dict[int, wintypes.HANDLE] = {} + self._breakpoints: dict[int, tuple[bytes, Optional[dict]]] = {} # addr -> (orig_byte, condition) + self._watchpoints = WatchpointSlots() + self._last_event: Optional[DEBUG_EVENT] = None + self._exited: Optional[int] = None # exit code, if the process has exited + self._stopped_thread_id: Optional[int] = None # thread that hit the last reported stop + # A debug event we received and reported as a stop but have NOT continued yet, so the + # debuggee stays suspended for inspection. (pid, tid); resumed on next continue_execution. + self._pending_event: Optional[tuple[int, int]] = None + # For single-stepping over a restored breakpoint so it can be re-armed (Task 3): + self._step_over: Optional[tuple[int, int]] = None # (addr_to_rearm, thread_id) + + def _read_process_memory(self, addr: int, length: int) -> bytes: + buf = (c_char * length)() + read = ctypes.c_size_t(0) + ok = k.ReadProcessMemory(self._process_info.hProcess, ctypes.c_void_p(addr), + buf, length, byref(read)) + if not ok: + raise OSError(f"ReadProcessMemory failed at 0x{addr:X}: {k.GetLastError()}") + return bytes(buf[: read.value]) + + def _write_process_memory(self, addr: int, data: bytes) -> None: + written = ctypes.c_size_t(0) + ok = k.WriteProcessMemory(self._process_info.hProcess, ctypes.c_void_p(addr), + data, len(data), byref(written)) + if not ok: + raise OSError(f"WriteProcessMemory failed at 0x{addr:X}: {k.GetLastError()}") + + def launch(self, path: str, cwd: str) -> None: + si = STARTUPINFO() + si.cb = sizeof(si) + pi = PROCESS_INFORMATION() + ok = k.CreateProcessA(path.encode(), None, None, None, False, + DEBUG_ONLY_THIS_PROCESS, None, cwd.encode(), byref(si), byref(pi)) + if not ok: + raise OSError(f"CreateProcess failed: {k.GetLastError()}") + self._process_info = pi + self._threads[pi.dwThreadId] = pi.hThread + + def set_breakpoint(self, addr: int, condition: Optional[dict] = None) -> None: + if addr in self._breakpoints: + orig_byte, _old = self._breakpoints[addr] + self._breakpoints[addr] = (orig_byte, condition) # update condition only + return + orig_byte = self._read_process_memory(addr, 1) + self._write_process_memory(addr, b"\xCC") + self._breakpoints[addr] = (orig_byte, condition) + + def remove_breakpoint(self, addr: int) -> None: + entry = self._breakpoints.pop(addr, None) + if entry is not None: + orig_byte, _condition = entry + self._write_process_memory(addr, orig_byte) + + def list_breakpoints(self) -> list[int]: + return sorted(self._breakpoints.keys()) + + def _get_context(self, thread_handle: wintypes.HANDLE) -> WOW64_CONTEXT: + ctx = WOW64_CONTEXT() + ctx.ContextFlags = CONTEXT_FULL + if not k.Wow64GetThreadContext(thread_handle, byref(ctx)): + raise OSError(f"Wow64GetThreadContext failed: {k.GetLastError()}") + return ctx + + def _set_context(self, thread_handle: wintypes.HANDLE, ctx: WOW64_CONTEXT) -> None: + ctx.ContextFlags = CONTEXT_FULL + if not k.Wow64SetThreadContext(thread_handle, byref(ctx)): + raise OSError(f"Wow64SetThreadContext failed: {k.GetLastError()}") + + def get_registers(self) -> dict: + if self._stopped_thread_id is not None and self._stopped_thread_id in self._threads: + thread_handle = self._threads[self._stopped_thread_id] + else: + thread_handle = next(iter(self._threads.values())) + ctx = self._get_context(thread_handle) + return {name.lower(): getattr(ctx, name) for name in _REGISTER_NAMES} + + def read_memory(self, addr: int, length: int) -> bytes: + return self._read_process_memory(addr, length) + + def write_memory(self, addr: int, data: bytes) -> None: + self._write_process_memory(addr, data) + + def continue_execution(self, timeout: float) -> dict: + if self._exited is not None: + return {"status": "exited", "reason": None, "exit_code": self._exited} + + # If we reported a stop last time, the debuggee is still suspended on that event. + # Resume it now before waiting for the next one. + if self._pending_event is not None: + pid, tid = self._pending_event + self._pending_event = None + k.ContinueDebugEvent(pid, tid, DBG_CONTINUE) + + evt = DEBUG_EVENT() + deadline = time.time() + timeout + while time.time() < deadline: + remaining_ms = max(1, int((deadline - time.time()) * 1000)) + if not k.WaitForDebugEvent(byref(evt), remaining_ms): + continue # timed out this poll; loop again until our own deadline + + code = evt.dwDebugEventCode + + if code == EXIT_PROCESS_DEBUG_EVENT: + # For an exit event the union holds EXIT_PROCESS_DEBUG_INFO; its first dword + # (dwExitCode) overlaps ExceptionRecord.ExceptionCode, so this reads the code. + self._exited = evt.u.Exception.ExceptionRecord.ExceptionCode & 0xFFFFFFFF + k.ContinueDebugEvent(evt.dwProcessId, evt.dwThreadId, DBG_CONTINUE) + return {"status": "exited", "reason": None, "exit_code": self._exited} + + if code == CREATE_THREAD_DEBUG_EVENT: + handle = k.OpenThread(0x1FFFFF, False, evt.dwThreadId) + if handle: + self._threads[evt.dwThreadId] = handle + elif code == EXCEPTION_DEBUG_EVENT: + result = self._handle_exception(evt) + if result is not None: + # Stop: leave the debuggee suspended on this event so registers/memory can + # be inspected. The next continue_execution() resumes it via _pending_event. + self._pending_event = (evt.dwProcessId, evt.dwThreadId) + return result + + # Not stopping: continue past this event. This deliberately swallows benign + # debugger-induced first-chance exceptions (e.g. STATUS_INVALID_HANDLE 0xC0000008, + # raised only because a debugger is attached, with no app handler). Passing + # DBG_EXCEPTION_NOT_HANDLED here kills the game on boot; DBG_CONTINUE lets it run + # as if undebugged (verified against the real process). + k.ContinueDebugEvent(evt.dwProcessId, evt.dwThreadId, DBG_CONTINUE) + + return {"status": "running", "reason": None, "exit_code": None} + + def _handle_exception(self, evt: DEBUG_EVENT) -> Optional[dict]: + """Returns a stop-reason dict if execution should actually stop and be reported, + or None if it should transparently resume (e.g. a conditional breakpoint whose + condition was false).""" + er = evt.u.Exception.ExceptionRecord + exc_code = er.ExceptionCode & 0xFFFFFFFF + addr = er.ExceptionAddress or 0 + thread_handle = self._threads.get(evt.dwThreadId) + + if exc_code in _BREAKPOINT_CODES and addr in self._breakpoints: + orig_byte, condition = self._breakpoints[addr] + ctx = self._get_context(thread_handle) + registers = {name.lower(): getattr(ctx, name) for name in _REGISTER_NAMES} + + should_stop = True + if condition is not None: + should_stop = evaluate_condition(condition, registers, self._read_process_memory) + + # Restore original byte and rewind EIP so the real instruction re-executes. + self._write_process_memory(addr, orig_byte) + ctx.Eip = addr + self._set_context(thread_handle, ctx) + # Task 3 re-arms this breakpoint by single-stepping past the restored + # instruction; Task 2 leaves it removed after the first hit. + + if should_stop: + self._stopped_thread_id = evt.dwThreadId + return {"status": "stopped", "reason": f"breakpoint@0x{addr:X}", "exit_code": None} + return None # condition false: caller's loop will ContinueDebugEvent and keep going + + if exc_code == EXCEPTION_ACCESS_VIOLATION: + self._stopped_thread_id = evt.dwThreadId + return {"status": "stopped", "reason": f"access_violation@0x{addr:X}", "exit_code": None} + + return None + + def get_status(self) -> dict: + if self._exited is not None: + return {"status": "exited", "exit_code": self._exited} + if self._process_info is None: + return {"status": "not_launched"} + return {"status": "running"} diff --git a/tools/havoc_debug_agent_smoke_test.py b/tools/havoc_debug_agent_smoke_test.py new file mode 100644 index 0000000..6c245b2 --- /dev/null +++ b/tools/havoc_debug_agent_smoke_test.py @@ -0,0 +1,31 @@ +"""Manual acceptance smoke tests for the Win32 debug agent. NOT a pytest file -- these +launch the real HAVOC_NOCD.EXE under the debugger and need a live Windows session. + +Run from the repo root: python -m tools.havoc_debug_agent_smoke_test +""" + +from tools.havoc_debug_agent import DebugAgent + +GAME = r"Z:\Development\devl\Havoc\HAVOC_NOCD.EXE" +CWD = r"Z:\Development\devl\Havoc" + + +def task2_breakpoint_smoke() -> None: + """Task 2: breakpoint at busywait_frame_limiter_60hz (0x40C8E0) should trip during the + boot palette-fade / state transition and report eip == 0x40C8E0.""" + agent = DebugAgent() + agent.launch(GAME, CWD) + agent.set_breakpoint(0x40C8E0) + result = agent.continue_execution(timeout=30) + print("continue_execution ->", result) + if result["status"] == "stopped": + regs = agent.get_registers() + print("registers ->", {r: hex(v) for r, v in regs.items()}) + assert regs["eip"] == 0x40C8E0, f"expected eip 0x40C8E0, got {regs['eip']:#x}" + print("TASK 2 SMOKE: PASS") + else: + print(f"TASK 2 SMOKE: did not stop at breakpoint (status={result['status']})") + + +if __name__ == "__main__": + task2_breakpoint_smoke()