From 81eb97053cf8e72ede111b7fa53f9560d7df62ea Mon Sep 17 00:00:00 2001 From: cbdmaul Date: Thu, 23 Apr 2026 23:07:13 -0400 Subject: [PATCH] fix: validate force options against module capabilities Only inject ForceExploit and AutoCheck in run_exploit when the selected module actually supports those options, preventing invalid-option failures on modules like php_cgi_arg_injection. Extend force-exploit tests with supported/unsupported option coverage and robust async mock assertions. Made-with: Cursor --- MetasploitMCP.py | 30 ++++++++--- tests/test_force_exploit_autocheck.py | 77 +++++++++++++++++++++++---- 2 files changed, 90 insertions(+), 17 deletions(-) diff --git a/MetasploitMCP.py b/MetasploitMCP.py index 03eb739..516afe5 100644 --- a/MetasploitMCP.py +++ b/MetasploitMCP.py @@ -3239,7 +3239,8 @@ async def run_exploit( run_as_job: If False, run sync via console. If True, run async via RPC. check_vulnerability: If True, run module's 'check' action first (if available). force_exploit: If True (default), automatically sets ForceExploit=true and - AutoCheck=false unless already provided in options. This avoids + AutoCheck=false unless already provided in options *and* those + options are supported by the selected module. This avoids AutoCheck aborting exploitation when check results are inconclusive. Set False to preserve module defaults. timeout_seconds: Max time for synchronous run via console (max: 120s, values above are capped). @@ -3265,9 +3266,10 @@ async def run_exploit( logger.info(f"Module {module} options: {options}") logger.info(f"Payload {payload} options: {payload_options}") + exploit_module = None # Validate module exists before proceeding try: - await _get_module_object('exploit', module) + exploit_module = await _get_module_object('exploit', module) logger.debug(f"Module '{module}' validated successfully") except InvalidModuleError as e: logger.warning(f"Exploit module '{module}' not found: {e}") @@ -3283,12 +3285,28 @@ async def run_exploit( return {"status": "error", "message": f"Invalid options format: {e}"} if force_exploit: + supported_options = await _get_module_valid_options(exploit_module) if exploit_module else set() + supports_force_exploit = "ForceExploit" in supported_options + supports_auto_check = "AutoCheck" in supported_options + if 'ForceExploit' not in parsed_options: - parsed_options['ForceExploit'] = True - logger.info("Auto-set ForceExploit=true for exploit execution.") + if supports_force_exploit: + parsed_options['ForceExploit'] = True + logger.info("Auto-set ForceExploit=true for exploit execution.") + else: + logger.info( + "Skipping ForceExploit auto-injection for module '%s' because the option is not supported.", + module, + ) if 'AutoCheck' not in parsed_options: - parsed_options['AutoCheck'] = False - logger.info("Auto-set AutoCheck=false for exploit execution.") + if supports_auto_check: + parsed_options['AutoCheck'] = False + logger.info("Auto-set AutoCheck=false for exploit execution.") + else: + logger.info( + "Skipping AutoCheck auto-injection for module '%s' because the option is not supported.", + module, + ) # Parse payload options gracefully try: diff --git a/tests/test_force_exploit_autocheck.py b/tests/test_force_exploit_autocheck.py index b95cf22..b718f53 100644 --- a/tests/test_force_exploit_autocheck.py +++ b/tests/test_force_exploit_autocheck.py @@ -15,18 +15,26 @@ ) -def _make_module(fullname: str) -> MagicMock: +def _make_module(fullname: str, valid_options: dict | None = None) -> MagicMock: module_obj = MagicMock() module_obj.fullname = fullname + module_obj.options = valid_options or {} return module_obj @pytest.mark.asyncio async def test_run_exploit_force_exploit_injected_by_default(monkeypatch): - exploit_module = _make_module("exploit/unix/webapp/drupal_drupalgeddon2") + exploit_module = _make_module( + "exploit/unix/webapp/drupal_drupalgeddon2", + valid_options={"ForceExploit": {}, "AutoCheck": {}, "RHOSTS": {}}, + ) monkeypatch.setattr(metasploit_mcp, "_get_module_object", AsyncMock(return_value=exploit_module)) - with patch.object(metasploit_mcp, "_execute_module_rpc", return_value={"status": "success"}): + with patch.object( + metasploit_mcp, + "_execute_module_rpc", + new=AsyncMock(return_value={"status": "success"}), + ) as mock_execute_module_rpc: result = await metasploit_mcp.run_exploit( module="exploit/unix/webapp/drupal_drupalgeddon2", options={"RHOSTS": "192.0.2.10"}, @@ -34,17 +42,24 @@ async def test_run_exploit_force_exploit_injected_by_default(monkeypatch): ) assert result["status"] == "success" - called_options = metasploit_mcp._execute_module_rpc.call_args.kwargs["module_options"] + called_options = mock_execute_module_rpc.await_args.kwargs["module_options"] assert called_options["ForceExploit"] is True assert called_options["AutoCheck"] is False @pytest.mark.asyncio async def test_run_exploit_force_exploit_false_preserves_options(monkeypatch): - exploit_module = _make_module("exploit/unix/webapp/drupal_drupalgeddon2") + exploit_module = _make_module( + "exploit/unix/webapp/drupal_drupalgeddon2", + valid_options={"ForceExploit": {}, "AutoCheck": {}, "RHOSTS": {}}, + ) monkeypatch.setattr(metasploit_mcp, "_get_module_object", AsyncMock(return_value=exploit_module)) - with patch.object(metasploit_mcp, "_execute_module_rpc", return_value={"status": "success"}): + with patch.object( + metasploit_mcp, + "_execute_module_rpc", + new=AsyncMock(return_value={"status": "success"}), + ) as mock_execute_module_rpc: result = await metasploit_mcp.run_exploit( module="exploit/unix/webapp/drupal_drupalgeddon2", options={"RHOSTS": "192.0.2.10"}, @@ -53,7 +68,7 @@ async def test_run_exploit_force_exploit_false_preserves_options(monkeypatch): ) assert result["status"] == "success" - called_options = metasploit_mcp._execute_module_rpc.call_args.kwargs["module_options"] + called_options = mock_execute_module_rpc.await_args.kwargs["module_options"] assert "ForceExploit" not in called_options assert "AutoCheck" not in called_options @@ -84,9 +99,15 @@ def call_side_effect(method, args=None): client.call = MagicMock(side_effect=call_side_effect) - exploit_module = _make_module("exploit/unix/webapp/drupal_drupalgeddon2") + exploit_module = _make_module( + "exploit/unix/webapp/drupal_drupalgeddon2", + valid_options={"ForceExploit": {}, "AutoCheck": {}, "RHOSTS": {}}, + ) exploit_module.execute = MagicMock(return_value={"job_id": 1, "uuid": "abc"}) - payload_module = _make_module("payload/php/meterpreter/reverse_tcp") + payload_module = _make_module( + "payload/php/meterpreter/reverse_tcp", + valid_options={"LHOST": {}, "LPORT": {}}, + ) async def get_module_object_stub(module_type, _module_name): return payload_module if module_type == "payload" else exploit_module @@ -131,9 +152,15 @@ def call_side_effect(method, args=None): client.call = MagicMock(side_effect=call_side_effect) - exploit_module = _make_module("exploit/unix/webapp/drupal_drupalgeddon2") + exploit_module = _make_module( + "exploit/unix/webapp/drupal_drupalgeddon2", + valid_options={"ForceExploit": {}, "AutoCheck": {}, "RHOSTS": {}}, + ) exploit_module.execute = MagicMock(return_value={"job_id": 1, "uuid": "abc"}) - payload_module = _make_module("payload/php/meterpreter/reverse_tcp") + payload_module = _make_module( + "payload/php/meterpreter/reverse_tcp", + valid_options={"LHOST": {}, "LPORT": {}}, + ) async def get_module_object_stub(module_type, _module_name): return payload_module if module_type == "payload" else exploit_module @@ -156,3 +183,31 @@ async def get_module_object_stub(module_type, _module_name): assert result["status"] == "warning" assert "ForceExploit=true" in result["message"] + + +@pytest.mark.asyncio +async def test_run_exploit_skips_unsupported_force_options(monkeypatch): + exploit_module = _make_module( + "exploit/multi/http/php_cgi_arg_injection", + valid_options={"RHOSTS": {}, "RPORT": {}, "TARGETURI": {}}, + ) + monkeypatch.setattr(metasploit_mcp, "_get_module_object", AsyncMock(return_value=exploit_module)) + + with patch.object( + metasploit_mcp, + "_execute_module_rpc", + new=AsyncMock(return_value={"status": "success"}), + ) as mock_execute_module_rpc: + result = await metasploit_mcp.run_exploit( + module="multi/http/php_cgi_arg_injection", + options={"RHOSTS": "192.0.2.10", "RPORT": 80, "TARGETURI": "/"}, + payload="php/meterpreter/reverse_tcp", + payload_options={"LHOST": "192.0.2.5", "LPORT": 4444}, + run_as_job=True, + force_exploit=True, + ) + + assert result["status"] == "success" + called_options = mock_execute_module_rpc.await_args.kwargs["module_options"] + assert "ForceExploit" not in called_options + assert "AutoCheck" not in called_options