autorecon/build-cve-triage.py

#!/usr/bin/env python3
"""Build CVE triage sheets from AutoRecon artifacts."""
from __future__ import annotations

import json
import re
from collections import defaultdict
from datetime import datetime, timezone
from pathlib import Path

RESULTS_DIR = Path("results")
CVE_RE = re.compile(r"CVE-\d{4}-\d{4,7}")
OPEN_PORT_RE = re.compile(r"^(\d+/(?:tcp|udp))\s+open\s+(.*)$")
SCRIPT_URL_RE = re.compile(r"\|\s+(/[^: ]*):\s+(.*CVE-\d{4}-\d{4,7}.*)$")

PRECONDITIONS = {
    "CVE-2011-3192": "Target must run Apache HTTPD with byte-range handling vulnerable to the Apache Range header DoS condition.",
    "CVE-2009-3733": "Target must expose VMware SDK endpoint (/sdk) and allow traversal into host files.",
    "CVE-2011-0966": "Target must be Cisco Unified Operations Manager 8.0/8.5 with vulnerable auditLog.do traversal handling.",
    "CVE-2018-10822": "Target must be affected D-Link router firmware exposing /uir/ traversal path.",
    "CVE-2018-10824": "Target must be affected D-Link router firmware exposing plaintext credential path under /tmp/csman.",
    "CVE-2017-9798": "Target must run Apache HTTP Server with mod_http2 and be susceptible to OptionBleed conditions.",
    "CVE-2005-3299": "Target must run Apache 2.0.x with mod_imap module enabled and vulnerable to cross-site scripting behavior.",
    "CVE-2003-1418": "Target must expose Apache mod_include with vulnerable SSI execution behavior.",
}

REQUIRED_REPRO_STEP = (
    "Before creating a ticket, reproduce vulnerable behavior directly (e.g., crafted request causing data exposure, traversal read, "
    "or exploitable crash) and attach request/response proof. Banner or script signature matches alone are insufficient."
)


def load_text(path: Path) -> str:
    try:
        return path.read_text(encoding="utf-8", errors="ignore")
    except FileNotFoundError:
        return ""


def parse_service_evidence(nmap_paths: list[Path]) -> dict[str, list[str]]:
    evidence: dict[str, list[str]] = defaultdict(list)
    for path in nmap_paths:
        for line in load_text(path).splitlines():
            m = OPEN_PORT_RE.match(line.strip())
            if m:
                endpoint = m.group(1)
                value = m.group(2).strip()
                if value and value not in evidence[endpoint]:
                    evidence[endpoint].append(value)
    return dict(evidence)


def parse_nmap_hits(nmap_paths: list[Path]) -> dict[str, list[dict[str, str]]]:
    hits: dict[str, list[dict[str, str]]] = defaultdict(list)
    for path in nmap_paths:
        lines = load_text(path).splitlines()
        current_endpoint = "unknown"
        for idx, line in enumerate(lines):
            stripped = line.strip()
            m_open = OPEN_PORT_RE.match(stripped)
            if m_open:
                current_endpoint = m_open.group(1)
            for cve in CVE_RE.findall(line):
                endpoint = current_endpoint
                detail = stripped
                m_url = SCRIPT_URL_RE.search(line)
                if m_url:
                    endpoint = m_url.group(1)
                    detail = m_url.group(2)
                # include short context for review
                ctx = "\n".join(lines[max(0, idx - 1): min(len(lines), idx + 2)])
                hits[cve].append(
                    {
                        "source_file": str(path),
                        "endpoint": endpoint,
                        "evidence": detail,
                        "context": ctx,
                    }
                )
    return dict(hits)


def parse_pattern_hits(pattern_path: Path) -> dict[str, int]:
    counts: dict[str, int] = defaultdict(int)
    for line in load_text(pattern_path).splitlines():
        for cve in CVE_RE.findall(line):
            counts[cve] += 1
    return dict(counts)


def infer_status(cve: str, evidence_blob: str) -> tuple[str, str]:
    blob = evidence_blob.lower()
    if "401 unauthorized" in blob:
        return (
            "not-applicable",
            "Only unauthorized responses were observed; vulnerable behavior was not reproduced and signature is likely a false positive.",
        )
    if cve in {"CVE-2017-9798", "CVE-2003-1418"}:
        return (
            "needs-manual-test",
            "Signature-based identification only; exploitability depends on module/configuration and requires targeted validation.",
        )
    return (
        "needs-manual-test",
        "Script/banner evidence exists, but no direct proof of exploit impact was captured in scan output.",
    )


def host_dirs() -> list[Path]:
    return sorted([p for p in RESULTS_DIR.iterdir() if p.is_dir() and (p / "scans").exists()])


def main() -> None:
    generated = datetime.now(timezone.utc).isoformat()
    global_suppressions = []

    for host_dir in host_dirs():
        scans_dir = host_dir / "scans"
        pattern_path = scans_dir / "_patterns.log"
        nmap_paths = sorted(scans_dir.glob("**/*nmap*.txt"))
        if not pattern_path.exists() and not nmap_paths:
            continue

        pattern_counts = parse_pattern_hits(pattern_path)
        nmap_hits = parse_nmap_hits(nmap_paths)
        cves = sorted(set(pattern_counts) | set(nmap_hits))
        if not cves:
            continue

        service_evidence = parse_service_evidence(nmap_paths)
        host_entries = []

        for cve in cves:
            hit_entries = nmap_hits.get(cve, [])
            endpoints = sorted({h["endpoint"] for h in hit_entries if h["endpoint"]}) or ["unknown"]
            endpoint = endpoints[0]

            product_versions = []
            for ep, products in service_evidence.items():
                if ep in endpoint or endpoint in ep or endpoint == "unknown":
                    product_versions.extend(products)
            product_versions = sorted(set(product_versions))
            if not product_versions:
                product_versions = ["No explicit product/version fingerprint in nmap service line."]

            evidence_blob = "\n".join([h["evidence"] for h in hit_entries])
            status, rationale = infer_status(cve, evidence_blob)
            reproducibility = (
                "not-reproduced" if status != "confirmed" else "reproduced"
            )

            sources = []
            if pattern_counts.get(cve, 0):
                sources.append({"file": str(pattern_path), "match_count": pattern_counts[cve]})
            for h in hit_entries:
                sources.append(
                    {
                        "file": h["source_file"],
                        "endpoint": h["endpoint"],
                        "evidence": h["evidence"],
                    }
                )

            entry = {
                "host": host_dir.name,
                "cve": cve,
                "affected_host": host_dir.name,
                "endpoint": endpoint,
                "product_version_evidence": product_versions,
                "exploit_precondition": PRECONDITIONS.get(
                    cve,
                    "Validate affected product/version and exploit path prerequisites before filing.",
                ),
                "reproducibility": reproducibility,
                "disposition": status,
                "disposition_rationale": rationale,
                "required_reproduction_step_before_ticket": REQUIRED_REPRO_STEP,
                "evidence_sources": sources,
            }
            host_entries.append(entry)

            if status == "not-applicable":
                global_suppressions.append(
                    {
                        "host": host_dir.name,
                        "signature": cve,
                        "reason": rationale,
                        "suppression_scope": "host-specific",
                        "source": "cve-triage",
                        "updated_at": generated,
                    }
                )

        host_entries.sort(key=lambda item: item["cve"])

        md_lines = [
            f"# CVE triage sheet - {host_dir.name}",
            "",
            f"Generated (UTC): {generated}",
            "",
            "| CVE | Affected host | Endpoint | Product/version evidence | Exploit precondition | Reproducibility | Disposition |",
            "|---|---|---|---|---|---|---|",
        ]

        for item in host_entries:
            evidence_short = "; ".join(item["product_version_evidence"][:2])
            md_lines.append(
                f"| {item['cve']} | {item['affected_host']} | `{item['endpoint']}` | {evidence_short} | {item['exploit_precondition']} | {item['reproducibility']} | **{item['disposition']}** |"
            )
            md_lines.append(
                f"|  |  |  |  |  |  | Required reproduction gate: {item['required_reproduction_step_before_ticket']} |"
            )

        (scans_dir / "_cve_triage.md").write_text("\n".join(md_lines) + "\n", encoding="utf-8")
        (scans_dir / "_cve_disposition.json").write_text(
            json.dumps(
                {
                    "host": host_dir.name,
                    "generated_at": generated,
                    "required_reproduction_step_before_ticket": REQUIRED_REPRO_STEP,
                    "findings": host_entries,
                },
                indent=2,
            )
            + "\n",
            encoding="utf-8",
        )

    if global_suppressions:
        (RESULTS_DIR / "_false_positive_suppressions.json").write_text(
            json.dumps(
                {
                    "generated_at": generated,
                    "description": "Host-level CVE suppressions derived from triage dispositions.",
                    "suppressions": sorted(
                        global_suppressions,
                        key=lambda item: (item["host"], item["signature"]),
                    ),
                },
                indent=2,
            )
            + "\n",
            encoding="utf-8",
        )


if __name__ == "__main__":
    main()