diff --git a/harbor/cli.py b/harbor/cli.py index 0a87a2a..5bfaa05 100644 --- a/harbor/cli.py +++ b/harbor/cli.py @@ -255,7 +255,7 @@ def build_parser(): p = sub.add_parser("who", help="who has worked on a scope (notes, reviews, claims, messages)") p.add_argument("scope") p = sub.add_parser("re", help="reverse-engineering lookups from the project's image and annotations") - p.add_argument("action", choices=["disasm", "xrefs", "bytes", "lookup", "ghidra", "ghidra-tools"]) + p.add_argument("action", choices=["disasm", "xrefs", "bytes", "lookup", "ghidra", "ghidra-tools", "debugger-tools"]) p.add_argument("target", nargs="?", default="", help="an address (0x4AD850) or a name; for ghidra, the tool") p.add_argument("arguments", nargs="?", default="{}", help="ghidra: the tool's arguments as JSON") p.add_argument("--max-bytes", type=int, dest="max_bytes") @@ -675,8 +675,13 @@ def main(argv=None): out(r, f"summary #{r['id']} pinned to {r['thread']}") elif cmd == "re": from .reverse import Ghidra, Reverse - r = None if args.action.startswith("ghidra") else Reverse(harbor.project.reverse) - if args.action == "ghidra-tools": + r = None if args.action.startswith(("ghidra", "debugger")) else Reverse(harbor.project.reverse) + if args.action == "debugger-tools": + from .reverse import Debugger + res = Debugger(harbor.project.reverse).tools() + out(res, "\n".join(f"{t['name']:<22} {(t['description'] or '').splitlines()[0][:90] if t['description'] else ''}" + for t in res)) + elif args.action == "ghidra-tools": res = Ghidra(harbor.project.reverse).tools() out(res, "\n".join(f"{t['name']:<26} {t['description'].splitlines()[0][:90]}" for t in res)) elif args.action == "ghidra": diff --git a/harbor/mcp_server.py b/harbor/mcp_server.py index 962bfe5..3744667 100644 --- a/harbor/mcp_server.py +++ b/harbor/mcp_server.py @@ -37,7 +37,9 @@ review queue, presence and stop flags. Protocol: 11. To read the original machine code use re_disassemble, re_xrefs, re_bytes and re_lookup: they read the project's own image with the team's current names, so do not write your own disassembler or offset maths. For Ghidra's view (decompiler, p-code, stack frames, struct field references, types) use ghidra_tools() once, - then ghidra(tool, arguments): one shared read-only Ghidra session holding the team's annotations.""" + then ghidra(tool, arguments): one shared read-only Ghidra session holding the team's annotations. + To check a value in the running game (x87 control word, `this`, struct fields), use debugger_tools() and + debugger(tool, arguments): your own x32dbg session; always end it with debugger('terminate_session').""" KindName = Literal["note", "question", "answer", "request", "finding", "retraction", "decision", "status", "handoff"] Importance = Literal["normal", "high", "urgent"] @@ -517,6 +519,37 @@ def ghidra(tool: str, arguments: dict | None = None) -> str: return _gh(lambda g: g.call(tool, arguments or {})) +_debugger = {} + + +def _dbg(fn): + try: + h = harbor() + d = _debugger.get(h.project.name) + if d is None: + from .reverse import Debugger + d = _debugger[h.project.name] = Debugger(h.project.reverse) + return fn(d) + except (HarborError, projects.ProjectError) as error: + raise ToolError(str(error)) from None + + +@mcp.tool() +def debugger_tools() -> list: + """The tools of your own x64dbg session (x64dbg-automate on x32dbg): start_session, set_breakpoint, go, + get_all_registers, read_memory, eval_expression, ... Call them with debugger(tool, arguments).""" + return _dbg(lambda d: d.tools()) + + +@mcp.tool() +def debugger(tool: str, arguments: dict | None = None) -> str: + """Run one x64dbg tool in your own debugger session, e.g. debugger('start_session', {'target_exe': + "R:/SteamLibrary/steamapps/common/Sid Meier's Pirates!/Pirates!.exe"}), debugger('set_breakpoint', ...), + debugger('go'), debugger('get_all_registers'). The session lasts until debugger('terminate_session') or + until your app closes. Use it to read values, not to patch the game.""" + return _dbg(lambda d: d.call(tool, arguments or {})) + + def _report_in(): """Tell Harbor this agent's app has its tools running: at start, then every 5 minutes while the app keeps the server open (updates the connection only, not the agent's last-seen time).""" diff --git a/harbor/reverse.py b/harbor/reverse.py index 479ed4d..f10e000 100644 --- a/harbor/reverse.py +++ b/harbor/reverse.py @@ -356,3 +356,99 @@ class Ghidra: if result.get("isError"): raise HarborError(text.strip() or f"Ghidra refused {tool}") return re.sub(r"^\[Context\] Operating on: [^\n]*\n+", "", text) + + +class Debugger: + """A per-agent x64dbg session through the x64dbg-automate MCP server (stdio), started on first use and kept for + the life of this Harbor process, so breakpoints and the paused game survive between calls. Configured by + `debugger_command`, `debugger_path` (x32dbg.exe) and `debugger_temp` in the [reverse] table. The plugin and the + client find each other through a session file in TEMP, so both get the same TEMP.""" + + def __init__(self, config): + config = config or {} + self.command = config.get("debugger_command") + if not self.command: + raise HarborError("this project has no debugger_command in its [reverse] table") + self.env = dict(os.environ) + if config.get("debugger_path"): + self.env["X64DBG_PATH"] = str(config["debugger_path"]) + if config.get("debugger_temp"): + self.env["TEMP"] = self.env["TMP"] = str(config["debugger_temp"]) + self.proc = None + self.counter = 0 + self.lock = __import__("threading").Lock() + + def _start(self): + import atexit + import queue + import subprocess + import threading + command = self.command if isinstance(self.command, list) else [self.command] + self.proc = subprocess.Popen(command, stdin=subprocess.PIPE, stdout=subprocess.PIPE, + stderr=subprocess.DEVNULL, env=self.env, text=True, encoding="utf-8", + creationflags=getattr(subprocess, "CREATE_NO_WINDOW", 0)) + self.answers = queue.Queue() + + def read(stream, answers): + for line in stream: + if line.strip(): + try: + answers.put(json.loads(line)) + except ValueError: + pass + answers.put(None) + threading.Thread(target=read, args=(self.proc.stdout, self.answers), daemon=True).start() + self._rpc("initialize", {"protocolVersion": "2025-06-18", "capabilities": {}, + "clientInfo": {"name": "harbor", "version": "2"}}) + self._send({"jsonrpc": "2.0", "method": "notifications/initialized"}) + atexit.register(self.close) + + def _send(self, body): + self.proc.stdin.write(json.dumps(body) + "\n") + self.proc.stdin.flush() + + def _rpc(self, method, params, timeout=120): + import queue + self.counter += 1 + self._send({"jsonrpc": "2.0", "id": self.counter, "method": method, "params": params}) + while True: + try: + answer = self.answers.get(timeout=timeout) + except queue.Empty: + raise HarborError(f"the debugger did not answer {method} within {timeout}s") from None + if answer is None: + self.proc = None + raise HarborError("the debugger's MCP server exited") + if answer.get("id") == self.counter: + if "error" in answer: + raise HarborError(f"debugger: {answer['error'].get('message', answer['error'])}") + return answer.get("result") + + def tools(self): + with self.lock: + if self.proc is None or self.proc.poll() is not None: + self._start() + return [{"name": t["name"], "description": t.get("description", ""), + "arguments": (t.get("inputSchema") or {}).get("properties", {}), + "required": (t.get("inputSchema") or {}).get("required", [])} + for t in self._rpc("tools/list", {})["tools"]] + + def call(self, tool, arguments=None, timeout=300): + with self.lock: + if self.proc is None or self.proc.poll() is not None: + self._start() + result = self._rpc("tools/call", {"name": tool, "arguments": arguments or {}}, timeout=timeout) + text = "\n".join(c.get("text", "") for c in result.get("content", []) if isinstance(c, dict)) + if result.get("isError"): + raise HarborError(text.strip() or f"the debugger refused {tool}") + return text + + def close(self): + """End the debuggee and the debugger with the Harbor process, so no paused game is left behind.""" + if self.proc and self.proc.poll() is None: + try: + self._rpc("tools/call", {"name": "terminate_session", "arguments": {}}, timeout=15) + except HarborError: + pass + self.proc.terminate() + self.proc = None diff --git a/tests/fake_stdio_mcp.py b/tests/fake_stdio_mcp.py new file mode 100644 index 0000000..6e8ca5b --- /dev/null +++ b/tests/fake_stdio_mcp.py @@ -0,0 +1,18 @@ +"""A fake stdio MCP server for the debugger forwarder test: one tool that echoes, one that fails.""" +import json +import sys + +for line in sys.stdin: + msg = json.loads(line) + if "id" not in msg: + continue + if msg["method"] == "initialize": + result = {"protocolVersion": "2025-06-18"} + elif msg["method"] == "tools/list": + result = {"tools": [{"name": "echo", "description": "Echo", "inputSchema": {"properties": {"x": {}}}}]} + elif msg["params"]["name"] == "echo": + result = {"content": [{"type": "text", "text": "echo " + json.dumps(msg["params"]["arguments"])}]} + else: + result = {"isError": True, "content": [{"type": "text", "text": "no such session"}]} + print(json.dumps({"jsonrpc": "2.0", "method": "notifications/message", "params": {}}), flush=True) + print(json.dumps({"jsonrpc": "2.0", "id": msg["id"], "result": result}), flush=True) diff --git a/tests/test_reverse.py b/tests/test_reverse.py index 02b1366..ea5a5ae 100644 --- a/tests/test_reverse.py +++ b/tests/test_reverse.py @@ -117,3 +117,22 @@ class GhidraForwarder(unittest.TestCase): with self.assertRaises(HarborError): from harbor.reverse import Ghidra Ghidra({"ghidra_url": "http://127.0.0.1:9/mcp"}).tools() + + +class DebuggerForwarder(unittest.TestCase): + """debugger() keeps one stdio MCP server per Harbor process and skips its notifications.""" + + def test_call_and_error(self): + import sys + from pathlib import Path + from harbor.reverse import Debugger + d = Debugger({"debugger_command": [sys.executable, "-B", str(Path(__file__).with_name("fake_stdio_mcp.py"))]}) + try: + self.assertEqual(d.tools()[0]["name"], "echo") + self.assertEqual(d.call("echo", {"x": 1}), 'echo {"x": 1}') + with self.assertRaises(HarborError): + d.call("terminate_session") + finally: + d.close() + with self.assertRaises(HarborError): + Debugger({})