masscan_high_speed
Perform high-speed port scanning on target networks to identify open ports and services for security assessments and vulnerability discovery.
Instructions
Execute Masscan for high-speed port scanning.
Args: target: Target IP address or CIDR range ports: Port range to scan rate: Packet transmission rate banners: Enable banner grabbing exclude_file: File containing IPs to exclude include_file: File containing IPs to include output_format: Output format (list, xml, json) additional_args: Additional Masscan arguments
Returns: High-speed scan results
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| additional_args | No | ||
| banners | No | ||
| exclude_file | No | ||
| include_file | No | ||
| output_format | No | list | |
| ports | No | 1-65535 | |
| rate | No | ||
| target | Yes |
Implementation Reference
- src/mcp_server/app.py:292-336 (handler)MCP tool handler and registration for 'masscan_high_speed'. Proxies parameters to REST API endpoint '/api/masscan' for execution.def masscan_high_speed( target: str, ports: str = "1-65535", rate: int = 1000, banners: bool = False, exclude_file: str = "", include_file: str = "", output_format: str = "list", additional_args: str = "", ) -> dict[str, Any]: """Execute Masscan for high-speed port scanning. Args: target: Target IP address or CIDR range ports: Port range to scan rate: Packet transmission rate banners: Enable banner grabbing exclude_file: File containing IPs to exclude include_file: File containing IPs to include output_format: Output format (list, xml, json) additional_args: Additional Masscan arguments Returns: High-speed scan results """ data = { "target": target, "ports": ports, "rate": rate, "banners": banners, "exclude_file": exclude_file, "include_file": include_file, "output_format": output_format, "additional_args": additional_args, } logger.info(f"🚀 Starting Masscan high-speed scan on {target}") result = api_client.safe_post("api/masscan", data) if result.get("success"): logger.info(f"✅ Masscan completed on {target}") else: logger.error("❌ Masscan failed") return result
- Core REST API handler for masscan execution, called by MCP tool. Builds command, executes masscan, parses JSON output into findings.@tool(required_fields=["target"]) def execute_masscan(): """Execute Masscan and return structured port findings.""" data = request.get_json() params = extract_masscan_params(data) logger.info("Executing Masscan on %s", params["target"]) started_at = datetime.now() command = build_masscan_command(params) execution_result = execute_command(command, timeout=600) ended_at = datetime.now() return parse_masscan_output(execution_result, params, command, started_at, ended_at)
- Helper function to safely construct the masscan command line from parameters.def build_masscan_command(params: dict[str, Any]) -> str: """Build a masscan command string with safe shell escaping.""" cmd_parts: list[str] = ["masscan"] if params.get("ports"): cmd_parts.extend(["-p", params["ports"]]) cmd_parts.extend(["--rate", str(params["rate"])]) if params.get("banners"): cmd_parts.append("--banners") if params.get("interface"): cmd_parts.extend(["-e", params["interface"]]) if params.get("router_mac"): cmd_parts.extend(["--router-mac", params["router_mac"]]) if params.get("source_ip"): cmd_parts.extend(["--source-ip", params["source_ip"]]) if params.get("exclude_file"): cmd_parts.extend(["--excludefile", params["exclude_file"]]) if params.get("include_file"): cmd_parts.extend(["--includefile", params["include_file"]]) additional_parts = ( shlex.split(params["additional_args"]) if params.get("additional_args") else [] ) has_output_directive = any( part.startswith("-o") or part.startswith("--output") for part in additional_parts ) if not has_output_directive: cmd_parts.extend(["-oJ", "-"]) cmd_parts.extend(additional_parts) target = params.get("target") if target: for item in target.split(): cmd_parts.append(item) return " ".join(shlex.quote(part) for part in cmd_parts)
- Helper function to parse masscan execution output, handle errors, and structure findings with deduplication.def parse_masscan_output( execution_result: dict[str, Any], params: dict[str, Any], command: str, started_at: datetime, ended_at: datetime, ) -> dict[str, Any]: """Parse masscan JSON output into structured findings.""" duration_ms = int((ended_at - started_at).total_seconds() * 1000) if not execution_result["success"]: error_message = ( execution_result.get("stderr") or execution_result.get("error") or "Masscan execution failed" ) error_response, status_code = create_error_response( error_message, stage="exec", details={ "return_code": execution_result.get("return_code"), "command": execution_result.get("command", command), }, status_code=500, ) return { "success": False, "tool": "masscan", "params": params, "command": command, "started_at": started_at.isoformat(), "ended_at": ended_at.isoformat(), "duration_ms": duration_ms, "error": error_response, "findings": [], "stats": {"findings": 0, "dupes": 0, "payload_bytes": 0}, } stdout = execution_result.get("stdout", "") try: findings, duplicates = _parse_masscan_findings(stdout) except MasscanParseError as exc: error_response, status_code = create_error_response( str(exc), stage="parse", details={ "command": execution_result.get("command", command), "output_sample": stdout[:500], }, status_code=500, ) return { "success": False, "tool": "masscan", "params": params, "command": command, "started_at": started_at.isoformat(), "ended_at": ended_at.isoformat(), "duration_ms": duration_ms, "error": error_response, "findings": [], "stats": {"findings": 0, "dupes": 0, "payload_bytes": 0}, } payload_bytes = len(stdout.encode("utf-8")) stats = create_stats(len(findings), duplicates, payload_bytes) return { "success": True, "tool": "masscan", "params": params, "command": command, "started_at": started_at.isoformat(), "ended_at": ended_at.isoformat(), "duration_ms": duration_ms, "findings": findings, "stats": stats, }
- Helper to extract and validate/default masscan parameters from JSON request.def extract_masscan_params(data: dict[str, Any]) -> dict[str, Any]: """Extract parameters for masscan execution.""" return { "target": data["target"], "ports": data.get("ports", "1-65535"), "rate": int(data.get("rate", 1000)), "banners": bool(data.get("banners", False)), "interface": data.get("interface"), "router_mac": data.get("router_mac"), "source_ip": data.get("source_ip"), "exclude_file": data.get("exclude_file"), "include_file": data.get("include_file"), "additional_args": data.get("additional_args", ""), }