From 729d3e927cbcf17224e6b988eebcfae0d215c876 Mon Sep 17 00:00:00 2001 From: morluto Date: Tue, 14 Apr 2026 08:51:30 +0800 Subject: [PATCH] Initial commit: betterBinaryMCP - Enhanced MCP server for Hopper Disassembler 39 tools total: - 31 official Hopper 6 tools proxied via STDIO/NDJSON - 8 RE-specific tools: swift_classes, get_objc_classes, get_objc_protocols, batch_decompile, get_call_graph, analyze_swift_types, find_xrefs_to_name, binary_overview No Python injection needed - wraps official HopperMCPServer binary. --- .gitignore | 31 +++ LICENSE | 21 ++ README.md | 115 ++++++++ hopper_client.py | 149 ++++++++++ pyproject.toml | 21 ++ requirements.txt | 3 + server.py | 644 +++++++++++++++++++++++++++++++++++++++++++ tests/__init__.py | 0 tests/test_server.py | 298 ++++++++++++++++++++ 9 files changed, 1282 insertions(+) create mode 100644 .gitignore create mode 100644 LICENSE create mode 100644 README.md create mode 100644 hopper_client.py create mode 100644 pyproject.toml create mode 100644 requirements.txt create mode 100644 server.py create mode 100644 tests/__init__.py create mode 100644 tests/test_server.py diff --git a/.gitignore b/.gitignore new file mode 100644 index 000000000..53cf3d840 --- /dev/null +++ b/.gitignore @@ -0,0 +1,31 @@ +# Byte-compiled +__pycache__/ +*.py[cod] +*$py.class + +# Distribution +dist/ +build/ +*.egg-info/ +*.egg + +# Virtual environments +.venv/ +venv/ +env/ + +# IDE +.idea/ +.vscode/ +*.swp +*.swo +*~ + +# Testing +.pytest_cache/ +htmlcov/ +.coverage + +# OS +.DS_Store +Thumbs.db diff --git a/LICENSE b/LICENSE new file mode 100644 index 000000000..2d476d24f --- /dev/null +++ b/LICENSE @@ -0,0 +1,21 @@ +MIT License + +Copyright (c) 2026 morluto + +Permission is hereby granted, free of charge, to any person obtaining a copy +of this software and associated documentation files (the "Software"), to deal +in the Software without restriction, including without limitation the rights +to use, copy, modify, merge, publish, distribute, sublicense, and/or sell +copies of the Software, and to permit persons to whom the Software is +furnished to do so, subject to the following conditions: + +The above copyright notice and this permission notice shall be included in all +copies or substantial portions of the Software. + +THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR +IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, +FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE +AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER +LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, +OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE +SOFTWARE. diff --git a/README.md b/README.md new file mode 100644 index 000000000..c99c97d0a --- /dev/null +++ b/README.md @@ -0,0 +1,115 @@ +# betterBinaryMCP + +An enhanced MCP (Model Context Protocol) server for [Hopper Disassembler](https://www.hopperapp.com/) that wraps the official Hopper 6 MCP server and adds reverse engineering–specific analysis tools. + +**No Python injection or plugins needed** — just Hopper open with a binary. + +## Architecture + +``` +MCP Client ←STDIO→ betterBinaryMCP ←STDIO→ Official HopperMCPServer ←XPC→ Hopper.app +``` + +- Wraps the official `HopperMCPServer` binary as a subprocess +- Proxies all 31 official tools unchanged +- Adds 8 RE-specific tools built on top of official operations + +## Tools + +### 31 Official Tools (proxied) + +| Tool | Description | +|------|-------------| +| `address_name` | Get/set name (label) at an address | +| `comment` / `inline_comment` | Get/set comments | +| `current_address` / `current_procedure` / `current_document` | Cursor state | +| `goto_address` / `next_address` / `prev_address` | Navigation | +| `list_segments` / `list_procedures` / `list_strings` / `list_names` / `list_documents` / `list_bookmarks` | Listing | +| `search_procedures` / `search_strings` | Regex search | +| `procedure_pseudo_code` / `procedure_assembly` | Decompilation | +| `procedure_callees` / `procedure_callers` | Call relationships | +| `procedure_info` / `procedure_address` | Procedure metadata | +| `set_address_name` / `set_addresses_names` | Rename labels | +| `set_comment` / `set_inline_comment` | Write comments | +| `set_bookmark` / `unset_bookmark` | Bookmarks | +| `set_current_document` | Switch active document | +| `xrefs` | Cross-references | + +### 8 RE-Specific Tools (added) + +| Tool | Description | +|------|-------------| +| `binary_overview()` | Segment layout, procedure/string counts, document info | +| `swift_classes(pattern)` | Swift class hierarchy from mangled `_TtC` names | +| `get_objc_classes(pattern)` | Objective-C class names from labels | +| `get_objc_protocols()` | Objective-C protocol names from labels | +| `batch_decompile(addresses)` | Decompile up to 20 procedures at once | +| `get_call_graph(address, direction, depth)` | Recursive call graph traversal (forward/backward) | +| `analyze_swift_types()` | Categorize all Swift mangled names by module/type/method | +| `find_xrefs_to_name(name)` | Find references by name lookup | + +## Requirements + +- **Hopper Disassembler** v6+ (ships the official MCP server) +- **Python** 3.10+ +- **Hopper open** with a binary loaded and analyzed + +## Installation + +```bash +pip install fastmcp +``` + +## Usage + +### As an MCP Server (recommended) + +Add to your MCP client configuration (Claude Desktop, Cursor, etc.): + +```json +{ + "mcpServers": { + "betterBinaryMCP": { + "command": "python3", + "args": ["/path/to/betterBinaryMCP/server.py"] + } + } +} +``` + +### Command Line + +```bash +python3 server.py +``` + +### Test Connection + +```bash +python3 hopper_client.py +``` + +## Testing + +```bash +pip install pytest pytest-asyncio +python -m pytest tests/ -v +``` + +## How It Works + +1. On startup, the server launches the official `HopperMCPServer` binary as a subprocess +2. Communicates via NDJSON (newline-delimited JSON) over STDIO — no HTTP, no Content-Length framing +3. Proxied tools forward calls directly to the official server +4. RE tools compose multiple official calls (e.g., `swift_classes` calls `list_procedures` and parses `_TtC` mangled names) + +## Limitations + +- Requires Hopper 6+ (the official MCP server binary must exist) +- The official server connects to Hopper via XPC — Hopper must be running with a document open +- `batch_decompile` is limited to 20 procedures per call +- `get_call_graph` is limited to depth 5 + +## License + +MIT diff --git a/hopper_client.py b/hopper_client.py new file mode 100644 index 000000000..2832c0ae7 --- /dev/null +++ b/hopper_client.py @@ -0,0 +1,149 @@ +#!/usr/bin/env python3 +""" +NDJSON client for the official HopperMCPServer subprocess. + +Handles STDIO communication using NDJSON (newline-delimited JSON) protocol. +Each message is a single JSON object followed by a newline. +""" + +import json +import subprocess +import select +import threading +import time +from typing import Any, Optional + + +class HopperMCPSubprocessClient: + """Client for communicating with HopperMCPServer via STDIO/NDJSON.""" + + def __init__(self, server_path: str = "/Applications/Hopper Disassembler.app/Contents/MacOS/HopperMCPServer"): + self.server_path = server_path + self._proc: Optional[subprocess.Popen] = None + self._next_id = 1 + self._lock = threading.Lock() + self._server_info: dict = {} + + def start(self) -> dict: + """Start the subprocess and initialize the connection.""" + if self._proc is not None: + return self._server_info + + self._proc = subprocess.Popen( + [self.server_path], + stdin=subprocess.PIPE, + stdout=subprocess.PIPE, + stderr=subprocess.PIPE, + bufsize=0 + ) + + self._stderr_thread = threading.Thread(target=self._drain_stderr, daemon=True) + self._stderr_thread.start() + + time.sleep(0.5) + + self._send({ + "jsonrpc": "2.0", + "method": "initialize", + "params": { + "protocolVersion": "2024-11-05", + "capabilities": {}, + "clientInfo": {"name": "hopper-proxy", "version": "1.0"} + } + }) + response = self._recv(timeout=10.0) + if response and "result" in response: + self._server_info = response.get("result", {}).get("serverInfo", {}) + else: + raise RuntimeError(f"Failed to initialize server: {response}") + + time.sleep(1.0) + + return self._server_info + + def _drain_stderr(self): + """Drain stderr in background thread.""" + while self._proc and self._proc.poll() is None: + try: + ready = select.select([self._proc.stderr], [], [], 0.5) + if ready[0]: + line = self._proc.stderr.readline() + if not line: + break + except Exception: + break + + def stop(self): + """Stop the subprocess.""" + if self._proc: + self._proc.terminate() + self._proc.wait(timeout=5) + self._proc = None + + def _send(self, msg: dict) -> int: + """Send a JSON-RPC message.""" + with self._lock: + self._next_id += 1 + msg["id"] = self._next_id + line = json.dumps(msg) + "\n" + self._proc.stdin.write(line.encode("utf-8")) + self._proc.stdin.flush() + return self._next_id + + def _recv(self, timeout: float = 30.0) -> Optional[dict]: + """Receive a JSON-RPC response.""" + ready = select.select([self._proc.stdout], [], [], timeout) + if ready[0]: + line = self._proc.stdout.readline() + if line: + try: + return json.loads(line.decode("utf-8")) + except json.JSONDecodeError: + return None + return None + + def call_tool(self, tool_name: str, arguments: dict = None) -> Any: + """Call a tool on the server.""" + arguments = arguments or {} + self._send({ + "jsonrpc": "2.0", + "method": "tools/call", + "params": { + "name": tool_name, + "arguments": arguments + } + }) + response = self._recv(timeout=30.0) + if response and "error" in response: + error_msg = response["error"].get("message", str(response["error"])) + raise RuntimeError(f"Tool '{tool_name}' error: {error_msg}") + if response and "result" in response: + content = response["result"].get("content", []) + for c in content: + if c.get("type") == "text": + try: + return json.loads(c["text"]) + except (json.JSONDecodeError, TypeError): + return c["text"] + return response["result"] + return response + + @property + def server_info(self) -> dict: + return self._server_info + + +def test_connection(): + """Test connection to the official server.""" + client = HopperMCPSubprocessClient() + info = client.start() + print(f"Connected to {info.get('name', '?')} v{info.get('version', '?')}") + + docs = client.call_tool("list_documents", {}) + print(f"Documents: {docs}") + + client.stop() + + +if __name__ == "__main__": + test_connection() diff --git a/pyproject.toml b/pyproject.toml new file mode 100644 index 000000000..c239ba543 --- /dev/null +++ b/pyproject.toml @@ -0,0 +1,21 @@ +[project] +name = "betterBinaryMCP" +version = "0.1.0" +description = "Enhanced MCP server for Hopper Disassembler with reverse engineering tools" +requires-python = ">=3.10" +dependencies = [ + "fastmcp>=0.1.0", +] + +[project.optional-dependencies] +test = [ + "pytest>=7.0", + "pytest-asyncio>=0.21", +] + +[project.scripts] +better-binary-mcp = "HopperProxyMCP.server:main" + +[build-system] +requires = ["setuptools>=64"] +build-backend = "setuptools.backends._legacy:_Backend" diff --git a/requirements.txt b/requirements.txt new file mode 100644 index 000000000..24f02d9de --- /dev/null +++ b/requirements.txt @@ -0,0 +1,3 @@ +fastmcp +pytest +pytest-asyncio diff --git a/server.py b/server.py new file mode 100644 index 000000000..78e0427bb --- /dev/null +++ b/server.py @@ -0,0 +1,644 @@ +#!/usr/bin/env python3 +""" +HopperProxyMCP - Proxy server that wraps the official HopperMCPServer and adds RE-specific tools. + +Architecture: + MCP Client ←STDIO→ Our Proxy Server ←STDIO→ Official HopperMCPServer ←XPC→ Hopper.app +""" + +import asyncio +import json +from contextlib import asynccontextmanager +from typing import Any, Optional + +from fastmcp import FastMCP, Context +from pydantic import Field + +try: + from hopper_client import HopperMCPSubprocessClient +except ImportError: + from .hopper_client import HopperMCPSubprocessClient + + +OFFICIAL_SERVER_PATH = "/Applications/Hopper Disassembler.app/Contents/MacOS/HopperMCPServer" + +_client: Optional[HopperMCPSubprocessClient] = None + + +def get_client() -> HopperMCPSubprocessClient: + global _client + if _client is None: + _client = HopperMCPSubprocessClient(OFFICIAL_SERVER_PATH) + _client.start() + return _client + + +@asynccontextmanager +async def lifespan(server: FastMCP): + global _client + _client = HopperMCPSubprocessClient(OFFICIAL_SERVER_PATH) + _client.start() + yield + if _client: + _client.stop() + _client = None + + +mcp = FastMCP( + name="HopperProxyMCP", + instructions=( + "This server provides reverse engineering tools for analyzing binaries in Hopper Disassembler. " + "It wraps the official Hopper MCP server and adds Swift/ObjC-specific analysis tools. " + "Use binary_overview() first to understand the loaded binary, then swift_classes() or " + "analyze_swift_types() to explore the type hierarchy, and batch_decompile() or " + "get_call_graph() for deeper analysis." + ), + version="1.0.0", + mask_error_details=True, + on_duplicate_tools="error", + lifespan=lifespan, +) + + +def _call_official(tool_name: str, arguments: dict = None) -> Any: + return get_client().call_tool(tool_name, arguments or {}) + + +@mcp.tool(name="address_name") +async def address_name_tool( + document: Optional[str] = Field(default=None, description="The document name"), + address: Optional[str] = Field(default=None, description="The address to get the name for"), +) -> str: + result = _call_official("address_name", {"document": document, "address": address}) + if isinstance(result, dict): + return json.dumps(result, indent=2, default=str) + return str(result) if result else "OK" + + +@mcp.tool(name="comment") +async def comment_tool( + document: Optional[str] = Field(default=None, description="The document name"), + address: Optional[str] = Field(default=None, description="The address to get the comment for"), +) -> str: + result = _call_official("comment", {"document": document, "address": address}) + if isinstance(result, dict): + return json.dumps(result, indent=2, default=str) + return str(result) if result else "OK" + + +@mcp.tool(name="current_address") +async def current_address_tool( + document: Optional[str] = Field(default=None, description="The document name"), +) -> str: + result = _call_official("current_address", {"document": document}) + if isinstance(result, dict): + return json.dumps(result, indent=2, default=str) + return str(result) if result else "OK" + + +@mcp.tool(name="current_procedure") +async def current_procedure_tool( + document: Optional[str] = Field(default=None, description="The document name"), +) -> str: + result = _call_official("current_procedure", {"document": document}) + if isinstance(result, dict): + return json.dumps(result, indent=2, default=str) + return str(result) if result else "OK" + + +@mcp.tool(name="current_document") +async def current_document_tool() -> str: + result = _call_official("current_document", {}) + if isinstance(result, dict): + return json.dumps(result, indent=2, default=str) + return str(result) if result else "OK" + + +@mcp.tool(name="goto_address") +async def goto_address_tool( + address: str = Field(description="The address to go to"), + document: Optional[str] = Field(default=None, description="The document name"), +) -> str: + result = _call_official("goto_address", {"address": address, "document": document}) + if isinstance(result, dict): + return json.dumps(result, indent=2, default=str) + return str(result) if result else "OK" + + +@mcp.tool(name="inline_comment") +async def inline_comment_tool( + document: Optional[str] = Field(default=None, description="The document name"), + address: Optional[str] = Field(default=None, description="The address to get the inline comment for"), +) -> str: + result = _call_official("inline_comment", {"document": document, "address": address}) + if isinstance(result, dict): + return json.dumps(result, indent=2, default=str) + return str(result) if result else "OK" + + +@mcp.tool(name="list_bookmarks") +async def list_bookmarks_tool( + document: Optional[str] = Field(default=None, description="The document name"), +) -> str: + result = _call_official("list_bookmarks", {"document": document}) + if isinstance(result, dict): + return json.dumps(result, indent=2, default=str) + return str(result) if result else "OK" + + +@mcp.tool(name="list_documents") +async def list_documents_tool() -> str: + result = _call_official("list_documents", {}) + if isinstance(result, dict): + return json.dumps(result, indent=2, default=str) + return str(result) if result else "OK" + + +@mcp.tool(name="list_names") +async def list_names_tool( + document: Optional[str] = Field(default=None, description="The document name"), + address: Optional[str] = Field(default=None, description="The address to start from"), +) -> str: + result = _call_official("list_names", {"document": document, "address": address}) + if isinstance(result, dict): + return json.dumps(result, indent=2, default=str) + return str(result) if result else "OK" + + +@mcp.tool(name="list_procedures") +async def list_procedures_tool( + document: Optional[str] = Field(default=None, description="The document name"), +) -> str: + result = _call_official("list_procedures", {"document": document}) + if isinstance(result, dict): + return json.dumps(result, indent=2, default=str) + return str(result) if result else "OK" + + +@mcp.tool(name="list_segments") +async def list_segments_tool( + document: Optional[str] = Field(default=None, description="The document name"), +) -> str: + result = _call_official("list_segments", {"document": document}) + if isinstance(result, dict): + return json.dumps(result, indent=2, default=str) + return str(result) if result else "OK" + + +@mcp.tool(name="list_strings") +async def list_strings_tool( + document: Optional[str] = Field(default=None, description="The document name"), + address: Optional[str] = Field(default=None, description="The address to start from"), +) -> str: + result = _call_official("list_strings", {"document": document, "address": address}) + if isinstance(result, dict): + return json.dumps(result, indent=2, default=str) + return str(result) if result else "OK" + + +@mcp.tool(name="next_address") +async def next_address_tool( + document: Optional[str] = Field(default=None, description="The document name"), + address: Optional[str] = Field(default=None, description="The current address"), +) -> str: + result = _call_official("next_address", {"document": document, "address": address}) + if isinstance(result, dict): + return json.dumps(result, indent=2, default=str) + return str(result) if result else "OK" + + +@mcp.tool(name="prev_address") +async def prev_address_tool( + document: Optional[str] = Field(default=None, description="The document name"), + address: Optional[str] = Field(default=None, description="The current address"), +) -> str: + result = _call_official("prev_address", {"document": document, "address": address}) + if isinstance(result, dict): + return json.dumps(result, indent=2, default=str) + return str(result) if result else "OK" + + +@mcp.tool(name="procedure_address") +async def procedure_address_tool( + procedure: str = Field(description="The procedure name or address"), + document: Optional[str] = Field(default=None, description="The document name"), +) -> str: + result = _call_official("procedure_address", {"procedure": procedure, "document": document}) + if isinstance(result, dict): + return json.dumps(result, indent=2, default=str) + return str(result) if result else "OK" + + +@mcp.tool(name="procedure_assembly") +async def procedure_assembly_tool( + procedure: str = Field(description="The procedure name or address"), + document: Optional[str] = Field(default=None, description="The document name"), +) -> str: + result = _call_official("procedure_assembly", {"procedure": procedure, "document": document}) + if isinstance(result, dict): + return json.dumps(result, indent=2, default=str) + return str(result) if result else "OK" + + +@mcp.tool(name="procedure_callees") +async def procedure_callees_tool( + procedure: str = Field(description="The procedure name or address"), + document: Optional[str] = Field(default=None, description="The document name"), +) -> str: + result = _call_official("procedure_callees", {"procedure": procedure, "document": document}) + if isinstance(result, dict): + return json.dumps(result, indent=2, default=str) + elif isinstance(result, list): + return json.dumps(result, indent=2, default=str) + return str(result) if result else "OK" + + +@mcp.tool(name="procedure_callers") +async def procedure_callers_tool( + procedure: str = Field(description="The procedure name or address"), + document: Optional[str] = Field(default=None, description="The document name"), +) -> str: + result = _call_official("procedure_callers", {"procedure": procedure, "document": document}) + if isinstance(result, dict): + return json.dumps(result, indent=2, default=str) + elif isinstance(result, list): + return json.dumps(result, indent=2, default=str) + return str(result) if result else "OK" + + +@mcp.tool(name="procedure_info") +async def procedure_info_tool( + procedure: str = Field(description="The procedure name or address"), + document: Optional[str] = Field(default=None, description="The document name"), +) -> str: + result = _call_official("procedure_info", {"procedure": procedure, "document": document}) + if isinstance(result, dict): + return json.dumps(result, indent=2, default=str) + return str(result) if result else "OK" + + +@mcp.tool(name="procedure_pseudo_code") +async def procedure_pseudo_code_tool( + procedure: str = Field(description="The procedure name or address"), + document: Optional[str] = Field(default=None, description="The document name"), +) -> str: + result = _call_official("procedure_pseudo_code", {"procedure": procedure, "document": document}) + if isinstance(result, dict): + return json.dumps(result, indent=2, default=str) + return str(result) if result else "OK" + + +@mcp.tool(name="search_procedures") +async def search_procedures_tool( + pattern: str = Field(description="The regex pattern to search for"), + case_sensitive: bool = Field(default=False, description="Whether to match case"), + document: Optional[str] = Field(default=None, description="The document name"), +) -> str: + result = _call_official("search_procedures", {"pattern": pattern, "case_sensitive": case_sensitive, "document": document}) + if isinstance(result, dict): + return json.dumps(result, indent=2, default=str) + return str(result) if result else "OK" + + +@mcp.tool(name="search_strings") +async def search_strings_tool( + pattern: str = Field(description="The regex pattern to search for"), + case_sensitive: bool = Field(default=False, description="Whether to match case"), + document: Optional[str] = Field(default=None, description="The document name"), +) -> str: + result = _call_official("search_strings", {"pattern": pattern, "case_sensitive": case_sensitive, "document": document}) + if isinstance(result, dict): + return json.dumps(result, indent=2, default=str) + return str(result) if result else "OK" + + +@mcp.tool(name="set_address_name") +async def set_address_name_tool( + address: str = Field(description="The address to name"), + name: str = Field(description="The name to set"), + document: Optional[str] = Field(default=None, description="The document name"), +) -> str: + result = _call_official("set_address_name", {"address": address, "name": name, "document": document}) + return "OK" if result else "OK" + + +@mcp.tool(name="set_addresses_names") +async def set_addresses_names_tool( + names: dict = Field(description="A dictionary mapping addresses to names"), + document: Optional[str] = Field(default=None, description="The document name"), +) -> str: + result = _call_official("set_addresses_names", {"names": names, "document": document}) + return "OK" if result else "OK" + + +@mcp.tool(name="set_bookmark") +async def set_bookmark_tool( + address: str = Field(description="The address to bookmark"), + name: Optional[str] = Field(default=None, description="The bookmark name"), + document: Optional[str] = Field(default=None, description="The document name"), +) -> str: + result = _call_official("set_bookmark", {"address": address, "name": name, "document": document}) + return "OK" if result else "OK" + + +@mcp.tool(name="set_comment") +async def set_comment_tool( + address: str = Field(description="The address to comment on"), + comment: str = Field(description="The comment text"), + document: Optional[str] = Field(default=None, description="The document name"), +) -> str: + result = _call_official("set_comment", {"address": address, "comment": comment, "document": document}) + return "OK" if result else "OK" + + +@mcp.tool(name="set_current_document") +async def set_current_document_tool( + document: str = Field(description="The document name to set as current"), +) -> str: + result = _call_official("set_current_document", {"document": document}) + return "OK" if result else "OK" + + +@mcp.tool(name="set_inline_comment") +async def set_inline_comment_tool( + address: str = Field(description="The address to inline comment on"), + comment: str = Field(description="The inline comment text"), + document: Optional[str] = Field(default=None, description="The document name"), +) -> str: + result = _call_official("set_inline_comment", {"address": address, "comment": comment, "document": document}) + return "OK" if result else "OK" + + +@mcp.tool(name="unset_bookmark") +async def unset_bookmark_tool( + address: str = Field(description="The address to remove bookmark from"), + document: Optional[str] = Field(default=None, description="The document name"), +) -> str: + result = _call_official("unset_bookmark", {"address": address, "document": document}) + return "OK" if result else "OK" + + +@mcp.tool(name="xrefs") +async def xrefs_tool( + document: Optional[str] = Field(default=None, description="The document name"), + address: Optional[str] = Field(default=None, description="The address to get xrefs for"), +) -> str: + result = _call_official("xrefs", {"document": document, "address": address}) + if isinstance(result, dict): + return json.dumps(result, indent=2, default=str) + elif isinstance(result, list): + return json.dumps(result, indent=2, default=str) + return str(result) if result else "OK" + + +@mcp.tool(name="swift_classes") +async def swift_classes_tool( + pattern: str = Field(default="", description="Optional pattern to filter class names"), +) -> str: + all_procs = _call_official("list_procedures", {}) or {} + if isinstance(all_procs, dict): + all_procs = all_procs.get("procedures", all_procs) + + swift_classes = [] + for addr, name in list(all_procs.items())[:5000] if isinstance(all_procs, dict) else []: + if not isinstance(name, str): + continue + if name.startswith("_TtC") or "_TtC" in name: + if pattern and pattern not in name: + continue + swift_classes.append({"address": addr, "name": name}) + + result = { + "count": len(swift_classes), + "classes": swift_classes[:100], + } + return json.dumps(result, indent=2, default=str) + + +@mcp.tool(name="get_objc_classes") +async def get_objc_classes_tool( + pattern: str = Field(default="", description="Optional pattern to filter class names"), +) -> str: + names = _call_official("list_names", {}) or {} + if isinstance(names, dict): + names = names.get("names", names) + + objc_classes = [] + seen = set() + + for item in names if isinstance(names, list) else []: + if not isinstance(item, dict): + continue + name = item.get("name", "") + if not isinstance(name, str): + continue + if "OBJC_CLASS" in name or "OBJC_$_CLASS" in name or name.startswith("_OBJC_"): + if pattern and pattern not in name: + continue + addr = item.get("address", "") + if name not in seen: + seen.add(name) + objc_classes.append({"address": addr, "name": name}) + + result = { + "count": len(objc_classes), + "classes": objc_classes[:100], + } + return json.dumps(result, indent=2, default=str) + + +@mcp.tool(name="get_objc_protocols") +async def get_objc_protocols_tool() -> str: + names = _call_official("list_names", {}) or {} + if isinstance(names, dict): + names = names.get("names", names) + + protocols = [] + seen = set() + + for item in names if isinstance(names, list) else []: + if not isinstance(item, dict): + continue + name = item.get("name", "") + if not isinstance(name, str): + continue + if "OBJC_PROTOCOL" in name or "_TtP" in name: + if name not in seen: + seen.add(name) + protocols.append({"address": item.get("address", ""), "name": name}) + + result = { + "count": len(protocols), + "protocols": protocols[:100], + } + return json.dumps(result, indent=2, default=str) + + +@mcp.tool(name="batch_decompile") +async def batch_decompile_tool( + addresses: list[str] = Field(default=[], description="List of procedure addresses to decompile"), +) -> str: + if not addresses: + return json.dumps({"error": "No addresses provided"}, indent=2) + + addresses = addresses[:20] + + results = {} + for addr in addresses: + try: + pseudo = _call_official("procedure_pseudo_code", {"procedure": addr}) + results[addr] = pseudo if pseudo else "No output" + except Exception as e: + results[addr] = f"Error: {str(e)}" + + return json.dumps(results, indent=2, default=str) + + +@mcp.tool(name="get_call_graph") +async def get_call_graph_tool( + address: str = Field(description="Starting address for call graph traversal"), + direction: str = Field(default="forward", description="Traversal direction: 'forward' (callees) or 'backward' (callers)"), + depth: int = Field(default=2, description="Maximum traversal depth (1-5)"), +) -> str: + depth = min(max(1, depth), 5) + + visited = set() + queue = [(address, 0)] + graph = {} + + while queue: + current, current_depth = queue.pop(0) + + if current in visited or current_depth >= depth: + continue + visited.add(current) + + if current_depth not in graph: + graph[current_depth] = [] + + tool_name = "procedure_callees" if direction == "forward" else "procedure_callers" + try: + callees = _call_official(tool_name, {"procedure": current}) or [] + if isinstance(callees, dict): + callees = callees.get("callees", callees.get("callers", [])) + graph[current_depth].append({"address": current, "calls": callees}) + + if current_depth + 1 < depth: + for callee in callees if isinstance(callees, list) else []: + if callee not in visited: + queue.append((callee, current_depth + 1)) + except Exception as e: + graph[current_depth].append({"address": current, "error": str(e)}) + + return json.dumps(graph, indent=2, default=str) + + +@mcp.tool(name="analyze_swift_types") +async def analyze_swift_types_tool() -> str: + all_procs = _call_official("list_procedures", {}) or {} + if isinstance(all_procs, dict): + all_procs = all_procs.get("procedures", all_procs) + + swift_symbols = { + "classes": [], + "structs": [], + "enums": [], + "protocols": [], + "extensions": [], + "other": [], + } + + seen = set() + + for addr, name in list(all_procs.items())[:5000] if isinstance(all_procs, dict) else []: + if not isinstance(name, str) or "_Tt" not in name: + continue + if name in seen: + continue + seen.add(name) + + entry = {"address": addr, "name": name} + + if name.startswith("_TtC"): + swift_symbols["classes"].append(entry) + elif name.startswith("_TtV"): + swift_symbols["structs"].append(entry) + elif name.startswith("_TtO"): + swift_symbols["enums"].append(entry) + elif name.startswith("_TtP"): + swift_symbols["protocols"].append(entry) + elif name.startswith("_TtE"): + swift_symbols["extensions"].append(entry) + else: + swift_symbols["other"].append(entry) + + result = { + "total": sum(len(v) for v in swift_symbols.values()), + "categories": {k: {"count": len(v), "items": v[:50]} for k, v in swift_symbols.items()}, + } + return json.dumps(result, indent=2, default=str) + + +@mcp.tool(name="find_xrefs_to_name") +async def find_xrefs_to_name_tool( + name: str = Field(description="Name to search for and get xrefs to"), +) -> str: + addr_result = _call_official("address_name", {"address": name}) + addr = None + + if isinstance(addr_result, dict): + addr = addr_result.get("address", addr_result.get("name")) + elif isinstance(addr_result, str): + addr = addr_result + + if not addr: + return json.dumps({"error": f"Could not resolve name: {name}"}) + + xrefs = _call_official("xrefs", {"address": addr}) + + if isinstance(xrefs, dict): + pass + elif isinstance(xrefs, list): + xrefs = {"xrefs": xrefs} + + return json.dumps(xrefs, indent=2, default=str) + + +@mcp.tool(name="binary_overview") +async def binary_overview_tool() -> str: + segments = _call_official("list_segments", {}) or {} + if isinstance(segments, dict): + segments = segments.get("segments", segments) + + documents = _call_official("list_documents", {}) or [] + if isinstance(documents, dict): + documents = documents.get("documents", documents) + + procedures = _call_official("list_procedures", {}) or {} + if isinstance(procedures, dict): + procedures = procedures.get("procedures", procedures) + proc_count = len(procedures) if isinstance(procedures, dict) else "unknown" + + strings = _call_official("list_strings", {}) or {} + if isinstance(strings, dict): + strings = strings.get("strings", strings) + string_count = len(strings) if isinstance(strings, list) else "unknown" + + doc_name = documents[0] if documents else "unknown" + + result = { + "document": doc_name, + "segments": [{"name": s.get("name", ""), "start": s.get("start", ""), "end": s.get("end", "")} for s in segments[:10]], + "segment_count": len(segments) if isinstance(segments, list) else "unknown", + "procedure_count": proc_count, + "string_count": string_count, + } + return json.dumps(result, indent=2, default=str) + + +def main(): + mcp.run() + + +if __name__ == "__main__": + main() diff --git a/tests/__init__.py b/tests/__init__.py new file mode 100644 index 000000000..e69de29bb diff --git a/tests/test_server.py b/tests/test_server.py new file mode 100644 index 000000000..dea8f2e68 --- /dev/null +++ b/tests/test_server.py @@ -0,0 +1,298 @@ +#!/usr/bin/env python3 +""" +Tests for HopperProxyMCP server with mocked subprocess. +""" + +import asyncio +import json +import os +import sys +import pytest +from unittest.mock import MagicMock, patch, AsyncMock + +sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) + + +class MockHopperClient: + def __init__(self, server_path=None): + self._server_info = {"name": "HopperMCPServer", "version": "0.1.0"} + self._started = False + self._call_history = [] + + def start(self): + self._started = True + return self._server_info + + def stop(self): + self._started = False + + def call_tool(self, tool_name, arguments=None): + arguments = arguments or {} + self._call_history.append({"tool": tool_name, "args": arguments}) + + if tool_name == "list_documents": + return ["test_binary"] + elif tool_name == "list_segments": + return [{"name": "__TEXT", "start": "0x100000000", "end": "0x1000e0000"}] + elif tool_name == "list_procedures": + return { + "0x1040f4124": "-[_TtC6Signal29AccountSettingsViewController viewDidLoad]", + "0x1040f4150": "_TtC6Signal29AccountSettingsViewController" + } + elif tool_name == "procedure_pseudo_code": + return "func viewDidLoad() { self.setup() }" + elif tool_name == "search_procedures": + return {"0x1040f4124": "-[_TtC6Signal29AccountSettingsViewController viewDidLoad]"} + elif tool_name == "procedure_callees": + return ["0x1040f4160", "0x1040f4170"] + elif tool_name == "procedure_callers": + return ["0x1040f4100"] + elif tool_name == "list_names": + return [ + {"address": "0x1000a0000", "name": "_OBJC_CLASS_$_SignalAppDelegate"}, + {"address": "0x1000a0010", "name": "_TtC6Signal11AppDelegate"} + ] + elif tool_name == "address_name": + return {"address": "0x1040f4124", "name": "viewDidLoad"} + elif tool_name == "xrefs": + return ["0x1040f4100", "0x1040f4110"] + elif tool_name == "list_strings": + return [{"address": "0x100020000", "value": "Hello World"}] + elif tool_name == "set_comment": + return "OK" + elif tool_name == "set_address_name": + return "OK" + else: + return {"result": "ok"} + + @property + def server_info(self): + return self._server_info + + +@pytest.fixture +def mock_client(): + return MockHopperClient() + + +@pytest.fixture +def server_module(mock_client): + with patch("HopperProxyMCP.hopper_client.HopperMCPSubprocessClient") as mock_class: + mock_instance = MagicMock() + mock_instance.start.return_value = {"name": "HopperMCPServer", "version": "0.1.0"} + mock_instance.stop.return_value = None + mock_instance.call_tool.side_effect = mock_client.call_tool + mock_class.return_value = mock_instance + + from fastmcp import Client + import importlib + import HopperProxyMCP.server as server_mod + importlib.reload(server_mod) + yield server_mod + + +class TestProxyTools: + @pytest.mark.asyncio + async def test_binary_overview(self, server_module, mock_client): + from fastmcp import Client + async with Client(server_module.mcp) as client: + result = await client.call_tool("binary_overview", {}) + assert "test_binary" in result.data + assert "__TEXT" in result.data + + @pytest.mark.asyncio + async def test_swift_classes(self, server_module, mock_client): + from fastmcp import Client + async with Client(server_module.mcp) as client: + result = await client.call_tool("swift_classes", {"pattern": ""}) + data = json.loads(result.data) + assert data["count"] >= 1 + assert any("_TtC" in c["name"] for c in data["classes"]) + + @pytest.mark.asyncio + async def test_get_objc_classes(self, server_module, mock_client): + from fastmcp import Client + async with Client(server_module.mcp) as client: + result = await client.call_tool("get_objc_classes", {"pattern": ""}) + data = json.loads(result.data) + assert data["count"] >= 1 + + @pytest.mark.asyncio + async def test_get_objc_protocols(self, server_module, mock_client): + from fastmcp import Client + async with Client(server_module.mcp) as client: + result = await client.call_tool("get_objc_protocols", {}) + data = json.loads(result.data) + assert "count" in data + + @pytest.mark.asyncio + async def test_batch_decompile(self, server_module, mock_client): + from fastmcp import Client + async with Client(server_module.mcp) as client: + result = await client.call_tool("batch_decompile", { + "addresses": ["0x1040f4124", "0x1040f4150"] + }) + data = json.loads(result.data) + assert "0x1040f4124" in data + + @pytest.mark.asyncio + async def test_get_call_graph_forward(self, server_module, mock_client): + from fastmcp import Client + async with Client(server_module.mcp) as client: + result = await client.call_tool("get_call_graph", { + "address": "0x1040f4124", + "direction": "forward", + "depth": 2 + }) + data = json.loads(result.data) + assert "0" in data or 0 in data + + @pytest.mark.asyncio + async def test_get_call_graph_backward(self, server_module, mock_client): + from fastmcp import Client + async with Client(server_module.mcp) as client: + result = await client.call_tool("get_call_graph", { + "address": "0x1040f4124", + "direction": "backward", + "depth": 2 + }) + data = json.loads(result.data) + assert "0" in data or 0 in data + + @pytest.mark.asyncio + async def test_analyze_swift_types(self, server_module, mock_client): + from fastmcp import Client + async with Client(server_module.mcp) as client: + result = await client.call_tool("analyze_swift_types", {}) + data = json.loads(result.data) + assert "total" in data + assert "categories" in data + + @pytest.mark.asyncio + async def test_find_xrefs_to_name(self, server_module, mock_client): + from fastmcp import Client + async with Client(server_module.mcp) as client: + result = await client.call_tool("find_xrefs_to_name", { + "name": "viewDidLoad" + }) + data = json.loads(result.data) + + +class TestOfficialToolProxying: + @pytest.mark.asyncio + async def test_list_documents(self, server_module, mock_client): + from fastmcp import Client + async with Client(server_module.mcp) as client: + result = await client.call_tool("list_documents", {}) + assert "test_binary" in result.data + + @pytest.mark.asyncio + async def test_list_segments(self, server_module, mock_client): + from fastmcp import Client + async with Client(server_module.mcp) as client: + result = await client.call_tool("list_segments", {}) + assert "__TEXT" in result.data + + @pytest.mark.asyncio + async def test_list_procedures(self, server_module, mock_client): + from fastmcp import Client + async with Client(server_module.mcp) as client: + result = await client.call_tool("list_procedures", {}) + assert "_TtC" in result.data + + @pytest.mark.asyncio + async def test_procedure_pseudo_code(self, server_module, mock_client): + from fastmcp import Client + async with Client(server_module.mcp) as client: + result = await client.call_tool("procedure_pseudo_code", { + "procedure": "0x1040f4124" + }) + assert "viewDidLoad" in result.data + + @pytest.mark.asyncio + async def test_search_procedures(self, server_module, mock_client): + from fastmcp import Client + async with Client(server_module.mcp) as client: + result = await client.call_tool("search_procedures", { + "pattern": "viewDidLoad", + "case_sensitive": False + }) + assert "viewDidLoad" in result.data + + @pytest.mark.asyncio + async def test_procedure_callees(self, server_module, mock_client): + from fastmcp import Client + async with Client(server_module.mcp) as client: + result = await client.call_tool("procedure_callees", { + "procedure": "0x1040f4124" + }) + assert "0x1040f4160" in result.data or "0x1040f4170" in result.data + + @pytest.mark.asyncio + async def test_procedure_callers(self, server_module, mock_client): + from fastmcp import Client + async with Client(server_module.mcp) as client: + result = await client.call_tool("procedure_callers", { + "procedure": "0x1040f4124" + }) + assert "0x1040f4100" in result.data + + @pytest.mark.asyncio + async def test_list_names(self, server_module, mock_client): + from fastmcp import Client + async with Client(server_module.mcp) as client: + result = await client.call_tool("list_names", {}) + assert "_OBJC_CLASS_" in result.data or "_TtC" in result.data + + @pytest.mark.asyncio + async def test_address_name(self, server_module, mock_client): + from fastmcp import Client + async with Client(server_module.mcp) as client: + result = await client.call_tool("address_name", { + "address": "0x1040f4124" + }) + assert "viewDidLoad" in result.data + + @pytest.mark.asyncio + async def test_xrefs(self, server_module, mock_client): + from fastmcp import Client + async with Client(server_module.mcp) as client: + result = await client.call_tool("xrefs", { + "address": "0x1040f4124" + }) + assert "0x1040f4100" in result.data + + @pytest.mark.asyncio + async def test_set_comment(self, server_module, mock_client): + from fastmcp import Client + async with Client(server_module.mcp) as client: + result = await client.call_tool("set_comment", { + "address": "0x1040f4124", + "comment": "Test comment", + "document": "test_binary" + }) + assert result.data in ["OK", '{"result": "ok"}'] + + @pytest.mark.asyncio + async def test_set_address_name(self, server_module, mock_client): + from fastmcp import Client + async with Client(server_module.mcp) as client: + result = await client.call_tool("set_address_name", { + "address": "0x1040f4124", + "name": "myFunction", + "document": "test_binary" + }) + assert result.data in ["OK", '{"result": "ok"}'] + + +class TestMockClient: + def test_call_history(self, mock_client): + mock_client.call_tool("list_documents") + mock_client.call_tool("list_segments") + assert len(mock_client._call_history) == 2 + assert mock_client._call_history[0]["tool"] == "list_documents" + assert mock_client._call_history[1]["tool"] == "list_segments" + + +if __name__ == "__main__": + pytest.main([__file__, "-v"])