diff --git a/harbor/cli.py b/harbor/cli.py index 310840d..0a87a2a 100644 --- a/harbor/cli.py +++ b/harbor/cli.py @@ -255,8 +255,9 @@ 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"]) - p.add_argument("target", help="an address (0x4AD850) or a name") + p.add_argument("action", choices=["disasm", "xrefs", "bytes", "lookup", "ghidra", "ghidra-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") p.add_argument("--count", type=int, default=16) p.add_argument("--as", dest="as_type", default="hex", choices=["hex", "int32", "uint32", "float", "double", "string"]) @@ -673,9 +674,15 @@ def main(argv=None): r = harbor.summarize(who(), args.thread, text) out(r, f"summary #{r['id']} pinned to {r['thread']}") elif cmd == "re": - from .reverse import Reverse - r = Reverse(harbor.project.reverse) - if args.action == "disasm": + from .reverse import Ghidra, Reverse + r = None if args.action.startswith("ghidra") else Reverse(harbor.project.reverse) + if 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": + text = Ghidra(harbor.project.reverse).call(args.target, json.loads(args.arguments)) + out({"result": text}, text) + elif args.action == "disasm": text = r.disassemble(args.target, args.max_bytes) out({"disassembly": text}, text) elif args.action == "xrefs": diff --git a/harbor/mcp_server.py b/harbor/mcp_server.py index 479aad7..962bfe5 100644 --- a/harbor/mcp_server.py +++ b/harbor/mcp_server.py @@ -35,7 +35,9 @@ review queue, presence and stop flags. Protocol: 9. A correction is send(..., corrects=). 10. Before your session ends or your allowance runs out, write handoff(): finished, in flight, next. 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.""" + 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.""" KindName = Literal["note", "question", "answer", "request", "finding", "retraction", "decision", "status", "handoff"] Importance = Literal["normal", "high", "urgent"] @@ -484,6 +486,37 @@ def re_lookup(target: str) -> dict: return _re(lambda r: r.lookup(target)) +_ghidra = {} + + +def _gh(fn): + try: + h = harbor() + g = _ghidra.get(h.project.name) + if g is None: + from .reverse import Ghidra + g = _ghidra[h.project.name] = Ghidra(h.project.reverse) + return fn(g) + except (HarborError, projects.ProjectError) as error: + raise ToolError(str(error)) from None + + +@mcp.tool() +def ghidra_tools() -> list: + """The tools of the team's shared Ghidra session (GhidrAssistMCP, read-only, the curated project with every + applied annotation): name, description and arguments. Call them with ghidra(tool, arguments).""" + return _gh(lambda g: g.tools()) + + +@mcp.tool() +def ghidra(tool: str, arguments: dict | None = None) -> str: + """Run one tool in the shared read-only Ghidra session, e.g. ghidra('get_code', {'function': 'GetFogLevel', + 'format': 'decompiler'}), ghidra('get_function_stack_layout', {'function_name_or_address': '0x4ADC70'}), + ghidra('struct', {'action': 'field_xrefs', ...}). Long decompiles are waited for. Renames, retypes and other + changes are refused: names and types change through annotations/ and land on main.""" + return _gh(lambda g: g.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 ef91bf6..479ed4d 100644 --- a/harbor/reverse.py +++ b/harbor/reverse.py @@ -276,3 +276,83 @@ class Reverse: out["file"] = {"path": str(path.relative_to(self.src.parent)).replace(os.sep, "/"), "status": status} break return out + + +class Ghidra: + """A small client for the shared, read-only Ghidra MCP server (GhidrAssistMCP over streamable HTTP), so every + agent reaches Ghidra through Harbor's own tools. Configured by `ghidra_url` in the [reverse] table.""" + + def __init__(self, config): + self.url = (config or {}).get("ghidra_url") + if not self.url: + raise HarborError("this project has no ghidra_url in its [reverse] table") + self.session = None + self.counter = 0 + + def _post(self, method, params=None, notify=False, timeout=120): + import urllib.error + import urllib.request + body = {"jsonrpc": "2.0", "method": method, **({"params": params} if params is not None else {})} + if not notify: + self.counter += 1 + body["id"] = self.counter + headers = {"Content-Type": "application/json", "Accept": "application/json, text/event-stream"} + if self.session: + headers["Mcp-Session-Id"] = self.session + request = urllib.request.Request(self.url, json.dumps(body).encode(), headers=headers, method="POST") + try: + with urllib.request.urlopen(request, timeout=timeout) as response: + self.session = response.headers.get("Mcp-Session-Id") or self.session + text = response.read().decode("utf-8", "replace") + except urllib.error.HTTPError as error: + if error.code in (400, 404) and self.session and method != "initialize": + self.session = None # the server restarted: start a new session once + self._initialize() + return self._post(method, params, notify, timeout) + raise HarborError(f"the Ghidra server answered {error.code}: {error.read()[:300]!r}") from None + except OSError as error: + raise HarborError(f"the shared Ghidra server at {self.url} is not answering ({error}); start it with " + "R:/Workspace/Tools/re/ghidra-serve.ps1, or ask the coordinator") from None + if notify or not text.strip(): + return None + if not text.lstrip().startswith("{"): + text = next((line[5:] for line in text.splitlines() if line.startswith("data:")), "{}") + answer = json.loads(text) + if "error" in answer: + raise HarborError(f"Ghidra: {answer['error'].get('message', answer['error'])}") + return answer.get("result") + + def _initialize(self): + self._post("initialize", {"protocolVersion": "2025-06-18", "capabilities": {}, + "clientInfo": {"name": "harbor", "version": "2"}}) + self._post("notifications/initialized", notify=True) + + def _ensure(self): + if not self.session: + self._initialize() + + def tools(self): + """The server's tools: name, description and argument schema.""" + self._ensure() + 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._post("tools/list")["tools"]] + + def call(self, tool, arguments=None, wait=180): + """Call one tool; long-running ones (the decompiler) are polled until they finish or `wait` seconds pass.""" + import time + self._ensure() + result = self._post("tools/call", {"name": tool, "arguments": arguments or {}}, timeout=wait + 30) + text = "\n".join(c.get("text", "") for c in result.get("content", []) if isinstance(c, dict)) + task = re.search(r"Task ID: ([0-9a-f-]{36})", text) if "async execution" in text else None + deadline = time.monotonic() + wait + while task and time.monotonic() < deadline: + time.sleep(1) + result = self._post("tools/call", {"name": "get_task_status", "arguments": {"task_id": task.group(1)}}) + text = "\n".join(c.get("text", "") for c in result.get("content", []) if isinstance(c, dict)) + if not re.search(r"Status: (RUNNING|PENDING|QUEUED)", text): + break + if result.get("isError"): + raise HarborError(text.strip() or f"Ghidra refused {tool}") + return re.sub(r"^\[Context\] Operating on: [^\n]*\n+", "", text) diff --git a/tests/test_reverse.py b/tests/test_reverse.py index fa4671a..02b1366 100644 --- a/tests/test_reverse.py +++ b/tests/test_reverse.py @@ -57,3 +57,63 @@ class ReverseTools(unittest.TestCase): self.r.bytes_at("0x500000") with self.assertRaises(HarborError): Reverse({}) + + +class GhidraForwarder(unittest.TestCase): + """The ghidra() tool against a fake streamable-HTTP MCP server: a session, a polled async task, a refusal.""" + + def setUp(self): + import http.server + import threading + calls = self.calls = [] + + class Fake(http.server.BaseHTTPRequestHandler): + def log_message(self, *a): + pass + + def do_POST(self): + body = json.loads(self.rfile.read(int(self.headers["Content-Length"]))) + calls.append((body["method"], body.get("params", {}).get("name"), self.headers.get("Mcp-Session-Id"))) + if "id" not in body: + self.send_response(202) + self.end_headers() + return + name = body.get("params", {}).get("name") + if body["method"] == "initialize": + result = {"protocolVersion": "2025-06-18"} + elif body["method"] == "tools/list": + result = {"tools": [{"name": "get_code", "description": "Code", + "inputSchema": {"properties": {"function": {}}, "required": ["function"]}}]} + elif name == "get_code": + result = {"content": [{"type": "text", "text": "Task submitted for async execution.\n\n" + "Task ID: 11111111-2222-3333-4444-555555555555\nStatus: RUNNING"}]} + elif name == "get_task_status": + result = {"content": [{"type": "text", "text": "[Context] Operating on: image.bin\n\nint F(void)"}]} + else: + result = {"isError": True, "content": [{"type": "text", "text": "Refused: read-only"}]} + data = ("data: " + json.dumps({"jsonrpc": "2.0", "id": body["id"], "result": result}) + "\n\n").encode() + self.send_response(200) + self.send_header("Content-Type", "text/event-stream") + self.send_header("Mcp-Session-Id", "s1") + self.send_header("Content-Length", str(len(data))) + self.end_headers() + self.wfile.write(data) + + self.server = http.server.ThreadingHTTPServer(("127.0.0.1", 0), Fake) + threading.Thread(target=self.server.serve_forever, daemon=True).start() + from harbor.reverse import Ghidra + self.g = Ghidra({"ghidra_url": f"http://127.0.0.1:{self.server.server_port}/mcp"}) + + def tearDown(self): + self.server.shutdown() + self.server.server_close() + + def test_tools_call_and_refusal(self): + self.assertEqual(self.g.tools()[0]["required"], ["function"]) + self.assertEqual(self.g.call("get_code", {"function": "F"}), "int F(void)") + self.assertIn(("tools/call", "get_task_status", "s1"), self.calls) + with self.assertRaises(HarborError): + self.g.call("rename_symbol", {}) + with self.assertRaises(HarborError): + from harbor.reverse import Ghidra + Ghidra({"ghidra_url": "http://127.0.0.1:9/mcp"}).tools()