pyr0ball
7d39ccddf0
Key fixes applied during Task 6 end-to-end MCP integration testing: - CONTEXT_ALL (CONTEXT_FULL | CONTEXT_DEBUG_REGISTERS) for Wow64Get/SetThreadContext so DR0-DR7 are read and written alongside general-purpose registers - Native x64 SetThreadContext path in _apply_watchpoints_to_thread: Wow64SetThreadContext updates the WOW64 context save area but does not propagate to hardware DR registers; the native path is required for hardware watchpoints to actually fire on WOW64 threads - _REGISTER_NAMES extended to include Dr0-Dr3, Dr6, Dr7 so get_registers returns them - launch() now terminates and drains the previously attached process before starting a new one, preventing stale debug events from a hung game from corrupting a new session - continue_execution filters events by process ID so late-arriving exit events from a previously terminated game do not trigger a false "exited" status for the new process - AV exception routing restored: EXCEPTION_ACCESS_VIOLATION stops and reports; callers skip boot-time AVs in a loop rather than silently continuing them in the agent - launch_with_breakpoints atomic op in agent server and MCP server - _DebugWorker dedicated thread: Windows debug API requires all WaitForDebugEvent, ContinueDebugEvent, and CreateProcessA calls to originate from the same OS thread; ThreadingTCPServer violated this -- replaced with single worker thread + queue - Gameplay screenshot captured at watchpoint stop (272 KB real frame) Acceptance checklist results: - Item 1 PASS: bp@0x40C8E0, eip/eax confirmed - Item 2 PASS: conditional breakpoint false-skip and true-fire verified - Item 3 PASS: watchpoint@slot0 fired at 0x40E998 (MOV [DAT_0047c488], EAX in player_cooldown_timers_tick -- the DAT write is in the caller, not inside compute_frame_delta_ticks_60hz_min1_clamp as the spec assumed) - Item 4 PASS: screenshot PNG captured (gameplay frame 272 KB) Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01TN4Ytn3gdWRonNmHpisWQv
411 lines
20 KiB
Python
411 lines
20 KiB
Python
"""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
|
|
CONTEXT_DEBUG_REGISTERS = 0x00010010
|
|
CONTEXT_ALL = CONTEXT_FULL | CONTEXT_DEBUG_REGISTERS # includes DR0-DR3, DR6, DR7
|
|
TRAP_FLAG = 0x100 # EFlags bit 8: single-step after the next instruction
|
|
# Native x64 context flag for debug registers (different base from WOW64 flags)
|
|
CONTEXT_AMD64_DEBUG = 0x00100010
|
|
# x64 CONTEXT size and field offsets (winnt.h layout, must be 16-byte aligned)
|
|
_CONTEXT64_SIZE = 0x4D0
|
|
_CTX64_FLAGS_OFF = 0x030 # DWORD ContextFlags
|
|
_CTX64_DR0_OFF = 0x048 # DWORD64 Dr0
|
|
_CTX64_DR6_OFF = 0x068 # DWORD64 Dr6
|
|
_CTX64_DR7_OFF = 0x070 # DWORD64 Dr7
|
|
|
|
|
|
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",
|
|
"Dr0", "Dr1", "Dr2", "Dr3", "Dr6", "Dr7"]
|
|
|
|
|
|
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:
|
|
# Terminate any previously-attached process. Without this, a game whose debug event
|
|
# loop was abandoned (e.g. after a timeout) stays hung with all threads suspended,
|
|
# and the new launch fails because the stale instance holds file/mutex locks.
|
|
if self._process_info is not None:
|
|
k.TerminateProcess(self._process_info.hProcess, 0)
|
|
# Drain the exit event so the debug subsystem releases the process cleanly.
|
|
evt = DEBUG_EVENT()
|
|
for _ in range(32):
|
|
if k.WaitForDebugEvent(byref(evt), 500):
|
|
k.ContinueDebugEvent(evt.dwProcessId, evt.dwThreadId, DBG_CONTINUE)
|
|
if evt.dwDebugEventCode == EXIT_PROCESS_DEBUG_EVENT:
|
|
break
|
|
k.CloseHandle(self._process_info.hProcess)
|
|
k.CloseHandle(self._process_info.hThread)
|
|
self._process_info = None
|
|
|
|
# Reset all process-local state so re-launching doesn't inherit stale breakpoints,
|
|
# thread handles, or watchpoints from a previous session on the same server instance.
|
|
self._threads.clear()
|
|
self._breakpoints.clear()
|
|
self._watchpoints = WatchpointSlots()
|
|
self._last_event = None
|
|
self._exited = None
|
|
self._stopped_thread_id = None
|
|
self._pending_event = None
|
|
self._step_over = 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 _set_native_dr_registers(self, thread_handle: int) -> None:
|
|
"""Set hardware DR registers via the native x64 SetThreadContext.
|
|
Wow64SetThreadContext only updates the WOW64 context save area; the actual
|
|
CPU hardware DR registers that trigger debug exceptions need the native path."""
|
|
# Allocate 16-byte aligned buffer for x64 CONTEXT (size 0x4D0)
|
|
raw = (ctypes.c_byte * (_CONTEXT64_SIZE + 15))()
|
|
base = ctypes.addressof(raw)
|
|
aligned = ctypes.c_void_p((base + 15) & ~15)
|
|
ctypes.memset(aligned, 0, _CONTEXT64_SIZE)
|
|
ctypes.c_uint32.from_address(aligned.value + _CTX64_FLAGS_OFF).value = CONTEXT_AMD64_DEBUG
|
|
# Read current native DR state (ignore failure on non-suspendable threads)
|
|
k.GetThreadContext(thread_handle, aligned)
|
|
# Write DR0-DR3 for each watchpoint slot
|
|
for slot, wp_addr in self._watchpoints.addresses().items():
|
|
ctypes.c_uint64.from_address(aligned.value + _CTX64_DR0_OFF + slot * 8).value = wp_addr
|
|
# Clear remaining DR slots
|
|
active_slots = set(self._watchpoints.addresses().keys())
|
|
for slot in range(4):
|
|
if slot not in active_slots:
|
|
ctypes.c_uint64.from_address(aligned.value + _CTX64_DR0_OFF + slot * 8).value = 0
|
|
ctypes.c_uint64.from_address(aligned.value + _CTX64_DR6_OFF).value = 0
|
|
ctypes.c_uint64.from_address(aligned.value + _CTX64_DR7_OFF).value = self._watchpoints.compute_dr7()
|
|
k.SetThreadContext(thread_handle, aligned)
|
|
|
|
def _apply_watchpoints_to_thread(self, thread_handle: int) -> None:
|
|
"""Set DR0-DR3 and DR7 on one thread. Uses both the WOW64 context (for Wow64GetThreadContext
|
|
readback) and the native x64 context (for actual hardware DR register effect on WOW64 threads).
|
|
Silently skips 64-bit WOW64 helper threads where Wow64GetThreadContext returns access denied."""
|
|
# Native x64 path: sets actual hardware DR registers (Wow64SetThreadContext alone is not enough)
|
|
self._set_native_dr_registers(thread_handle)
|
|
# WOW64 path: keeps WOW64_CONTEXT DR fields in sync for get_registers readback
|
|
try:
|
|
ctx = self._get_context(thread_handle)
|
|
except OSError:
|
|
return
|
|
for slot, wp_addr in self._watchpoints.addresses().items():
|
|
setattr(ctx, f"Dr{slot}", wp_addr)
|
|
ctx.Dr7 = self._watchpoints.compute_dr7()
|
|
self._set_context(thread_handle, ctx)
|
|
|
|
def set_watchpoint(self, addr: int, size: int, mode: str) -> int:
|
|
slot = self._watchpoints.allocate(addr, size, mode)
|
|
for thread_handle in self._threads.values():
|
|
self._apply_watchpoints_to_thread(thread_handle)
|
|
return slot
|
|
|
|
def remove_watchpoint(self, slot: int) -> None:
|
|
self._watchpoints.release(slot)
|
|
for thread_handle in self._threads.values():
|
|
self._apply_watchpoints_to_thread(thread_handle)
|
|
|
|
def list_watchpoints(self) -> dict:
|
|
return self._watchpoints.addresses()
|
|
|
|
def _get_context(self, thread_handle: wintypes.HANDLE) -> WOW64_CONTEXT:
|
|
ctx = WOW64_CONTEXT()
|
|
ctx.ContextFlags = CONTEXT_ALL
|
|
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_ALL
|
|
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
|
|
|
|
# Stale events from a previously-terminated process (drained but not fully gone)
|
|
# can arrive after a new launch. Ignore them to avoid misidentifying the new game.
|
|
if evt.dwProcessId != self._process_info.dwProcessId:
|
|
k.ContinueDebugEvent(evt.dwProcessId, evt.dwThreadId, DBG_CONTINUE)
|
|
continue
|
|
|
|
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
|
|
if self._watchpoints.addresses():
|
|
self._apply_watchpoints_to_thread(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
|
|
|
|
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, rewind EIP, and set trap flag to single-step past
|
|
# the restored instruction so we can re-arm the 0xCC afterwards.
|
|
self._write_process_memory(addr, orig_byte)
|
|
ctx.Eip = addr
|
|
ctx.EFlags |= TRAP_FLAG
|
|
self._set_context(thread_handle, ctx)
|
|
self._step_over = (addr, evt.dwThreadId)
|
|
|
|
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 in _SINGLE_STEP_CODES:
|
|
ctx = self._get_context(thread_handle)
|
|
# Dr6 bits 0-3: which of DR0-DR3 fired. If any are set this is a watchpoint hit;
|
|
# if Dr6 is clear it is a trap-flag single-step from breakpoint re-arm.
|
|
triggered_slots = [slot for slot in range(4) if ctx.Dr6 & (1 << slot)]
|
|
if triggered_slots:
|
|
ctx.Dr6 = 0 # clear condition bits so the next single-step starts clean
|
|
self._set_context(thread_handle, ctx)
|
|
self._stopped_thread_id = evt.dwThreadId
|
|
return {
|
|
"status": "stopped",
|
|
"reason": f"watchpoint@slot{triggered_slots[0]}",
|
|
"exit_code": None,
|
|
}
|
|
# Trap-flag single-step: re-arm the breakpoint we just stepped past.
|
|
if self._step_over is not None:
|
|
rearm_addr, _tid = self._step_over
|
|
self._step_over = None
|
|
if rearm_addr in self._breakpoints:
|
|
self._write_process_memory(rearm_addr, b"\xCC")
|
|
return None # transparent; do not stop
|
|
|
|
# AV: stop and report. The game loops on DBG_CONTINUE for its boot-time AVs, so the
|
|
# caller must explicitly continue past each one. Callers that don't care about AVs
|
|
# can loop: while r['reason'] and 'access_violation' in r['reason']: r = continue()
|
|
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 screenshot(self) -> bytes:
|
|
"""Capture the game window as PNG bytes. Requires Pillow (pip install pillow)."""
|
|
from PIL import ImageGrab # local import: only needed on Caspian, not for pure-logic tests
|
|
import io
|
|
|
|
hwnd = ctypes.windll.user32.FindWindowA(None, b"HAVOC(tm) by Reality Bytes")
|
|
if not hwnd:
|
|
raise OSError("game window not found -- is HAVOC_NOCD.EXE running and visible?")
|
|
rect = wintypes.RECT()
|
|
ctypes.windll.user32.GetWindowRect(hwnd, byref(rect))
|
|
img = ImageGrab.grab(bbox=(rect.left, rect.top, rect.right, rect.bottom))
|
|
buf = io.BytesIO()
|
|
img.save(buf, format="PNG")
|
|
return buf.getvalue()
|
|
|
|
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"}
|