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

Add testing and update README for JSON & testing

Browse Source
This commit is contained in:
Teal Bauer 2025-04-04 16:06:36 +02:00
parent cbe5dcc1f3
commit 1e737ed44b
5 changed files with 545 additions and 0 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

51
README.md
View File

@ -254,6 +254,57 @@ View result from get_function from ghydra (local){
Based on this analysis, I can see these binaries communicate using a simple protocol where... Based on this analysis, I can see these binaries communicate using a simple protocol where...
``` ```
# JSON Communication
GhydraMCP uses structured JSON for all communication between the Python bridge and Java plugin. This ensures consistent and reliable data exchange.
## Response Format
All responses follow a standard format:
```json
{
"success": true,
"result": "...",
"timestamp": 1712159482123,
"port": 8192,
"instanceType": "base"
}
```
Error responses include additional information:
```json
{
"success": false,
"error": "Error message",
"status_code": 404,
"timestamp": 1712159482123
}
```
This structured approach makes the communication more reliable and easier to debug.
# Testing
GhydraMCP includes comprehensive test suites for both the HTTP API and MCP bridge. See [TESTING.md](TESTING.md) for details on running the tests.
## HTTP API Tests
Tests the HTTP endpoints exposed by the Java plugin:
- Response format and structure
- JSON structure consistency
- Required fields in responses
- Error handling
## MCP Bridge Tests
Tests the MCP bridge functionality:
- MCP protocol communication
- Tool availability and structure
- Response format and structure
- JSON structure consistency
# Building from Source # Building from Source
You can build different artifacts with Maven: You can build different artifacts with Maven:

154
TESTING.md Normal file
View File

@ -0,0 +1,154 @@
# Testing GhydraMCP
This document describes how to test the GhydraMCP plugin and bridge.
## Prerequisites
- Python 3.11 or higher
- Ghidra with the GhydraMCP plugin installed and running
- The `requests` Python package (`pip install requests`)
## Running All Tests
The easiest way to run all tests is to use the test runner script:
```bash
python run_tests.py
```
This will run both the HTTP API tests and the MCP bridge tests and provide a summary of the results.
You can also run specific test suites:
```bash
# Run only the HTTP API tests
python run_tests.py --http
# Run only the MCP bridge tests
python run_tests.py --mcp
```
## HTTP API Tests
The `test_http_api.py` script tests the HTTP API exposed by the Java plugin. It verifies that the endpoints return the expected JSON structure and that the response format is consistent.
### Running the HTTP API Tests
1. Make sure Ghidra is running with the GhydraMCP plugin loaded
2. Run the tests:
```bash
python test_http_api.py
```
The tests will automatically skip if Ghidra is not running or if the plugin is not responding.
### What's Being Tested
- Basic connectivity to the plugin
- Response format and structure
- JSON structure consistency
- Required fields in responses
- Error handling
## MCP Bridge Tests
The `test_mcp_client.py` script tests the MCP bridge functionality using the MCP client library. It verifies that the bridge responds correctly to MCP requests and that the response format is consistent.
### Running the MCP Bridge Tests
1. Make sure Ghidra is running with the GhydraMCP plugin loaded
2. Run the tests:
```bash
python test_mcp_client.py
```
The test script will:
1. Connect to the bridge using the MCP client
2. Initialize the session
3. List the available tools
4. Call the list_instances tool
5. Call the discover_instances tool
6. Call the list_functions tool
### What's Being Tested
- MCP protocol communication
- Tool availability and structure
- Response format and structure
- JSON structure consistency
- Required fields in responses
- Proper initialization of the MCP session
- Ability to call tools and receive responses
## Troubleshooting
### HTTP API Tests
- If tests are skipped with "Ghidra server not running or not accessible", make sure Ghidra is running and the GhydraMCP plugin is loaded.
- If tests fail with connection errors, check that the plugin is listening on the expected port (default: 8192).
### MCP Bridge Tests
- If tests are skipped with "Failed to start MCP bridge process", check that the bridge script is executable and that all dependencies are installed.
- If tests fail with JSON parsing errors, check that the bridge is responding with valid JSON.
## Adding New Tests
### HTTP API Tests
To add a new test for an HTTP endpoint:
1. Add a new test method to the `GhydraMCPHttpApiTests` class
2. Use the `requests` library to make HTTP requests to the endpoint
3. Verify the response using assertions
Example:
```python
def test_new_endpoint(self):
"""Test the /new_endpoint endpoint"""
response = requests.get(f"{BASE_URL}/new_endpoint")
self.assertEqual(response.status_code, 200)
# Verify response is valid JSON
data = response.json()
# Check required fields in the standard response format
self.assertIn("success", data)
self.assertTrue(data["success"])
self.assertIn("timestamp", data)
self.assertIn("port", data)
```
### MCP Bridge Tests
To add a new test for an MCP tool:
1. Add a new test method to the `MCPBridgeTests` class
2. Use the `send_mcp_request` method to send an MCP request to the bridge
3. Verify the response using assertions
Example:
```python
def test_new_tool(self):
"""Test the new_tool tool"""
response = self.send_mcp_request("call_tool", {
"name": "new_tool",
"arguments": {
"param1": "value1",
"param2": "value2"
}
})
# Check basic response structure
self.assertIn("result", response)
self.assertIn("content", response["result"])
# Parse the content
content = response["result"]["content"]
self.assertIsInstance(content, list)
self.assertGreaterEqual(len(content), 1)
```

116
run_tests.py Normal file
View File

@ -0,0 +1,116 @@
#!/usr/bin/env python3
"""
Test runner for GhydraMCP tests.
This script runs both the HTTP API tests and the MCP bridge tests.
"""
import os
import subprocess
import sys
import unittest
import time
def print_header(text):
"""Print a header with the given text"""
print("\n" + "=" * 80)
print(f" {text} ".center(80, "="))
print("=" * 80 + "\n")
def run_http_api_tests():
"""Run the HTTP API tests"""
print_header("Running HTTP API Tests")
# Import and run the tests
try:
from test_http_api import GhydraMCPHttpApiTests
# Create a test suite with all tests from GhydraMCPHttpApiTests
suite = unittest.TestLoader().loadTestsFromTestCase(GhydraMCPHttpApiTests)
# Run the tests
result = unittest.TextTestRunner(verbosity=2).run(suite)
return result.wasSuccessful()
except ImportError:
print("Error: Could not import test_http_api.py")
return False
except Exception as e:
print(f"Error running HTTP API tests: {str(e)}")
return False
def run_mcp_bridge_tests():
"""Run the MCP bridge tests using the MCP client"""
print_header("Running MCP Bridge Tests")
try:
# Run the MCP client test script
import subprocess
import sys
print("Running MCP client test script...")
result = subprocess.run(
[sys.executable, "test_mcp_client.py"],
capture_output=True,
text=True
)
# Print the output
if result.stdout:
print("STDOUT:")
print(result.stdout)
if result.stderr:
print("STDERR:")
print(result.stderr)
# Return True if the process exited with code 0
return result.returncode == 0
except Exception as e:
print(f"Error running MCP bridge tests: {str(e)}")
return False
def run_all_tests():
"""Run all tests"""
print_header("GhydraMCP Test Suite")
# Run the HTTP API tests
http_api_success = run_http_api_tests()
# Run the MCP bridge tests
mcp_bridge_success = run_mcp_bridge_tests()
# Print a summary
print_header("Test Summary")
print(f"HTTP API Tests: {'PASSED' if http_api_success else 'FAILED'}")
print(f"MCP Bridge Tests: {'PASSED' if mcp_bridge_success else 'FAILED'}")
print(f"Overall: {'PASSED' if http_api_success and mcp_bridge_success else 'FAILED'}")
# Return True if all tests passed, False otherwise
return http_api_success and mcp_bridge_success
if __name__ == "__main__":
# Check if we have the required dependencies
try:
import requests
except ImportError:
print("Error: The 'requests' package is required to run the tests.")
print("Please install it with 'pip install requests'")
sys.exit(1)
# Parse command line arguments
if len(sys.argv) > 1:
if sys.argv[1] == "--http":
# Run only the HTTP API tests
success = run_http_api_tests()
elif sys.argv[1] == "--mcp":
# Run only the MCP bridge tests
success = run_mcp_bridge_tests()
else:
print(f"Unknown argument: {sys.argv[1]}")
print("Usage: python run_tests.py [--http|--mcp]")
sys.exit(1)
else:
# Run all tests
success = run_all_tests()
# Exit with the appropriate status code
sys.exit(0 if success else 1)

153
test_http_api.py Normal file
View File

@ -0,0 +1,153 @@
#!/usr/bin/env python3
"""
Test script for the GhydraMCP HTTP API.
This script tests the HTTP endpoints of the Java plugin.
"""
import json
import requests
import time
import unittest
# Default Ghidra server port
DEFAULT_PORT = 8192
BASE_URL = f"http://localhost:{DEFAULT_PORT}"
class GhydraMCPHttpApiTests(unittest.TestCase):
"""Test cases for the GhydraMCP HTTP API"""
def setUp(self):
"""Setup before each test"""
# Check if the server is running
try:
response = requests.get(f"{BASE_URL}/info", timeout=2)
if response.status_code != 200:
self.skipTest("Ghidra server not running or not responding")
except requests.exceptions.RequestException:
self.skipTest("Ghidra server not running or not accessible")
def test_info_endpoint(self):
"""Test the /info endpoint"""
response = requests.get(f"{BASE_URL}/info")
self.assertEqual(response.status_code, 200)
# Verify response is valid JSON
data = response.json()
# Check required fields
self.assertIn("port", data)
self.assertIn("isBaseInstance", data)
self.assertIn("project", data)
self.assertIn("file", data)
def test_root_endpoint(self):
"""Test the / endpoint"""
response = requests.get(BASE_URL)
self.assertEqual(response.status_code, 200)
# Verify response is valid JSON
data = response.json()
# Check required fields
self.assertIn("port", data)
self.assertIn("isBaseInstance", data)
self.assertIn("project", data)
self.assertIn("file", data)
def test_instances_endpoint(self):
"""Test the /instances endpoint"""
response = requests.get(f"{BASE_URL}/instances")
self.assertEqual(response.status_code, 200)
# Verify response is valid JSON
data = response.json()
# Check required fields in the standard response format
self.assertIn("success", data)
self.assertTrue(data["success"])
self.assertIn("timestamp", data)
self.assertIn("port", data)
# Check that we have either result or data
self.assertTrue("result" in data or "data" in data)
def test_functions_endpoint(self):
"""Test the /functions endpoint"""
response = requests.get(f"{BASE_URL}/functions")
self.assertEqual(response.status_code, 200)
# Verify response is valid JSON
data = response.json()
# Check required fields in the standard response format
self.assertIn("success", data)
self.assertTrue(data["success"])
self.assertIn("timestamp", data)
self.assertIn("port", data)
# Check that we have either result or data
self.assertTrue("result" in data or "data" in data)
def test_functions_with_pagination(self):
"""Test the /functions endpoint with pagination"""
response = requests.get(f"{BASE_URL}/functions?offset=0&limit=5")
self.assertEqual(response.status_code, 200)
# Verify response is valid JSON
data = response.json()
# Check required fields in the standard response format
self.assertIn("success", data)
self.assertTrue(data["success"])
self.assertIn("timestamp", data)
self.assertIn("port", data)
def test_classes_endpoint(self):
"""Test the /classes endpoint"""
response = requests.get(f"{BASE_URL}/classes")
self.assertEqual(response.status_code, 200)
# Verify response is valid JSON
data = response.json()
# Check required fields in the standard response format
self.assertIn("success", data)
self.assertTrue(data["success"])
self.assertIn("timestamp", data)
self.assertIn("port", data)
def test_segments_endpoint(self):
"""Test the /segments endpoint"""
response = requests.get(f"{BASE_URL}/segments")
self.assertEqual(response.status_code, 200)
# Verify response is valid JSON
data = response.json()
# Check required fields in the standard response format
self.assertIn("success", data)
self.assertTrue(data["success"])
self.assertIn("timestamp", data)
self.assertIn("port", data)
def test_variables_endpoint(self):
"""Test the /variables endpoint"""
response = requests.get(f"{BASE_URL}/variables")
self.assertEqual(response.status_code, 200)
# Verify response is valid JSON
data = response.json()
# Check required fields in the standard response format
self.assertIn("success", data)
self.assertTrue(data["success"])
self.assertIn("timestamp", data)
self.assertIn("port", data)
def test_error_handling(self):
"""Test error handling for non-existent endpoints"""
response = requests.get(f"{BASE_URL}/nonexistent_endpoint")
# This should return 404, but some servers might return other codes
self.assertNotEqual(response.status_code, 200)
if __name__ == "__main__":
unittest.main()

71
test_mcp_client.py Normal file
View File

@ -0,0 +1,71 @@
#!/usr/bin/env python3
"""
Test script for the GhydraMCP bridge using the MCP client.
This script tests the bridge by sending MCP requests and handling responses.
"""
import asyncio
import logging
import sys
from typing import Any
import anyio
from mcp.client.session import ClientSession
from mcp.client.stdio import StdioServerParameters, stdio_client
# Set up logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger("mcp_client_test")
async def test_bridge():
"""Test the bridge using the MCP client"""
# Configure the server parameters
server_parameters = StdioServerParameters(
command=sys.executable,
args=["bridge_mcp_hydra.py"],
)
# Connect to the bridge
logger.info("Connecting to bridge...")
async with stdio_client(server_parameters) as (read_stream, write_stream):
# Create a session
logger.info("Creating session...")
async with ClientSession(read_stream, write_stream) as session:
# Initialize the session
logger.info("Initializing session...")
init_result = await session.initialize()
logger.info(f"Initialization result: {init_result}")
# List tools
logger.info("Listing tools...")
tools_result = await session.list_tools()
logger.info(f"Tools result: {tools_result}")
# Call the list_instances tool
logger.info("Calling list_instances tool...")
list_instances_result = await session.call_tool("list_instances")
logger.info(f"List instances result: {list_instances_result}")
# Call the discover_instances tool
logger.info("Calling discover_instances tool...")
discover_instances_result = await session.call_tool("discover_instances")
logger.info(f"Discover instances result: {discover_instances_result}")
# Call the list_functions tool
logger.info("Calling list_functions tool...")
list_functions_result = await session.call_tool(
"list_functions",
arguments={"port": 8192, "offset": 0, "limit": 5}
)
logger.info(f"List functions result: {list_functions_result}")
def main():
"""Main entry point"""
try:
anyio.run(test_bridge)
except Exception as e:
logger.error(f"Error: {e}")
sys.exit(1)
if __name__ == "__main__":
main()