diff --git a/hopper_client.py b/hopper_client.py deleted file mode 100644 index 2832c0ae7..000000000 --- a/hopper_client.py +++ /dev/null @@ -1,149 +0,0 @@ -#!/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 deleted file mode 100644 index c239ba543..000000000 --- a/pyproject.toml +++ /dev/null @@ -1,21 +0,0 @@ -[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 deleted file mode 100644 index 24f02d9de..000000000 --- a/requirements.txt +++ /dev/null @@ -1,3 +0,0 @@ -fastmcp -pytest -pytest-asyncio diff --git a/server.py b/server.py deleted file mode 100644 index 78e0427bb..000000000 --- a/server.py +++ /dev/null @@ -1,644 +0,0 @@ -#!/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 deleted file mode 100644 index e69de29bb..000000000 diff --git a/tests/test_server.py b/tests/test_server.py deleted file mode 100644 index dea8f2e68..000000000 --- a/tests/test_server.py +++ /dev/null @@ -1,298 +0,0 @@ -#!/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"])