Method run_scan

modules/exploit/metasploit.py:179–231

Run a Metasploit scan against target.

(self, target: Target, scan_type: str = "portscan")

    async def run_scan(self, target: Target, scan_type: str = "portscan") -> ScanResult:
        """Run a Metasploit scan against target."""
        result = ScanResult(target=target, module="exploit.metasploit")

        if not self.token:
            if not await self.connect():
                result.errors.append("Failed to connect to Metasploit RPC")
                result.success = False
                result.complete()
                return result

        host = self._extract_host(target.value)

        try:
            if scan_type == "portscan":
                # Use auxiliary TCP port scanner
                options = {
                    "RHOSTS": host,
                    "PORTS": "1-1000",
                    "THREADS": 10,
                }
                scan_result = await self.run_auxiliary(
                    "scanner/portscan/tcp",
                    options
                )

                result.raw_data["scan_result"] = scan_result
                result.add_finding(
                    title="Metasploit Port Scan Complete",
                    description=f"Scanned {host}",
                    severity=Severity.INFO,
                    data=scan_result,
                )

            elif scan_type == "vuln":
                # Search for vulnerabilities
                modules = await self.search_modules(host, "exploit")

                result.raw_data["potential_exploits"] = modules
                if modules:
                    result.add_finding(
                        title="Potential Exploits Found",
                        description=f"Found {len(modules)} potential exploit module(s)",
                        severity=Severity.MEDIUM,
                        data={"modules": [m.get("fullname") for m in modules[:20]]},
                    )

        except Exception as e:
            result.errors.append(f"Metasploit scan failed: {str(e)}")
            result.success = False

        result.complete()
        return result

    async def search_exploits_for_service(
        self,
        service: str,