diff --git a/tests/test_havoc_mcp_server.py b/tests/test_havoc_mcp_server.py new file mode 100644 index 0000000..816fd83 --- /dev/null +++ b/tests/test_havoc_mcp_server.py @@ -0,0 +1,40 @@ +import json +import socket +import threading +import pytest + +from tools.havoc_mcp_server import DebugAgentClient, DebugAgentError + + +class _StubAgentServer: + """A minimal in-process stand-in for tools/havoc_debug_agent_server.py, so this + test never touches ctypes or a real Windows process.""" + + def __init__(self, response: dict): + self._response = response + self._sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) + self._sock.bind(("127.0.0.1", 0)) + self._sock.listen(1) + self.port = self._sock.getsockname()[1] + self._thread = threading.Thread(target=self._serve_one, daemon=True) + self._thread.start() + + def _serve_one(self): + conn, _ = self._sock.accept() + conn.recv(4096) + conn.sendall((json.dumps(self._response) + "\n").encode("utf-8")) + conn.close() + + +def test_client_returns_result_on_success(): + stub = _StubAgentServer({"ok": True, "result": {"status": "running"}}) + client = DebugAgentClient(host="127.0.0.1", port=stub.port) + result = client.call("get_status", {}) + assert result == {"status": "running"} + + +def test_client_raises_on_error_response(): + stub = _StubAgentServer({"ok": False, "error": "all 4 hardware debug registers (DR0-DR3) are in use"}) + client = DebugAgentClient(host="127.0.0.1", port=stub.port) + with pytest.raises(DebugAgentError, match="DR0-DR3"): + client.call("set_watchpoint", {"addr": "0x5000", "size": 4, "mode": "write"}) diff --git a/tools/havoc_mcp_server.py b/tools/havoc_mcp_server.py new file mode 100644 index 0000000..659ba21 --- /dev/null +++ b/tools/havoc_mcp_server.py @@ -0,0 +1,36 @@ +"""Thin MCP server translating MCP tool calls into calls to the local debug agent. +No Win32/ctypes dependency of its own -- runs and is tested on any platform, though at +runtime it must reach the real debug agent server (tools/havoc_debug_agent_server.py) +running on the same Caspian box.""" + +from __future__ import annotations + +import json +import os +import socket + +from tools.havoc_debug_protocol import decode_message, encode_message + + +class DebugAgentError(Exception): + """Raised when the debug agent reports {"ok": false}.""" + + +class DebugAgentClient: + def __init__(self, host: str, port: int) -> None: + self._host = host + self._port = port + + def call(self, op: str, params: dict) -> dict: + with socket.create_connection((self._host, self._port), timeout=60) as sock: + sock.sendall(encode_message(op, params)) + buf = b"" + while not buf.endswith(b"\n"): + chunk = sock.recv(65536) + if not chunk: + break + buf += chunk + response = decode_message(buf) + if not response.get("ok"): + raise DebugAgentError(response.get("error", "unknown debug agent error")) + return response.get("result", {})