admin / Synapse-NetscanXi

"""
CVE enrichment: CISA KEV (known-exploited) catalog + optional NVD lookups.

Both are best-effort and fully degrade if offline: enrichment never raises,
it just leaves the base nmap data untouched. KEV is a single small JSON file
we cache locally; NVD is rate-limited so we only query CVEs we haven't seen.
"""

import json
import os
import time
import urllib.request

KEV_URL = "https://www.cisa.gov/sites/default/files/feeds/known_exploited_vulnerabilities.json"
NVD_API = "https://services.nvd.nist.gov/rest/json/cves/2.0?cveId="
CACHE_DIR = os.path.join(os.path.dirname(os.path.dirname(__file__)), "data", "cache")
KEV_CACHE = os.path.join(CACHE_DIR, "kev.json")
NVD_CACHE = os.path.join(CACHE_DIR, "nvd.json")
KEV_TTL = 24 * 3600
HTTP_TIMEOUT = 8


def _http_get_json(url):
    req = urllib.request.Request(url, headers={"User-Agent": "netscan-xi/1.0"})
    with urllib.request.urlopen(req, timeout=HTTP_TIMEOUT) as resp:
        return json.loads(resp.read().decode("utf-8", "replace"))


def _load_json(path, default):
    try:
        with open(path, "r", encoding="utf-8") as fh:
            return json.load(fh)
    except (OSError, ValueError):
        return default


def _save_json(path, obj):
    os.makedirs(os.path.dirname(path), exist_ok=True)
    tmp = path + ".tmp"
    try:
        with open(tmp, "w", encoding="utf-8") as fh:
            json.dump(obj, fh)
        os.replace(tmp, path)
    except OSError:
        pass


def load_kev(force=False):
    """Return a set of CVE ids in CISA's Known Exploited Vulnerabilities list."""
    cached = _load_json(KEV_CACHE, None)
    if cached and not force:
        if time.time() - cached.get("_fetched", 0) < KEV_TTL:
            return set(cached.get("cves", []))
    try:
        data = _http_get_json(KEV_URL)
        cves = [v["cveID"].upper() for v in data.get("vulnerabilities", [])
                if v.get("cveID")]
        # Capture CISA's authoritative remediation guidance per CVE.
        actions = {}
        for v in data.get("vulnerabilities", []):
            cid = (v.get("cveID") or "").upper()
            if not cid:
                continue
            action = (v.get("requiredAction") or "").strip()
            due = (v.get("dueDate") or "").strip()
            if action:
                actions[cid] = {"action": action, "due": due}
        _save_json(KEV_CACHE, {"_fetched": time.time(), "cves": cves,
                               "actions": actions})
        return set(cves)
    except Exception:
        if cached:
            return set(cached.get("cves", []))
        return set()


def load_kev_actions():
    """Return {CVE: {action, due}} from the cached KEV catalog (best-effort)."""
    cached = _load_json(KEV_CACHE, None)
    if cached and isinstance(cached.get("actions"), dict):
        return cached["actions"]
    return {}


def _nvd_lookup(cve_id, cache):
    if cve_id in cache:
        return cache[cve_id]
    try:
        data = _http_get_json(NVD_API + cve_id)
        vulns = data.get("vulnerabilities", [])
        if not vulns:
            cache[cve_id] = {}
            return {}
        cve = vulns[0]["cve"]
        desc = ""
        for d in cve.get("descriptions", []):
            if d.get("lang") == "en":
                desc = d.get("value", "")
                break
        cvss = ""
        metrics = cve.get("metrics", {})
        for key in ("cvssMetricV31", "cvssMetricV30", "cvssMetricV2"):
            if metrics.get(key):
                cvss = str(metrics[key][0]["cvssData"]["baseScore"])
                break
        info = {"description": desc, "cvss": cvss}
        cache[cve_id] = info
        return info
    except Exception:
        return {}


# Service/port -> generic hardening advice used when no CVE-specific guidance
# is available. Keyed by nmap service name.
SERVICE_MITIGATION = {
    "telnet": "Disable Telnet and use SSH instead; it transmits credentials in cleartext.",
    "ftp": "Replace plain FTP with SFTP/FTPS; disable anonymous access.",
    "smb": "Patch SMB, disable SMBv1, and restrict access with firewall rules.",
    "microsoft-ds": "Patch SMB, disable SMBv1, and restrict access with firewall rules.",
    "rdp": "Restrict RDP to a VPN, enable NLA, and apply the latest Windows patches.",
    "vnc": "Put VNC behind a VPN, require strong auth, and avoid exposing it on the LAN.",
    "snmp": "Disable SNMP or move to SNMPv3 with auth+privacy; never use 'public' community.",
    "http": "Update the web server/app to the latest version and restrict admin interfaces.",
    "https": "Update the web server/app and TLS stack to the latest version.",
    "upnp": "Disable UPnP on the device unless strictly required.",
}


def suggest_mitigation(vuln, kev_actions):
    """Return a human-readable mitigation suggestion for a vuln dict."""
    vid = (vuln.get("id") or "").upper()
    # 1) CISA's authoritative required action for known-exploited CVEs.
    if vid in kev_actions and kev_actions[vid].get("action"):
        a = kev_actions[vid]["action"]
        due = kev_actions[vid].get("due")
        return a + (f" (CISA remediation due {due})." if due else "")
    # 2) Service-specific hardening advice.
    svc = (vuln.get("service") or "").lower()
    if svc in SERVICE_MITIGATION:
        return SERVICE_MITIGATION[svc]
    # 3) Product/version guidance if we know what's affected.
    product = (vuln.get("product") or "").strip()
    version = (vuln.get("version") or "").strip()
    patch = (vuln.get("patch") or "").strip()
    if product:
        # v9.1: when we captured a concrete version/patch token, cite it so the
        # remediation is specific to the patch state we actually observed.
        detected = product + (f" {version}" if version else "")
        if patch:
            detected += f" (patch/build {patch})"
        return (f"Detected {detected}. Update to the latest patched release and "
                f"review the vendor advisory for {vid}." if vid.startswith("CVE-")
                else f"Detected {detected}. Update to the latest patched release.")
    # 4) Generic CVE fallback.
    if vid.startswith("CVE-"):
        return ("Apply the vendor patch for this CVE; if none exists, restrict "
                "network access to the affected service and monitor for exploit "
                "attempts. See https://nvd.nist.gov/vuln/detail/" + vid)
    return ("Investigate the affected service, apply available updates, and "
            "restrict network exposure where possible.")


def enrich_hosts(hosts, use_nvd=False, nvd_budget=20):
    """
    Annotate each vuln with kev=True/False, a CVE code, a mitigation suggestion
    and, optionally, NVD cvss/description. Mutates and returns hosts.
    Best-effort; never raises.
    """
    kev = load_kev()
    kev_actions = load_kev_actions()
    nvd_cache = _load_json(NVD_CACHE, {}) if use_nvd else {}
    used = 0

    for h in hosts:
        for v in h.get("vulns", []):
            vid = (v.get("id") or "").upper()
            v["kev"] = vid in kev
            # Surface a clean CVE code field for the UI (empty for non-CVE finds).
            v["cve"] = vid if vid.startswith("CVE-") else ""
            # v9.1: classify how the finding was matched so the UI can show
            # confidence. version-confirmed (product+version known) is far more
            # reliable than a distro-version heuristic.
            if v.get("confirmed") or (v.get("product") and v.get("version")):
                v["match_basis"] = "version-confirmed"
            elif (v.get("source") or "").lower().startswith("vulners"):
                v["match_basis"] = "version-confirmed"
            else:
                v["match_basis"] = "heuristic"
            if use_nvd and vid.startswith("CVE-") and used < nvd_budget:
                if not v.get("cvss") or not v.get("description"):
                    info = _nvd_lookup(vid, nvd_cache)
                    if vid not in nvd_cache or info:
                        used += 1
                    if info.get("cvss") and not v.get("cvss"):
                        v["cvss"] = info["cvss"]
                    if info.get("description"):
                        v["description"] = info["description"]
            # Mitigation suggestion (always set).
            v["mitigation"] = suggest_mitigation(v, kev_actions)
        # Re-sort vulns now that KEV/cvss may have changed, KEV first.
        def _key(v):
            try:
                c = float(v.get("cvss") or 0)
            except ValueError:
                c = 0.0
            return (v.get("kev", False), v.get("exploit", False), c)
        h["vulns"].sort(key=_key, reverse=True)

    if use_nvd:
        _save_json(NVD_CACHE, nvd_cache)
    return hosts


def enrich_container_vulns(hosts, use_nvd=False, nvd_budget=20):
    """
    v10r1: apply KEV flagging + mitigation to the CVEs found inside container
    images (docker.image_scan.reports[].vulns). Recomputes KEV totals so the
    UI/dashboard can surface known-exploited container CVEs. Best-effort.
    """
    kev = load_kev()
    kev_actions = load_kev_actions()
    for h in hosts:
        d = h.get("docker") or {}
        scan = d.get("image_scan") or {}
        reports = scan.get("reports") or []
        if not reports:
            continue
        kev_total = 0
        for rep in reports:
            rep_kev = 0
            for v in rep.get("vulns", []):
                vid = (v.get("id") or "").upper()
                v["kev"] = vid in kev
                v["cve"] = vid if vid.startswith("CVE-") else ""
                if v["kev"]:
                    rep_kev += 1
                v["mitigation"] = suggest_mitigation(v, kev_actions)
            # KEV-first ordering within each image.
            def _key(v):
                try:
                    c = float(v.get("cvss") or 0)
                except ValueError:
                    c = 0.0
                return (v.get("kev", False), c)
            rep.get("vulns", []).sort(key=_key, reverse=True)
            rep["kev_count"] = rep_kev
            kev_total += rep_kev
        scan["kev_count"] = kev_total
        d.setdefault("summary", {})["image_kev"] = kev_total
    return hosts