pyr0ball/havoc-remaster
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

feat: add hardware watchpoints (DR0-DR3/DR7) and re-armable breakpoints to debug agent

Browse source 
Implements set_watchpoint/remove_watchpoint/list_watchpoints via Wow64SetThreadContext
on DR0-DR3/DR7. Skips 64-bit WOW64 helper threads (Wow64GetThreadContext returns
ERROR_ACCESS_DENIED on those -- found against the real process). Propagates active
watchpoints to new threads as they spawn. Re-armable breakpoints: on breakpoint hit,
restores original byte and sets EFlags trap flag; the resulting single-step re-plants
the 0xCC, making breakpoints persistent across multiple hits. EXCEPTION_SINGLE_STEP
handler distinguishes watchpoint trips (Dr6 bits 0-3 set) from re-arm single-steps
(Dr6 clear). Smoke test confirmed: re-armed breakpoint fires on second continue;
watchpoint on 0x0047C48C set on slot 0 -- full watchpoint trip validation requires
gameplay state==0x47 (deferred to Task 6 end-to-end run).

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01TN4Ytn3gdWRonNmHpisWQv
This commit is contained in:
pyr0ball 2026-07-04 09:26:37 -07:00
parent 9360d1ea7a
commit 12ec33b41a
2 changed files with 109 additions and 4 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

57
tools/havoc_debug_agent.py
View file

@ -157,6 +157,32 @@ class DebugAgent:
def list_breakpoints(self) -> list[int]:
return sorted(self._breakpoints.keys())
def _apply_watchpoints_to_thread(self, thread_handle: int) -> None:
"""Set DR0-DR3 and DR7 on one thread. Silently skips 64-bit WOW64 helper threads
(Wow64GetThreadContext returns ERROR_ACCESS_DENIED on those)."""
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_FULL
@ -214,6 +240,8 @@ class DebugAgent:
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:
@ -249,18 +277,41 @@ class DebugAgent:
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.
# 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)
# Task 3 re-arms this breakpoint by single-stepping past the restored
# instruction; Task 2 leaves it removed after the first hit.
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
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}

56
tools/havoc_debug_agent_smoke_test.py
View file

@ -27,5 +27,59 @@ def task2_breakpoint_smoke() -> None:
print(f"TASK 2 SMOKE: did not stop at breakpoint (status={result['status']})")
def task3_watchpoint_smoke() -> None:
"""Task 3: write-watchpoint on the delta-time baseline (0x0047c48c).
compute_frame_delta_ticks_60hz_min1_clamp is only called from player_cooldown_timers_tick
which only runs during gameplay (state==0x47). So: stop at the known boot breakpoint first
to confirm debugging works, set the watchpoint while paused, then continue. Requires the
user to start a race/level within the 90s window to trigger the write.
Expected EIP at watchpoint trip: inside 0x4092D0-0x409328."""
agent = DebugAgent()
agent.launch(GAME, CWD)
# Phase 1: confirm basic debugging is live by stopping at the boot-time frame limiter
agent.set_breakpoint(0x40C8E0)
r = agent.continue_execution(timeout=30)
if r["status"] != "stopped":
print(f"TASK 3 SMOKE FAIL: game did not hit boot breakpoint ({r})")
return
print("Phase 1: stopped at boot breakpoint OK")
# Phase 2: set the watchpoint while the thread is suspended (DR regs applied to known-live thread)
slot = agent.set_watchpoint(0x0047C48C, size=4, mode="write")
print(f"Phase 2: watchpoint set on 0x0047C48C -> slot {slot}")
# Phase 3: wait for gameplay to begin; user must start a level within the timeout window
print("Phase 3: continuing -- start a level to trigger compute_frame_delta_ticks_60hz_min1_clamp (90s window)")
result = agent.continue_execution(timeout=90)
print("continue_execution ->", result)
reason = result.get("reason", "")
if result["status"] == "stopped" and "watchpoint" in reason:
regs = agent.get_registers()
eip = regs["eip"]
print("EIP at watchpoint trip:", hex(eip))
assert 0x4092D0 <= eip <= 0x409328, (
f"expected eip inside compute_frame_delta_ticks_60hz_min1_clamp "
f"(0x4092D0-0x409328), got {eip:#x}"
)
print("TASK 3 SMOKE: PASS (watchpoint fired)")
elif result["status"] == "stopped" and "breakpoint" in reason:
# Re-armed breakpoint fired again -- confirms re-arming mechanism works.
# Watchpoint was set on slot 0 but needs gameplay (state==0x47) to trigger.
print(
f"TASK 3 SMOKE: re-armed breakpoint confirmed ({reason}), "
f"watchpoint set on slot {slot} but target requires gameplay state to be written. "
f"Re-run and start a level during the 90s window to validate watchpoint fire."
)
else:
print(
f"TASK 3 SMOKE: unexpected result (status={result['status']}, reason={reason})"
)
if __name__ == "__main__":
task2_breakpoint_smoke()
import sys
if len(sys.argv) > 1 and sys.argv[1] == "task3":
task3_watchpoint_smoke()
else:
task2_breakpoint_smoke()