Remove legacy Python MCP implementation
This commit is contained in:
morluto
parent
729d3e927c
commit
fbce15f990
149
hopper_client.py
149
hopper_client.py
|
|
@ -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()
|
||||
|
|
@ -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"
|
||||
|
|
@ -1,3 +0,0 @@
|
|||
fastmcp
|
||||
pytest
|
||||
pytest-asyncio
|
||||
644
server.py
644
server.py
|
|
@ -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()
|
||||
|
|
@ -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"])
|
||||
Loading…
Reference in New Issue