MCP/mcghidra
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Switch to JSON as bridge/plugin comm protocol

Browse Source
This commit is contained in:
Teal Bauer 2025-04-04 16:05:22 +02:00
parent 04d088591b
commit cbe5dcc1f3
4 changed files with 228 additions and 149 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

12
CHANGELOG.md
View File

@ -6,6 +6,18 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.1.0/).
## [Unreleased] ## [Unreleased]
### Added
- Structured JSON communication between Python bridge and Java plugin
- Consistent response format with metadata (timestamp, port, instance type)
- Comprehensive test suites for HTTP API and MCP bridge
- Test runner script for easy test execution
- Detailed testing documentation in TESTING.md
### Changed
- Improved error handling in API responses
- Enhanced JSON parsing in the Java plugin
- Updated documentation with JSON communication details
## [1.3.0] - 2025-04-02 ## [1.3.0] - 2025-04-02
### Added ### Added

220
bridge_mcp_hydra.py
View File

@ -1,7 +1,7 @@
# /// script # /// script
# requires-python = ">=3.11" # requires-python = ">=3.11"
# dependencies = [ # dependencies = [
# "mcp==1.5.0", # "mcp==1.6.0",
# "requests==2.32.3", # "requests==2.32.3",
# ] # ]
# /// # ///
@ -12,6 +12,7 @@ import threading
import time import time
from threading import Lock from threading import Lock
from typing import Dict from typing import Dict
from urllib.parse import quote
import requests import requests
from mcp.server.fastmcp import FastMCP from mcp.server.fastmcp import FastMCP
@ -25,23 +26,29 @@ DEFAULT_GHIDRA_HOST = "localhost"
QUICK_DISCOVERY_RANGE = range(8192, 8202) # Limited range for interactive/triggered discovery (10 ports) QUICK_DISCOVERY_RANGE = range(8192, 8202) # Limited range for interactive/triggered discovery (10 ports)
FULL_DISCOVERY_RANGE = range(8192, 8212) # Wider range for background discovery (20 ports) FULL_DISCOVERY_RANGE = range(8192, 8212) # Wider range for background discovery (20 ports)
mcp = FastMCP("hydra-mcp") instructions = """
GhydraMCP allows interacting with multiple Ghidra SRE instances. Ghidra SRE is a tool for reverse engineering and analyzing binaries, e.g. malware.
First, run `discover_instances` to find open Ghidra instances. List tools to see what GhydraMCP can do.
"""
mcp = FastMCP("GhydraMCP", instructions=instructions)
ghidra_host = os.environ.get("GHIDRA_HYDRA_HOST", DEFAULT_GHIDRA_HOST) ghidra_host = os.environ.get("GHIDRA_HYDRA_HOST", DEFAULT_GHIDRA_HOST)
print(f"Using Ghidra host: {ghidra_host}") # print(f"Using Ghidra host: {ghidra_host}")
def get_instance_url(port: int) -> str: def get_instance_url(port: int) -> str:
"""Get URL for a Ghidra instance by port""" """Get URL for a Ghidra instance by port"""
with instances_lock: with instances_lock:
if port in active_instances: if port in active_instances:
return active_instances[port]["url"] return active_instances[port]["url"]
# Auto-register if not found but port is valid # Auto-register if not found but port is valid
if 8192 <= port <= 65535: if 8192 <= port <= 65535:
register_instance(port) register_instance(port)
if port in active_instances: if port in active_instances:
return active_instances[port]["url"] return active_instances[port]["url"]
return f"http://{ghidra_host}:{port}" return f"http://{ghidra_host}:{port}"
def safe_get(port: int, endpoint: str, params: dict = None) -> dict: def safe_get(port: int, endpoint: str, params: dict = None) -> dict:
@ -53,21 +60,21 @@ def safe_get(port: int, endpoint: str, params: dict = None) -> dict:
try: try:
response = requests.get( response = requests.get(
url, url,
params=params, params=params,
headers={'Accept': 'application/json'}, headers={'Accept': 'application/json'},
timeout=5 timeout=5
) )
if response.ok: if response.ok:
try: try:
# Always expect JSON response # Always expect JSON response
json_data = response.json() json_data = response.json()
# If the response has a 'result' field that's a string, extract it # If the response has a 'result' field that's a string, extract it
if isinstance(json_data, dict) and 'result' in json_data: if isinstance(json_data, dict) and 'result' in json_data:
return json_data return json_data
# Otherwise, wrap the response in a standard format # Otherwise, wrap the response in a standard format
return { return {
"success": True, "success": True,
@ -86,7 +93,7 @@ def safe_get(port: int, endpoint: str, params: dict = None) -> dict:
# Try falling back to default instance if this was a secondary instance # Try falling back to default instance if this was a secondary instance
if port != DEFAULT_GHIDRA_PORT and response.status_code == 404: if port != DEFAULT_GHIDRA_PORT and response.status_code == 404:
return safe_get(DEFAULT_GHIDRA_PORT, endpoint, params) return safe_get(DEFAULT_GHIDRA_PORT, endpoint, params)
try: try:
error_data = response.json() error_data = response.json()
return { return {
@ -130,7 +137,7 @@ def safe_put(port: int, endpoint: str, data: dict) -> dict:
headers={'Content-Type': 'application/json'}, headers={'Content-Type': 'application/json'},
timeout=5 timeout=5
) )
if response.ok: if response.ok:
try: try:
return response.json() return response.json()
@ -143,7 +150,7 @@ def safe_put(port: int, endpoint: str, data: dict) -> dict:
# Try falling back to default instance if this was a secondary instance # Try falling back to default instance if this was a secondary instance
if port != DEFAULT_GHIDRA_PORT and response.status_code == 404: if port != DEFAULT_GHIDRA_PORT and response.status_code == 404:
return safe_put(DEFAULT_GHIDRA_PORT, endpoint, data) return safe_put(DEFAULT_GHIDRA_PORT, endpoint, data)
try: try:
error_data = response.json() error_data = response.json()
return { return {
@ -176,7 +183,7 @@ def safe_post(port: int, endpoint: str, data: dict | str) -> dict:
"""Perform a POST request to a specific Ghidra instance with JSON payload""" """Perform a POST request to a specific Ghidra instance with JSON payload"""
try: try:
url = f"{get_instance_url(port)}/{endpoint}" url = f"{get_instance_url(port)}/{endpoint}"
if isinstance(data, dict): if isinstance(data, dict):
response = requests.post( response = requests.post(
url, url,
@ -191,7 +198,7 @@ def safe_post(port: int, endpoint: str, data: dict | str) -> dict:
headers={'Content-Type': 'text/plain'}, headers={'Content-Type': 'text/plain'},
timeout=5 timeout=5
) )
if response.ok: if response.ok:
try: try:
return response.json() return response.json()
@ -201,10 +208,10 @@ def safe_post(port: int, endpoint: str, data: dict | str) -> dict:
"result": response.text.strip() "result": response.text.strip()
} }
else: else:
# Try falling back to default instance if this was a secondary instance # # Try falling back to default instance if this was a secondary instance
if port != DEFAULT_GHIDRA_PORT and response.status_code == 404: # if port != DEFAULT_GHIDRA_PORT and response.status_code == 404:
return safe_post(DEFAULT_GHIDRA_PORT, endpoint, data) # return safe_post(DEFAULT_GHIDRA_PORT, endpoint, data)
try: try:
error_data = response.json() error_data = response.json()
return { return {
@ -241,7 +248,7 @@ def list_instances() -> dict:
return { return {
"instances": [ "instances": [
{ {
"port": port, "port": port,
"url": info["url"], "url": info["url"],
"project": info.get("project", ""), "project": info.get("project", ""),
"file": info.get("file", "") "file": info.get("file", "")
@ -255,45 +262,39 @@ def register_instance(port: int, url: str = None) -> str:
"""Register a new Ghidra instance""" """Register a new Ghidra instance"""
if url is None: if url is None:
url = f"http://{ghidra_host}:{port}" url = f"http://{ghidra_host}:{port}"
# Verify instance is reachable before registering # Verify instance is reachable before registering
try: try:
test_url = f"{url}/instances" test_url = f"{url}/instances"
response = requests.get(test_url, timeout=2) response = requests.get(test_url, timeout=2)
if not response.ok: if not response.ok:
return f"Error: Instance at {url} is not responding properly" return f"Error: Instance at {url} is not responding properly"
# Try to get project info # Try to get project info
project_info = {"url": url} project_info = {"url": url}
try: try:
# Try the root endpoint first # Try the root endpoint first
root_url = f"{url}/" root_url = f"{url}/"
print(f"Trying to get root info from {root_url}", file=sys.stderr)
root_response = requests.get(root_url, timeout=1.5) # Short timeout for root root_response = requests.get(root_url, timeout=1.5) # Short timeout for root
if root_response.ok: if root_response.ok:
try: try:
print(f"Got response from root: {root_response.text}", file=sys.stderr)
root_data = root_response.json() root_data = root_response.json()
# Extract basic information from root # Extract basic information from root
if "project" in root_data and root_data["project"]: if "project" in root_data and root_data["project"]:
project_info["project"] = root_data["project"] project_info["project"] = root_data["project"]
if "file" in root_data and root_data["file"]: if "file" in root_data and root_data["file"]:
project_info["file"] = root_data["file"] project_info["file"] = root_data["file"]
print(f"Root data parsed: {project_info}", file=sys.stderr)
except Exception as e: except Exception as e:
print(f"Error parsing root info: {e}", file=sys.stderr) print(f"Error parsing root info: {e}", file=sys.stderr)
else:
print(f"Root endpoint returned {root_response.status_code}", file=sys.stderr)
# If we don't have project info yet, try the /info endpoint as a fallback # If we don't have project info yet, try the /info endpoint as a fallback
if not project_info.get("project") and not project_info.get("file"): if not project_info.get("project") and not project_info.get("file"):
info_url = f"{url}/info" info_url = f"{url}/info"
print(f"Trying fallback info from {info_url}", file=sys.stderr)
try: try:
info_response = requests.get(info_url, timeout=2) info_response = requests.get(info_url, timeout=2)
if info_response.ok: if info_response.ok:
@ -302,7 +303,7 @@ def register_instance(port: int, url: str = None) -> str:
# Extract relevant information # Extract relevant information
if "project" in info_data and info_data["project"]: if "project" in info_data and info_data["project"]:
project_info["project"] = info_data["project"] project_info["project"] = info_data["project"]
# Handle file information # Handle file information
file_info = info_data.get("file", {}) file_info = info_data.get("file", {})
if isinstance(file_info, dict) and file_info.get("name"): if isinstance(file_info, dict) and file_info.get("name"):
@ -318,10 +319,10 @@ def register_instance(port: int, url: str = None) -> str:
except Exception: except Exception:
# Non-critical, continue with registration even if project info fails # Non-critical, continue with registration even if project info fails
pass pass
with instances_lock: with instances_lock:
active_instances[port] = project_info active_instances[port] = project_info
return f"Registered instance on port {port} at {url}" return f"Registered instance on port {port} at {url}"
except Exception as e: except Exception as e:
return f"Error: Could not connect to instance at {url}: {str(e)}" return f"Error: Could not connect to instance at {url}: {str(e)}"
@ -338,7 +339,7 @@ def unregister_instance(port: int) -> str:
@mcp.tool() @mcp.tool()
def discover_instances(host: str = None) -> dict: def discover_instances(host: str = None) -> dict:
"""Auto-discover Ghidra instances by scanning ports (quick discovery with limited range) """Auto-discover Ghidra instances by scanning ports (quick discovery with limited range)
Args: Args:
host: Optional host to scan (defaults to configured ghidra_host) host: Optional host to scan (defaults to configured ghidra_host)
""" """
@ -348,11 +349,11 @@ def _discover_instances(port_range, host=None, timeout=0.5) -> dict:
"""Internal function to discover Ghidra instances by scanning ports""" """Internal function to discover Ghidra instances by scanning ports"""
found_instances = [] found_instances = []
scan_host = host if host is not None else ghidra_host scan_host = host if host is not None else ghidra_host
for port in port_range: for port in port_range:
if port in active_instances: if port in active_instances:
continue continue
url = f"http://{scan_host}:{port}" url = f"http://{scan_host}:{port}"
try: try:
test_url = f"{url}/instances" test_url = f"{url}/instances"
@ -363,19 +364,24 @@ def _discover_instances(port_range, host=None, timeout=0.5) -> dict:
except requests.exceptions.RequestException: except requests.exceptions.RequestException:
# Instance not available, just continue # Instance not available, just continue
continue continue
return { return {
"found": len(found_instances), "found": len(found_instances),
"instances": found_instances "instances": found_instances
} }
# Updated tool implementations with port parameter
from urllib.parse import quote
@mcp.tool() @mcp.tool()
def list_functions(port: int = DEFAULT_GHIDRA_PORT, offset: int = 0, limit: int = 100) -> list: def list_functions(port: int = DEFAULT_GHIDRA_PORT, offset: int = 0, limit: int = 100) -> list:
"""List all functions with pagination""" """List all functions in the current program
Args:
port: Ghidra instance port (default: 8192)
offset: Pagination offset (default: 0)
limit: Maximum number of segments to return (default: 100)
Returns:
List of strings with function names and addresses
"""
return safe_get(port, "functions", {"offset": offset, "limit": limit}) return safe_get(port, "functions", {"offset": offset, "limit": limit})
@mcp.tool() @mcp.tool()
@ -401,12 +407,12 @@ def update_data(port: int = DEFAULT_GHIDRA_PORT, address: str = "", new_name: st
@mcp.tool() @mcp.tool()
def list_segments(port: int = DEFAULT_GHIDRA_PORT, offset: int = 0, limit: int = 100) -> list: def list_segments(port: int = DEFAULT_GHIDRA_PORT, offset: int = 0, limit: int = 100) -> list:
"""List all memory segments in the current program with pagination """List all memory segments in the current program with pagination
Args: Args:
port: Ghidra instance port (default: 8192) port: Ghidra instance port (default: 8192)
offset: Pagination offset (default: 0) offset: Pagination offset (default: 0)
limit: Maximum number of segments to return (default: 100) limit: Maximum number of segments to return (default: 100)
Returns: Returns:
List of segment information strings List of segment information strings
""" """
@ -415,12 +421,12 @@ def list_segments(port: int = DEFAULT_GHIDRA_PORT, offset: int = 0, limit: int =
@mcp.tool() @mcp.tool()
def list_imports(port: int = DEFAULT_GHIDRA_PORT, offset: int = 0, limit: int = 100) -> list: def list_imports(port: int = DEFAULT_GHIDRA_PORT, offset: int = 0, limit: int = 100) -> list:
"""List all imported symbols with pagination """List all imported symbols with pagination
Args: Args:
port: Ghidra instance port (default: 8192) port: Ghidra instance port (default: 8192)
offset: Pagination offset (default: 0) offset: Pagination offset (default: 0)
limit: Maximum number of imports to return (default: 100) limit: Maximum number of imports to return (default: 100)
Returns: Returns:
List of import information strings List of import information strings
""" """
@ -429,12 +435,12 @@ def list_imports(port: int = DEFAULT_GHIDRA_PORT, offset: int = 0, limit: int =
@mcp.tool() @mcp.tool()
def list_exports(port: int = DEFAULT_GHIDRA_PORT, offset: int = 0, limit: int = 100) -> list: def list_exports(port: int = DEFAULT_GHIDRA_PORT, offset: int = 0, limit: int = 100) -> list:
"""List all exported symbols with pagination """List all exported symbols with pagination
Args: Args:
port: Ghidra instance port (default: 8192) port: Ghidra instance port (default: 8192)
offset: Pagination offset (default: 0) offset: Pagination offset (default: 0)
limit: Maximum number of exports to return (default: 100) limit: Maximum number of exports to return (default: 100)
Returns: Returns:
List of export information strings List of export information strings
""" """
@ -443,12 +449,12 @@ def list_exports(port: int = DEFAULT_GHIDRA_PORT, offset: int = 0, limit: int =
@mcp.tool() @mcp.tool()
def list_namespaces(port: int = DEFAULT_GHIDRA_PORT, offset: int = 0, limit: int = 100) -> list: def list_namespaces(port: int = DEFAULT_GHIDRA_PORT, offset: int = 0, limit: int = 100) -> list:
"""List all namespaces in the current program with pagination """List all namespaces in the current program with pagination
Args: Args:
port: Ghidra instance port (default: 8192) port: Ghidra instance port (default: 8192)
offset: Pagination offset (default: 0) offset: Pagination offset (default: 0)
limit: Maximum number of namespaces to return (default: 100) limit: Maximum number of namespaces to return (default: 100)
Returns: Returns:
List of namespace information strings List of namespace information strings
""" """
@ -457,12 +463,12 @@ def list_namespaces(port: int = DEFAULT_GHIDRA_PORT, offset: int = 0, limit: int
@mcp.tool() @mcp.tool()
def list_data_items(port: int = DEFAULT_GHIDRA_PORT, offset: int = 0, limit: int = 100) -> list: def list_data_items(port: int = DEFAULT_GHIDRA_PORT, offset: int = 0, limit: int = 100) -> list:
"""List all defined data items with pagination """List all defined data items with pagination
Args: Args:
port: Ghidra instance port (default: 8192) port: Ghidra instance port (default: 8192)
offset: Pagination offset (default: 0) offset: Pagination offset (default: 0)
limit: Maximum number of data items to return (default: 100) limit: Maximum number of data items to return (default: 100)
Returns: Returns:
List of data item information strings List of data item information strings
""" """
@ -471,13 +477,13 @@ def list_data_items(port: int = DEFAULT_GHIDRA_PORT, offset: int = 0, limit: int
@mcp.tool() @mcp.tool()
def search_functions_by_name(port: int = DEFAULT_GHIDRA_PORT, query: str = "", offset: int = 0, limit: int = 100) -> list: def search_functions_by_name(port: int = DEFAULT_GHIDRA_PORT, query: str = "", offset: int = 0, limit: int = 100) -> list:
"""Search for functions by name with pagination """Search for functions by name with pagination
Args: Args:
port: Ghidra instance port (default: 8192) port: Ghidra instance port (default: 8192)
query: Search string to match against function names query: Search string to match against function names
offset: Pagination offset (default: 0) offset: Pagination offset (default: 0)
limit: Maximum number of functions to return (default: 100) limit: Maximum number of functions to return (default: 100)
Returns: Returns:
List of matching function information strings or error message if query is empty List of matching function information strings or error message if query is empty
""" """
@ -488,11 +494,11 @@ def search_functions_by_name(port: int = DEFAULT_GHIDRA_PORT, query: str = "", o
@mcp.tool() @mcp.tool()
def get_function_by_address(port: int = DEFAULT_GHIDRA_PORT, address: str = "") -> str: def get_function_by_address(port: int = DEFAULT_GHIDRA_PORT, address: str = "") -> str:
"""Get function details by its memory address """Get function details by its memory address
Args: Args:
port: Ghidra instance port (default: 8192) port: Ghidra instance port (default: 8192)
address: Memory address of the function (hex string) address: Memory address of the function (hex string)
Returns: Returns:
Multiline string with function details including name, address, and signature Multiline string with function details including name, address, and signature
""" """
@ -501,10 +507,10 @@ def get_function_by_address(port: int = DEFAULT_GHIDRA_PORT, address: str = "")
@mcp.tool() @mcp.tool()
def get_current_address(port: int = DEFAULT_GHIDRA_PORT) -> str: def get_current_address(port: int = DEFAULT_GHIDRA_PORT) -> str:
"""Get the address currently selected in Ghidra's UI """Get the address currently selected in Ghidra's UI
Args: Args:
port: Ghidra instance port (default: 8192) port: Ghidra instance port (default: 8192)
Returns: Returns:
String containing the current memory address (hex format) String containing the current memory address (hex format)
""" """
@ -513,35 +519,23 @@ def get_current_address(port: int = DEFAULT_GHIDRA_PORT) -> str:
@mcp.tool() @mcp.tool()
def get_current_function(port: int = DEFAULT_GHIDRA_PORT) -> str: def get_current_function(port: int = DEFAULT_GHIDRA_PORT) -> str:
"""Get the function currently selected in Ghidra's UI """Get the function currently selected in Ghidra's UI
Args: Args:
port: Ghidra instance port (default: 8192) port: Ghidra instance port (default: 8192)
Returns: Returns:
Multiline string with function details including name, address, and signature Multiline string with function details including name, address, and signature
""" """
return "\n".join(safe_get(port, "get_current_function")) return "\n".join(safe_get(port, "get_current_function"))
@mcp.tool()
def list_functions(port: int = DEFAULT_GHIDRA_PORT) -> list:
"""List all functions in the current program
Args:
port: Ghidra instance port (default: 8192)
Returns:
List of strings with function names and addresses
"""
return safe_get(port, "list_functions")
@mcp.tool() @mcp.tool()
def decompile_function_by_address(port: int = DEFAULT_GHIDRA_PORT, address: str = "") -> str: def decompile_function_by_address(port: int = DEFAULT_GHIDRA_PORT, address: str = "") -> str:
"""Decompile a function at a specific memory address """Decompile a function at a specific memory address
Args: Args:
port: Ghidra instance port (default: 8192) port: Ghidra instance port (default: 8192)
address: Memory address of the function (hex string) address: Memory address of the function (hex string)
Returns: Returns:
Multiline string containing the decompiled pseudocode Multiline string containing the decompiled pseudocode
""" """
@ -550,11 +544,11 @@ def decompile_function_by_address(port: int = DEFAULT_GHIDRA_PORT, address: str
@mcp.tool() @mcp.tool()
def disassemble_function(port: int = DEFAULT_GHIDRA_PORT, address: str = "") -> list: def disassemble_function(port: int = DEFAULT_GHIDRA_PORT, address: str = "") -> list:
"""Get disassembly for a function at a specific address """Get disassembly for a function at a specific address
Args: Args:
port: Ghidra instance port (default: 8192) port: Ghidra instance port (default: 8192)
address: Memory address of the function (hex string) address: Memory address of the function (hex string)
Returns: Returns:
List of strings showing assembly instructions with addresses and comments List of strings showing assembly instructions with addresses and comments
""" """
@ -563,12 +557,12 @@ def disassemble_function(port: int = DEFAULT_GHIDRA_PORT, address: str = "") ->
@mcp.tool() @mcp.tool()
def set_decompiler_comment(port: int = DEFAULT_GHIDRA_PORT, address: str = "", comment: str = "") -> str: def set_decompiler_comment(port: int = DEFAULT_GHIDRA_PORT, address: str = "", comment: str = "") -> str:
"""Add/edit a comment in the decompiler view at a specific address """Add/edit a comment in the decompiler view at a specific address
Args: Args:
port: Ghidra instance port (default: 8192) port: Ghidra instance port (default: 8192)
address: Memory address to place comment (hex string) address: Memory address to place comment (hex string)
comment: Text of the comment to add comment: Text of the comment to add
Returns: Returns:
Confirmation message or error if failed Confirmation message or error if failed
""" """
@ -577,12 +571,12 @@ def set_decompiler_comment(port: int = DEFAULT_GHIDRA_PORT, address: str = "", c
@mcp.tool() @mcp.tool()
def set_disassembly_comment(port: int = DEFAULT_GHIDRA_PORT, address: str = "", comment: str = "") -> str: def set_disassembly_comment(port: int = DEFAULT_GHIDRA_PORT, address: str = "", comment: str = "") -> str:
"""Add/edit a comment in the disassembly view at a specific address """Add/edit a comment in the disassembly view at a specific address
Args: Args:
port: Ghidra instance port (default: 8192) port: Ghidra instance port (default: 8192)
address: Memory address to place comment (hex string) address: Memory address to place comment (hex string)
comment: Text of the comment to add comment: Text of the comment to add
Returns: Returns:
Confirmation message or error if failed Confirmation message or error if failed
""" """
@ -591,13 +585,13 @@ def set_disassembly_comment(port: int = DEFAULT_GHIDRA_PORT, address: str = "",
@mcp.tool() @mcp.tool()
def rename_local_variable(port: int = DEFAULT_GHIDRA_PORT, function_address: str = "", old_name: str = "", new_name: str = "") -> str: def rename_local_variable(port: int = DEFAULT_GHIDRA_PORT, function_address: str = "", old_name: str = "", new_name: str = "") -> str:
"""Rename a local variable within a function """Rename a local variable within a function
Args: Args:
port: Ghidra instance port (default: 8192) port: Ghidra instance port (default: 8192)
function_address: Memory address of the function (hex string) function_address: Memory address of the function (hex string)
old_name: Current name of the variable old_name: Current name of the variable
new_name: New name for the variable new_name: New name for the variable
Returns: Returns:
Confirmation message or error if failed Confirmation message or error if failed
""" """
@ -606,12 +600,12 @@ def rename_local_variable(port: int = DEFAULT_GHIDRA_PORT, function_address: str
@mcp.tool() @mcp.tool()
def rename_function_by_address(port: int = DEFAULT_GHIDRA_PORT, function_address: str = "", new_name: str = "") -> str: def rename_function_by_address(port: int = DEFAULT_GHIDRA_PORT, function_address: str = "", new_name: str = "") -> str:
"""Rename a function at a specific memory address """Rename a function at a specific memory address
Args: Args:
port: Ghidra instance port (default: 8192) port: Ghidra instance port (default: 8192)
function_address: Memory address of the function (hex string) function_address: Memory address of the function (hex string)
new_name: New name for the function new_name: New name for the function
Returns: Returns:
Confirmation message or error if failed Confirmation message or error if failed
""" """
@ -620,12 +614,12 @@ def rename_function_by_address(port: int = DEFAULT_GHIDRA_PORT, function_address
@mcp.tool() @mcp.tool()
def set_function_prototype(port: int = DEFAULT_GHIDRA_PORT, function_address: str = "", prototype: str = "") -> str: def set_function_prototype(port: int = DEFAULT_GHIDRA_PORT, function_address: str = "", prototype: str = "") -> str:
"""Update a function's signature/prototype """Update a function's signature/prototype
Args: Args:
port: Ghidra instance port (default: 8192) port: Ghidra instance port (default: 8192)
function_address: Memory address of the function (hex string) function_address: Memory address of the function (hex string)
prototype: New function prototype string (e.g. "int func(int param1)") prototype: New function prototype string (e.g. "int func(int param1)")
Returns: Returns:
Confirmation message or error if failed Confirmation message or error if failed
""" """
@ -634,13 +628,13 @@ def set_function_prototype(port: int = DEFAULT_GHIDRA_PORT, function_address: st
@mcp.tool() @mcp.tool()
def set_local_variable_type(port: int = DEFAULT_GHIDRA_PORT, function_address: str = "", variable_name: str = "", new_type: str = "") -> str: def set_local_variable_type(port: int = DEFAULT_GHIDRA_PORT, function_address: str = "", variable_name: str = "", new_type: str = "") -> str:
"""Change the data type of a local variable in a function """Change the data type of a local variable in a function
Args: Args:
port: Ghidra instance port (default: 8192) port: Ghidra instance port (default: 8192)
function_address: Memory address of the function (hex string) function_address: Memory address of the function (hex string)
variable_name: Name of the variable to modify variable_name: Name of the variable to modify
new_type: New data type for the variable (e.g. "int", "char*") new_type: New data type for the variable (e.g. "int", "char*")
Returns: Returns:
Confirmation message or error if failed Confirmation message or error if failed
""" """
@ -659,7 +653,7 @@ def list_function_variables(port: int = DEFAULT_GHIDRA_PORT, function: str = "")
"""List variables in a specific function""" """List variables in a specific function"""
if not function: if not function:
return "Error: function name is required" return "Error: function name is required"
encoded_name = quote(function) encoded_name = quote(function)
return safe_get(port, f"functions/{encoded_name}/variables", {}) return safe_get(port, f"functions/{encoded_name}/variables", {})
@ -668,7 +662,7 @@ def rename_variable(port: int = DEFAULT_GHIDRA_PORT, function: str = "", name: s
"""Rename a variable in a function""" """Rename a variable in a function"""
if not function or not name or not new_name: if not function or not name or not new_name:
return "Error: function, name, and new_name parameters are required" return "Error: function, name, and new_name parameters are required"
encoded_function = quote(function) encoded_function = quote(function)
encoded_var = quote(name) encoded_var = quote(name)
return safe_put(port, f"functions/{encoded_function}/variables/{encoded_var}", {"newName": new_name}) return safe_put(port, f"functions/{encoded_function}/variables/{encoded_var}", {"newName": new_name})
@ -678,7 +672,7 @@ def retype_variable(port: int = DEFAULT_GHIDRA_PORT, function: str = "", name: s
"""Change the data type of a variable in a function""" """Change the data type of a variable in a function"""
if not function or not name or not data_type: if not function or not name or not data_type:
return "Error: function, name, and data_type parameters are required" return "Error: function, name, and data_type parameters are required"
encoded_function = quote(function) encoded_function = quote(function)
encoded_var = quote(name) encoded_var = quote(name)
return safe_put(port, f"functions/{encoded_function}/variables/{encoded_var}", {"dataType": data_type}) return safe_put(port, f"functions/{encoded_function}/variables/{encoded_var}", {"dataType": data_type})
@ -692,7 +686,7 @@ def periodic_discovery():
try: try:
# Use the full discovery range # Use the full discovery range
_discover_instances(FULL_DISCOVERY_RANGE, timeout=0.5) _discover_instances(FULL_DISCOVERY_RANGE, timeout=0.5)
# Also check if any existing instances are down # Also check if any existing instances are down
with instances_lock: with instances_lock:
ports_to_remove = [] ports_to_remove = []
@ -704,31 +698,31 @@ def periodic_discovery():
ports_to_remove.append(port) ports_to_remove.append(port)
except requests.exceptions.RequestException: except requests.exceptions.RequestException:
ports_to_remove.append(port) ports_to_remove.append(port)
# Remove any instances that are down # Remove any instances that are down
for port in ports_to_remove: for port in ports_to_remove:
del active_instances[port] del active_instances[port]
print(f"Removed unreachable instance on port {port}") print(f"Removed unreachable instance on port {port}")
except Exception as e: except Exception as e:
print(f"Error in periodic discovery: {e}") print(f"Error in periodic discovery: {e}")
# Sleep for 30 seconds before next scan # Sleep for 30 seconds before next scan
time.sleep(30) time.sleep(30)
if __name__ == "__main__": if __name__ == "__main__":
# Auto-register default instance # # Auto-register default instance
register_instance(DEFAULT_GHIDRA_PORT, f"http://{ghidra_host}:{DEFAULT_GHIDRA_PORT}") # register_instance(DEFAULT_GHIDRA_PORT, f"http://{ghidra_host}:{DEFAULT_GHIDRA_PORT}")
# Auto-discover other instances # # Auto-discover other instances
discover_instances() # discover_instances()
# Start periodic discovery in background thread # # Start periodic discovery in background thread
discovery_thread = threading.Thread( # discovery_thread = threading.Thread(
target=periodic_discovery, # target=periodic_discovery,
daemon=True, # daemon=True,
name="GhydraMCP-Discovery" # name="GhydraMCP-Discovery"
) # )
discovery_thread.start() # discovery_thread.start()
signal.signal(signal.SIGINT, handle_sigint) # signal.signal(signal.SIGINT, handle_sigint)
mcp.run() mcp.run(transport="stdio")

6
pom.xml
View File

@ -23,9 +23,9 @@
<dependencies> <dependencies>
<!-- JSON handling --> <!-- JSON handling -->
<dependency> <dependency>
<groupId>com.googlecode.json-simple</groupId> <groupId>com.google.code.gson</groupId>
<artifactId>json-simple</artifactId> <artifactId>gson</artifactId>
<version>1.1.1</version> <version>2.10.1</version>
</dependency> </dependency>
<!-- Ghidra JARs as system-scoped dependencies --> <!-- Ghidra JARs as system-scoped dependencies -->

139
src/main/java/eu/starsong/ghidra/GhydraMCPPlugin.java
View File

@ -46,7 +46,8 @@ import java.util.concurrent.*;
import java.util.concurrent.atomic.*; import java.util.concurrent.atomic.*;
// For JSON response handling // For JSON response handling
import org.json.simple.JSONObject; import com.google.gson.Gson;
import com.google.gson.JsonObject;
import ghidra.app.services.CodeViewerService; import ghidra.app.services.CodeViewerService;
import ghidra.app.util.PseudoDisassembler; import ghidra.app.util.PseudoDisassembler;
@ -95,10 +96,12 @@ public class GhydraMCPPlugin extends Plugin implements ApplicationLevelPlugin {
Msg.info(this, "Starting as base instance on port " + port); Msg.info(this, "Starting as base instance on port " + port);
} }
} }
Msg.info(this, "Marker");
// Log to both console and log file // Log to both console and log file
Msg.info(this, "GhydraMCPPlugin loaded on port " + port); Msg.info(this, "GhydraMCPPlugin loaded on port " + port);
System.out.println("[GhydraMCP] Plugin loaded on port " + port); System.out.println("[GhydraMCP] Plugin loaded on port " + port);
try { try {
startServer(); startServer();
@ -201,15 +204,52 @@ public class GhydraMCPPlugin extends Plugin implements ApplicationLevelPlugin {
} }
}); });
// Class resources // Class resources with detailed logging
server.createContext("/classes", exchange -> { server.createContext("/classes", exchange -> {
if ("GET".equals(exchange.getRequestMethod())) { try {
Map<String, String> qparams = parseQueryParams(exchange); if ("GET".equals(exchange.getRequestMethod())) {
int offset = parseIntOrDefault(qparams.get("offset"), 0); try {
int limit = parseIntOrDefault(qparams.get("limit"), 100); Map<String, String> qparams = parseQueryParams(exchange);
sendResponse(exchange, getAllClassNames(offset, limit)); int offset = parseIntOrDefault(qparams.get("offset"), 0);
} else { int limit = parseIntOrDefault(qparams.get("limit"), 100);
exchange.sendResponseHeaders(405, -1); // Method Not Allowed
String result = getAllClassNames(offset, limit);
JsonObject json = new JsonObject();
json.addProperty("success", true);
json.addProperty("result", result);
json.addProperty("timestamp", System.currentTimeMillis());
json.addProperty("port", this.port);
Gson gson = new Gson();
String jsonStr = gson.toJson(json);
exchange.getResponseHeaders().set("Content-Type", "application/json; charset=utf-8");
byte[] bytes = jsonStr.getBytes(StandardCharsets.UTF_8);
exchange.sendResponseHeaders(200, bytes.length);
OutputStream os = exchange.getResponseBody();
os.write(bytes);
os.flush();
os.close();
} catch (Exception e) {
Msg.error(this, "/classes: Error in request processing: " + e.getMessage(), e);
try {
sendErrorResponse(exchange, 500, "Internal server error: " + e.getMessage());
} catch (IOException ioe) {
Msg.error(this, "/classes: Failed to send error response: " + ioe.getMessage(), ioe);
}
}
} else {
exchange.sendResponseHeaders(405, -1); // Method Not Allowed
}
} catch (Exception e) {
Msg.error(this, "/classes: Unhandled error: " + e.getMessage(), e);
} }
}); });
@ -353,8 +393,16 @@ public class GhydraMCPPlugin extends Plugin implements ApplicationLevelPlugin {
} }
}); });
// Super simple root endpoint - exact same as /info for consistency // Root endpoint - only handle exact "/" path
server.createContext("/", exchange -> { server.createContext("/", exchange -> {
// Only handle exact root path
if (!exchange.getRequestURI().getPath().equals("/")) {
// Return 404 for any other path that reaches this handler
Msg.info(this, "Received request for unknown path: " + exchange.getRequestURI().getPath());
sendErrorResponse(exchange, 404, "Endpoint not found");
return;
}
try { try {
String response = "{\n"; String response = "{\n";
response += "\"port\": " + port + ",\n"; response += "\"port\": " + port + ",\n";
@ -1229,39 +1277,64 @@ public class GhydraMCPPlugin extends Plugin implements ApplicationLevelPlugin {
private void sendResponse(HttpExchange exchange, Object response) throws IOException { private void sendResponse(HttpExchange exchange, Object response) throws IOException {
JSONObject json = new JSONObject(); JsonObject json = new JsonObject();
json.put("success", true); json.addProperty("success", true);
if (response instanceof String) { if (response instanceof String) {
json.put("result", response); json.addProperty("result", (String)response);
} else { } else {
json.put("data", response); json.addProperty("data", response.toString());
} }
json.put("timestamp", System.currentTimeMillis()); json.addProperty("timestamp", System.currentTimeMillis());
json.put("port", this.port); json.addProperty("port", this.port);
if (this.isBaseInstance) { if (this.isBaseInstance) {
json.put("instanceType", "base"); json.addProperty("instanceType", "base");
} else { } else {
json.put("instanceType", "secondary"); json.addProperty("instanceType", "secondary");
} }
sendJsonResponse(exchange, json); sendJsonResponse(exchange, json);
} }
private void sendJsonResponse(HttpExchange exchange, JSONObject jsonObj) throws IOException { private void sendJsonResponse(HttpExchange exchange, JsonObject jsonObj) throws IOException {
String json = jsonObj.toJSONString(); try {
byte[] bytes = json.getBytes(StandardCharsets.UTF_8); Gson gson = new Gson();
exchange.getResponseHeaders().set("Content-Type", "application/json; charset=utf-8"); String json = gson.toJson(jsonObj);
exchange.sendResponseHeaders(200, bytes.length); Msg.debug(this, "Sending JSON response: " + json);
try (OutputStream os = exchange.getResponseBody()) {
os.write(bytes); byte[] bytes = json.getBytes(StandardCharsets.UTF_8);
exchange.getResponseHeaders().set("Content-Type", "application/json; charset=utf-8");
exchange.sendResponseHeaders(200, bytes.length);
OutputStream os = null;
try {
os = exchange.getResponseBody();
os.write(bytes);
os.flush();
} catch (IOException e) {
Msg.error(this, "Error writing response body: " + e.getMessage(), e);
throw e;
} finally {
if (os != null) {
try {
os.close();
} catch (IOException e) {
Msg.error(this, "Error closing output stream: " + e.getMessage(), e);
}
}
}
} catch (Exception e) {
Msg.error(this, "Error in sendJsonResponse: " + e.getMessage(), e);
throw new IOException("Failed to send JSON response", e);
} }
} }
private void sendErrorResponse(HttpExchange exchange, int statusCode, String message) throws IOException { private void sendErrorResponse(HttpExchange exchange, int statusCode, String message) throws IOException {
JSONObject error = new JSONObject(); JsonObject error = new JsonObject();
error.put("error", message); error.addProperty("error", message);
error.put("status", statusCode); error.addProperty("status", statusCode);
error.put("success", false); error.addProperty("success", false);
byte[] bytes = error.toJSONString().getBytes(StandardCharsets.UTF_8);
Gson gson = new Gson();
byte[] bytes = gson.toJson(error).getBytes(StandardCharsets.UTF_8);
exchange.getResponseHeaders().set("Content-Type", "application/json; charset=utf-8"); exchange.getResponseHeaders().set("Content-Type", "application/json; charset=utf-8");
exchange.sendResponseHeaders(statusCode, bytes.length); exchange.sendResponseHeaders(statusCode, bytes.length);
try (OutputStream os = exchange.getResponseBody()) { try (OutputStream os = exchange.getResponseBody()) {