pyr0ball
a77db1edcb
DebugAgentClient.call() hardcoded a 60s socket timeout that also bounded recv(), so any continue_execution call asking for a longer wait (e.g. waiting for a user to alt-tab) would time out on the client side while the debug agent was still legitimately waiting. call() now accepts an optional socket_timeout override, and the continue_execution tool passes timeout + 30 so the client socket outlives the agent's own timeout + 10s of headroom. Also adds a round-trip test for encode_message/decode_message (the wire contract between the two independently-built sides of this protocol), and corrects the WatchpointSlots docstring, which overstated what Wow64SetThreadContext does -- the agent actually arms the hardware watchpoints via a native x64 SetThreadContext call, with the Wow64 variant used only for 32-bit register readback. Claude-Session: https://claude.ai/code/session_01E4YCRPfAE3328FcYPyKu9T
109 lines
4 KiB
Python
109 lines
4 KiB
Python
"""Wire protocol and pure-logic helpers shared between the debug agent and MCP server.
|
|
No ctypes, no Windows dependency -- runs and tests on any platform."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import operator
|
|
from typing import Callable, Optional
|
|
|
|
_OPS = {
|
|
"==": operator.eq,
|
|
"!=": operator.ne,
|
|
"<": operator.lt,
|
|
">": operator.gt,
|
|
"<=": operator.le,
|
|
">=": operator.ge,
|
|
}
|
|
|
|
|
|
def _parse_memory_expr(expr: str) -> int:
|
|
"""Parse "0x47eb48+0x18" or "0x47eb48" into an absolute address."""
|
|
if "+" in expr:
|
|
base_str, offset_str = expr.split("+", 1)
|
|
return int(base_str.strip(), 16) + int(offset_str.strip(), 16)
|
|
return int(expr.strip(), 16)
|
|
|
|
|
|
def evaluate_condition(
|
|
condition: dict,
|
|
registers: dict,
|
|
read_memory: Optional[Callable[[int, int], bytes]],
|
|
) -> bool:
|
|
"""Evaluate a conditional-breakpoint expression against a register snapshot and/or
|
|
memory reader. `condition` is either {"register": name, "op": ..., "value": hexstr}
|
|
or {"memory": expr, "size": n, "op": ..., "value": hexstr}."""
|
|
op_fn = _OPS[condition["op"]]
|
|
expected = int(condition["value"], 16)
|
|
|
|
if "register" in condition:
|
|
actual = registers[condition["register"]]
|
|
elif "memory" in condition:
|
|
addr = _parse_memory_expr(condition["memory"])
|
|
size = condition["size"]
|
|
raw = read_memory(addr, size)
|
|
actual = int.from_bytes(raw, "little")
|
|
else:
|
|
raise ValueError(f"condition must have 'register' or 'memory': {condition}")
|
|
|
|
return op_fn(actual, expected)
|
|
|
|
|
|
def encode_message(op: str, params: dict) -> bytes:
|
|
"""Encode a request as a single NDJSON line."""
|
|
return (json.dumps({"op": op, "params": params}) + "\n").encode("utf-8")
|
|
|
|
|
|
def decode_message(line: bytes) -> dict:
|
|
"""Decode a single NDJSON line (request or response) back into a dict."""
|
|
return json.loads(line.decode("utf-8"))
|
|
|
|
|
|
class WatchpointLimitError(Exception):
|
|
"""Raised when all 4 hardware debug address registers are already in use."""
|
|
|
|
|
|
_MODE_RW_BITS = {"write": 0b01, "readwrite": 0b11}
|
|
_SIZE_LEN_BITS = {1: 0b00, 2: 0b01, 4: 0b11}
|
|
|
|
|
|
class WatchpointSlots:
|
|
"""Tracks which of the 4 x86 hardware debug address registers (DR0-DR3) are in use
|
|
and computes the DR7 control register value. Pure bookkeeping -- does not touch
|
|
ctypes or an actual thread context; the debug agent applies compute_dr7()'s result
|
|
(and the DR0-DR3 addresses from addresses()) to the target thread via a native
|
|
x64 SetThreadContext call, which is what actually arms the hardware watchpoints.
|
|
Wow64SetThreadContext/Wow64GetThreadContext are used only for WOW64 (32-bit)
|
|
register readback, not for arming DR0-DR3/DR7 themselves."""
|
|
|
|
def __init__(self) -> None:
|
|
self._slots: dict[int, tuple[int, int, str]] = {} # slot -> (addr, size, mode)
|
|
|
|
def allocate(self, addr: int, size: int, mode: str) -> int:
|
|
if mode not in _MODE_RW_BITS:
|
|
raise ValueError(f"mode must be 'write' or 'readwrite', got {mode!r}")
|
|
if size not in _SIZE_LEN_BITS:
|
|
raise ValueError(f"size must be 1, 2, or 4 bytes, got {size}")
|
|
for slot in range(4):
|
|
if slot not in self._slots:
|
|
self._slots[slot] = (addr, size, mode)
|
|
return slot
|
|
raise WatchpointLimitError("all 4 hardware debug registers (DR0-DR3) are in use")
|
|
|
|
def release(self, slot: int) -> None:
|
|
self._slots.pop(slot, None)
|
|
|
|
def addresses(self) -> dict[int, int]:
|
|
"""slot -> address, for setting DR0-DR3 themselves."""
|
|
return {slot: addr for slot, (addr, _size, _mode) in self._slots.items()}
|
|
|
|
def compute_dr7(self) -> int:
|
|
dr7 = 0
|
|
for slot, (_addr, size, mode) in self._slots.items():
|
|
dr7 |= 1 << (slot * 2) # local enable bit (L0=bit0, L1=bit2, L2=bit4, L3=bit6)
|
|
rw_bits = _MODE_RW_BITS[mode]
|
|
len_bits = _SIZE_LEN_BITS[size]
|
|
shift = 16 + slot * 4
|
|
dr7 |= rw_bits << shift
|
|
dr7 |= len_bits << (shift + 2)
|
|
return dr7
|