Class FindingCorrelator

modules/ai/correlator.py:43–390

Correlate and deduplicate findings across scans.

class FindingCorrelator:
    """Correlate and deduplicate findings across scans."""

    # Severity weights for risk scoring
    SEVERITY_WEIGHTS = {
        Severity.CRITICAL: 10.0,
        Severity.HIGH: 7.0,
        Severity.MEDIUM: 4.0,
        Severity.LOW: 2.0,
        Severity.INFO: 0.5,
    }

    # Attack chain patterns
    ATTACK_PATTERNS = {
        "web_compromise": {
            "name": "Web Application Compromise",
            "description": "Path from reconnaissance to potential system compromise via web vulnerabilities",
            "stages": [
                ["osint", "tech_detect"],  # Recon
                ["webscanner.crawler", "webscanner.dirbrute"],  # Discovery
                ["webscanner.xss", "webscanner.sqli"],  # Exploitation
            ],
            "mitre": ["TA0043", "TA0001", "TA0002"],  # Recon, Initial Access, Execution
        },
        "exposed_services": {
            "name": "Exposed Service Attack",
            "description": "Exposed services leading to potential unauthorized access",
            "stages": [
                ["osint.port_scan", "osint.shodan"],  # Service discovery
                ["exploit.searchsploit", "exploit.metasploit"],  # Exploit research
            ],
            "mitre": ["TA0043", "TA0001"],
        },
        "credential_exposure": {
            "name": "Credential Exposure Risk",
            "description": "Paths that could lead to credential theft",
            "stages": [
                ["osint.email_harvest"],  # Target identification
                ["webscanner.xss", "phishing"],  # Credential capture
            ],
            "mitre": ["TA0043", "TA0006"],  # Recon, Credential Access
        },
    }

    def __init__(self):
        self.logger = get_logger("ai.correlator")

    def correlate(self, scan_results: list[ScanResult]) -> CorrelationReport:
        """Correlate findings across multiple scan results.

        Args:
            scan_results: List of scan results to correlate

        Returns:
            CorrelationReport with analysis
        """
        if not scan_results:
            return CorrelationReport(
                target="",