Make GHYDRAMCP host and port configurable for tests

This commit is contained in:
Teal Bauer 2025-04-09 10:15:49 +02:00
parent 5d588ba853
commit 9a1f97fa80
3 changed files with 30 additions and 9 deletions

View File

@ -6,7 +6,16 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.1.0/).
## [Unreleased] ## [Unreleased]
### Changed
- Updated port range calculations to use DEFAULT_GHIDRA_PORT
- Cleaned up comments and simplified code in bridge_mcp_hydra.py
- Improved error handling and response formatting
- Standardized API response structure across all endpoints
### Added ### Added
- Added GHIDRA_HTTP_API.md with documentation of the Java Plugin's HTTP API
- Added better docstrings and type hints for all MCP tools
- Added improved content-type handling for API requests
- Added decompiler output controls to customize analysis results: - Added decompiler output controls to customize analysis results:
- Choose between clean C-like pseudocode (default) or raw decompiler output - Choose between clean C-like pseudocode (default) or raw decompiler output
- Toggle syntax tree visibility for detailed analysis - Toggle syntax tree visibility for detailed analysis

View File

@ -7,10 +7,17 @@ import json
import requests import requests
import time import time
import unittest import unittest
import os
# Default Ghidra server port # Default Ghidra server port
DEFAULT_PORT = 8192 DEFAULT_PORT = 8192
BASE_URL = f"http://localhost:{DEFAULT_PORT}"
# Get host from environment variable or default to localhost
GHYDRAMCP_TEST_HOST = os.getenv('GHYDRAMCP_TEST_HOST')
if GHYDRAMCP_TEST_HOST and GHYDRAMCP_TEST_HOST.strip():
BASE_URL = f"http://{GHYDRAMCP_TEST_HOST}:{DEFAULT_PORT}"
else:
BASE_URL = f"http://localhost:{DEFAULT_PORT}"
class GhydraMCPHttpApiTests(unittest.TestCase): class GhydraMCPHttpApiTests(unittest.TestCase):
"""Test cases for the GhydraMCP HTTP API""" """Test cases for the GhydraMCP HTTP API"""

View File

@ -1,14 +1,19 @@
#!/usr/bin/env python3 #!/usr/bin/env python3
""" """
Test script for the GhydraMCP bridge using the MCP client. Test script for the GhydraMCP bridge using the MCP client.
This script tests the bridge by sending MCP requests and handling responses. This script tests the bridge by sending MCP requests and handling responses.
""" """
import json import json
import logging import logging
import os
import sys import sys
from typing import Any from typing import Any
import anyio import anyio
# Get host and port from environment variables or use defaults
GHYDRAMCP_TEST_HOST = os.getenv('GHYDRAMCP_TEST_HOST', 'localhost')
GHYDRAMCP_TEST_PORT = int(os.getenv('GHYDRAMCP_TEST_PORT', '8192'))
from mcp.client.session import ClientSession from mcp.client.session import ClientSession
from mcp.client.stdio import StdioServerParameters, stdio_client from mcp.client.stdio import StdioServerParameters, stdio_client
@ -74,7 +79,7 @@ async def test_bridge():
logger.info("Calling list_functions tool...") logger.info("Calling list_functions tool...")
list_functions_result = await session.call_tool( list_functions_result = await session.call_tool(
"list_functions", "list_functions",
arguments={"port": 8192, "offset": 0, "limit": 5} arguments={"port": GHYDRAMCP_TEST_PORT, "offset": 0, "limit": 5}
) )
logger.info(f"List functions result: {list_functions_result}") logger.info(f"List functions result: {list_functions_result}")
@ -85,7 +90,7 @@ async def test_bridge():
# Get a known function to test with from list_functions result # Get a known function to test with from list_functions result
list_funcs = await session.call_tool( list_funcs = await session.call_tool(
"list_functions", "list_functions",
arguments={"port": 8192, "offset": 0, "limit": 5} arguments={"port": GHYDRAMCP_TEST_PORT, "offset": 0, "limit": 5}
) )
if not list_funcs or not list_funcs.content: if not list_funcs or not list_funcs.content:
@ -117,7 +122,7 @@ async def test_bridge():
test_name = f"{func_name}_test" test_name = f"{func_name}_test"
# Test successful rename operations (These return simple success/message, not full result) # Test successful rename operations (These return simple success/message, not full result)
rename_args = {"port": 8192, "name": original_name, "new_name": test_name} rename_args = {"port": GHYDRAMCP_TEST_PORT, "name": original_name, "new_name": test_name}
logger.info(f"Calling update_function with args: {rename_args}") logger.info(f"Calling update_function with args: {rename_args}")
rename_result = await session.call_tool("update_function", arguments=rename_args) rename_result = await session.call_tool("update_function", arguments=rename_args)
rename_data = json.loads(rename_result.content[0].text) # Parse simple response rename_data = json.loads(rename_result.content[0].text) # Parse simple response
@ -131,7 +136,7 @@ async def test_bridge():
logger.info(f"Renamed function result: {renamed_func}") logger.info(f"Renamed function result: {renamed_func}")
# Rename back to original # Rename back to original
revert_args = {"port": 8192, "name": test_name, "new_name": original_name} revert_args = {"port": GHYDRAMCP_TEST_PORT, "name": test_name, "new_name": original_name}
logger.info(f"Calling update_function with args: {revert_args}") logger.info(f"Calling update_function with args: {revert_args}")
revert_result = await session.call_tool("update_function", arguments=revert_args) revert_result = await session.call_tool("update_function", arguments=revert_args)
revert_data = json.loads(revert_result.content[0].text) # Parse simple response revert_data = json.loads(revert_result.content[0].text) # Parse simple response
@ -139,14 +144,14 @@ async def test_bridge():
logger.info(f"Revert rename result: {revert_result}") logger.info(f"Revert rename result: {revert_result}")
# Verify revert by getting the function # Verify revert by getting the function
original_func = await session.call_tool("get_function", arguments={"port": 8192, "name": original_name}) original_func = await session.call_tool("get_function", arguments={"port": GHYDRAMCP_TEST_PORT, "name": original_name})
original_data = await assert_standard_mcp_success_response(original_func.content, expected_result_type=dict) original_data = await assert_standard_mcp_success_response(original_func.content, expected_result_type=dict)
assert original_data.get("result", {}).get("name") == original_name, f"Original function has wrong name: {original_data}" assert original_data.get("result", {}).get("name") == original_name, f"Original function has wrong name: {original_data}"
logger.info(f"Original function result: {original_func}") logger.info(f"Original function result: {original_func}")
# Test get_function_by_address # Test get_function_by_address
logger.info(f"Calling get_function_by_address with address: {func_address}") logger.info(f"Calling get_function_by_address with address: {func_address}")
get_by_addr_result = await session.call_tool("get_function_by_address", arguments={"port": 8192, "address": func_address}) get_by_addr_result = await session.call_tool("get_function_by_address", arguments={"port": GHYDRAMCP_TEST_PORT, "address": func_address})
get_by_addr_data = await assert_standard_mcp_success_response(get_by_addr_result.content, expected_result_type=dict) get_by_addr_data = await assert_standard_mcp_success_response(get_by_addr_result.content, expected_result_type=dict)
result_data = get_by_addr_data.get("result", {}) result_data = get_by_addr_data.get("result", {})
assert "name" in result_data, "Missing name field in get_function_by_address result" assert "name" in result_data, "Missing name field in get_function_by_address result"
@ -158,7 +163,7 @@ async def test_bridge():
# Test decompile_function_by_address # Test decompile_function_by_address
logger.info(f"Calling decompile_function_by_address with address: {func_address}") logger.info(f"Calling decompile_function_by_address with address: {func_address}")
decompile_result = await session.call_tool("decompile_function_by_address", arguments={"port": 8192, "address": func_address}) decompile_result = await session.call_tool("decompile_function_by_address", arguments={"port": GHYDRAMCP_TEST_PORT, "address": func_address})
decompile_data = await assert_standard_mcp_success_response(decompile_result.content, expected_result_type=dict) decompile_data = await assert_standard_mcp_success_response(decompile_result.content, expected_result_type=dict)
assert "decompilation" in decompile_data.get("result", {}), f"Decompile result missing 'decompilation': {decompile_data}" assert "decompilation" in decompile_data.get("result", {}), f"Decompile result missing 'decompilation': {decompile_data}"
assert isinstance(decompile_data.get("result", {}).get("decompilation", ""), str), f"Decompilation is not a string: {decompile_data}" assert isinstance(decompile_data.get("result", {}).get("decompilation", ""), str), f"Decompilation is not a string: {decompile_data}"