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 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01TN4Ytn3gdWRonNmHpisWQv

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"}

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()