nmap_scan
Execute enhanced Nmap scans on targets with real-time logging for version detection, script execution, and custom port scanning to identify network vulnerabilities and services.
Instructions
Execute an enhanced Nmap scan against a target with real-time logging.
Args: target: The IP address or hostname to scan scan_type: Scan type (e.g., -sV for version detection, -sC for scripts) ports: Comma-separated list of ports or port ranges additional_args: Additional Nmap arguments
Returns: Scan results with enhanced telemetry
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| additional_args | No | ||
| ports | No | ||
| scan_type | No | -sV | |
| target | Yes |
Implementation Reference
- src/mcp_server/app.py:159-189 (handler)MCP tool 'nmap_scan' handler: defines the tool interface, parameters, and proxies execution to the REST API server endpoint '/api/nmap' with logging.
@mcp.tool() def nmap_scan( target: str, scan_type: str = "-sV", ports: str = "", additional_args: str = "" ) -> dict[str, Any]: """Execute an enhanced Nmap scan against a target with real-time logging. Args: target: The IP address or hostname to scan scan_type: Scan type (e.g., -sV for version detection, -sC for scripts) ports: Comma-separated list of ports or port ranges additional_args: Additional Nmap arguments Returns: Scan results with enhanced telemetry """ data = { "target": target, "scan_type": scan_type, "ports": ports, "additional_args": additional_args, } logger.info(f"🔍 Starting Nmap scan on {target}") result = api_client.safe_post("api/nmap", data) if result.get("success"): logger.info(f"✅ Nmap scan completed on {target}") else: logger.error("❌ Nmap scan failed") return result - REST API handler 'execute_nmap' for '/api/nmap': builds and executes the nmap command securely, parses XML output into structured findings.
@tool(required_fields=["target"]) def execute_nmap(): """Execute Nmap scan against a target.""" data = request.get_json() params = extract_nmap_params(data) logger.info(f"Executing Nmap scan on {params['target']}") started_at = datetime.now() command = build_nmap_command(params) execution_result = execute_command( command, timeout=1800 ) # 30 minute timeout for nmap ended_at = datetime.now() return parse_nmap_result(execution_result, params, command, started_at, ended_at) - Helper function to securely construct the nmap command line with shell escaping and XML output.
def build_nmap_command(params: dict) -> str: """Build nmap command from parameters with proper security escaping.""" cmd_parts = ["nmap"] # Add scan type if params["scan_type"]: for scan_arg in params["scan_type"].split(): cmd_parts.append(shlex.quote(scan_arg)) # Add timing if params["timing"]: cmd_parts.append(shlex.quote(params["timing"])) # Add ports if specified if params["ports"]: cmd_parts.extend(["-p", shlex.quote(params["ports"])]) # Add host timeout if params["host_timeout"]: cmd_parts.extend(["--host-timeout", shlex.quote(params["host_timeout"])]) # Add max retries if params["max_retries"]: cmd_parts.extend(["--max-retries", str(params["max_retries"])]) # Force XML output for structured parsing cmd_parts.extend(["-oX", "-"]) # Add additional arguments if provided if params["additional_args"]: try: # Split and quote each argument properly additional_parts = shlex.split(params["additional_args"]) for part in additional_parts: cmd_parts.append(shlex.quote(part)) except ValueError as e: # Invalid shell syntax - reject raise ValueError("Invalid additional arguments format") from e # Add target cmd_parts.append(shlex.quote(params["target"])) return " ".join(cmd_parts) - Helper function to parse Nmap XML output into structured findings with host/port/service details and deduplication logic.
def parse_nmap_output(xml_output: str) -> list[dict[str, Any]]: """Parse nmap XML output into basic findings.""" findings = [] if not xml_output.strip(): return findings try: # Parse XML root = ET.fromstring(xml_output) # Iterate through hosts for host in root.findall("host"): # Get host information address_elem = host.find("address") if address_elem is None: continue host_ip = address_elem.get("addr", "") if not host_ip: continue # Get hostname if available hostname = host_ip hostnames_elem = host.find("hostnames") if hostnames_elem is not None: hostname_elem = hostnames_elem.find("hostname") if hostname_elem is not None: hostname = hostname_elem.get("name", host_ip) # Get host state status_elem = host.find("status") host_state = ( status_elem.get("state", "unknown") if status_elem is not None else "unknown" ) # Parse ports ports_elem = host.find("ports") if ports_elem is not None: for port_elem in ports_elem.findall("port"): port_id = int(port_elem.get("portid", 0)) protocol = port_elem.get("protocol", "tcp") # Get port state state_elem = port_elem.find("state") port_state = ( state_elem.get("state", "unknown") if state_elem is not None else "unknown" ) # Get service information service_elem = port_elem.find("service") service_name = "" service_product = "" service_version = "" service_extra = "" if service_elem is not None: service_name = service_elem.get("name", "") service_product = service_elem.get("product", "") service_version = service_elem.get("version", "") service_extra = service_elem.get("extrainfo", "") # Get scripts output if available scripts = [] for script_elem in port_elem.findall("script"): script_id = script_elem.get("id", "") script_output = script_elem.get("output", "") if script_id and script_output: scripts.append({"id": script_id, "output": script_output}) # Basic confidence mapping confidence = "high" if port_state == "open" else "medium" # Build basic tags tags = ["port", "scan", protocol, port_state] if service_name: tags.append(service_name) if scripts: tags.append("scripted") # Create basic finding without severity calculation finding = create_finding( finding_type="port", target=hostname, evidence={ "host_ip": host_ip, "hostname": hostname, "port": port_id, "protocol": protocol, "state": port_state, "service_name": service_name, "service_product": service_product, "service_version": service_version, "service_extra": service_extra, "scripts": scripts, "discovered_by": "nmap", "host_state": host_state, }, severity="info", # Default to info for all findings confidence=confidence, tags=tags, raw_ref=ET.tostring(port_elem, encoding="unicode"), ) findings.append(finding) # If no ports found but host is up, create a host discovery finding if host_state == "up" and ports_elem is None: finding = create_finding( finding_type="port", target=hostname, evidence={ "host_ip": host_ip, "hostname": hostname, "state": host_state, "discovered_by": "nmap", "scan_type": "host_discovery", }, severity="info", confidence="high", tags=["host", "discovery", host_state], raw_ref=ET.tostring(host, encoding="unicode"), ) findings.append(finding) except ET.ParseError as e: logger.error(f"Failed to parse nmap XML output: {e}") # No text fallback - return empty list return [] return findings