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
98 lines
4 KiB
Python
98 lines
4 KiB
Python
import pytest
|
|
from tools.havoc_debug_protocol import decode_message, encode_message, evaluate_condition
|
|
|
|
def test_encode_decode_message_round_trip():
|
|
# encode_message/decode_message are the literal wire contract between the
|
|
# Windows-side debug agent and this Linux-side MCP server, built independently on
|
|
# two different machines -- prove they actually agree with each other.
|
|
op = "set_watchpoint"
|
|
params = {"addr": "0x47eb48", "size": 4, "mode": "write"}
|
|
assert decode_message(encode_message(op, params)) == {"op": op, "params": params}
|
|
|
|
def test_register_condition_true():
|
|
registers = {"eax": 0x50}
|
|
condition = {"register": "eax", "op": "!=", "value": "0x47"}
|
|
assert evaluate_condition(condition, registers, read_memory=None) is True
|
|
|
|
def test_register_condition_false():
|
|
registers = {"eax": 0x47}
|
|
condition = {"register": "eax", "op": "!=", "value": "0x47"}
|
|
assert evaluate_condition(condition, registers, read_memory=None) is False
|
|
|
|
def test_register_condition_equals():
|
|
registers = {"eip": 0x436070}
|
|
condition = {"register": "eip", "op": "==", "value": "0x436070"}
|
|
assert evaluate_condition(condition, registers, read_memory=None) is True
|
|
|
|
def test_memory_condition_with_offset():
|
|
def fake_read_memory(addr, size):
|
|
assert addr == 0x47eb48 + 0x18
|
|
assert size == 4
|
|
return (0x47).to_bytes(4, "little")
|
|
condition = {"memory": "0x47eb48+0x18", "size": 4, "op": "==", "value": "0x47"}
|
|
assert evaluate_condition(condition, {}, fake_read_memory) is True
|
|
|
|
def test_memory_condition_no_offset():
|
|
def fake_read_memory(addr, size):
|
|
assert addr == 0x47da88
|
|
return (0).to_bytes(size, "little")
|
|
condition = {"memory": "0x47da88", "size": 4, "op": "==", "value": "0x0"}
|
|
assert evaluate_condition(condition, {}, fake_read_memory) is True
|
|
|
|
def test_comparison_operators():
|
|
registers = {"ecx": 10}
|
|
assert evaluate_condition({"register": "ecx", "op": "<", "value": "0xb"}, registers, None)
|
|
assert evaluate_condition({"register": "ecx", "op": ">", "value": "0x9"}, registers, None)
|
|
assert evaluate_condition({"register": "ecx", "op": "<=", "value": "0xa"}, registers, None)
|
|
assert evaluate_condition({"register": "ecx", "op": ">=", "value": "0xa"}, registers, None)
|
|
|
|
def test_unknown_register_raises():
|
|
with pytest.raises(KeyError):
|
|
evaluate_condition({"register": "bogus", "op": "==", "value": "0x0"}, {}, None)
|
|
|
|
from tools.havoc_debug_protocol import WatchpointSlots, WatchpointLimitError
|
|
|
|
def test_allocate_returns_free_slots_in_order():
|
|
slots = WatchpointSlots()
|
|
assert slots.allocate(0x47c48c, 4, "write") == 0
|
|
assert slots.allocate(0x47da88, 4, "readwrite") == 1
|
|
|
|
def test_allocate_raises_past_four():
|
|
slots = WatchpointSlots()
|
|
slots.allocate(0x1000, 4, "write")
|
|
slots.allocate(0x2000, 4, "write")
|
|
slots.allocate(0x3000, 4, "write")
|
|
slots.allocate(0x4000, 4, "write")
|
|
with pytest.raises(WatchpointLimitError):
|
|
slots.allocate(0x5000, 4, "write")
|
|
|
|
def test_release_frees_slot_for_reuse():
|
|
slots = WatchpointSlots()
|
|
slot0 = slots.allocate(0x1000, 4, "write")
|
|
slots.release(slot0)
|
|
assert slots.allocate(0x2000, 4, "write") == 0
|
|
|
|
def test_compute_dr7_enables_local_bit_per_active_slot():
|
|
slots = WatchpointSlots()
|
|
slots.allocate(0x1000, 4, "write") # slot 0 -> L0 = bit 0
|
|
dr7 = slots.compute_dr7()
|
|
assert dr7 & 0b1 == 0b1 # L0 enabled
|
|
assert dr7 & 0b100 == 0 # L1 not enabled
|
|
|
|
def test_compute_dr7_encodes_write_mode_and_4byte_len():
|
|
slots = WatchpointSlots()
|
|
slots.allocate(0x1000, 4, "write")
|
|
dr7 = slots.compute_dr7()
|
|
rw0 = (dr7 >> 16) & 0b11
|
|
len0 = (dr7 >> 18) & 0b11
|
|
assert rw0 == 0b01 # write-only
|
|
assert len0 == 0b11 # 4 bytes
|
|
|
|
def test_compute_dr7_encodes_readwrite_and_1byte_len():
|
|
slots = WatchpointSlots()
|
|
slots.allocate(0x2000, 1, "readwrite")
|
|
dr7 = slots.compute_dr7()
|
|
rw0 = (dr7 >> 16) & 0b11
|
|
len0 = (dr7 >> 18) & 0b11
|
|
assert rw0 == 0b11 # read/write
|
|
assert len0 == 0b00 # 1 byte
|