exploit_intel.py

# Copyright (c) 2026 Omar Rao
# SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-Commercial
# Available under the GNU Affero General Public License v3.0, or under a
# separate commercial license. See LICENSE and COMMERCIAL-LICENSE.md.

"""
Exploitability enrichment for dependency CVEs (SCA prioritisation).

Turns a flat list of dependency vulnerabilities into a *prioritised* one by
answering two questions that severity/CVSS alone cannot:

  - EPSS  — the FIRST.org Exploit Prediction Scoring System probability that a
            CVE will be exploited in the wild in the next 30 days (0..1).
  - KEV   — whether the CVE is on CISA's Known Exploited Vulnerabilities
            catalogue (i.e. confirmed exploited, not just theoretically).

Both sources are free and require no API key. Results are cached in-memory so
scans are never blocked and the upstream services are not hammered. Every call
is best-effort: if a source is unreachable the enrichment degrades gracefully
and the report simply shows no EPSS/KEV data rather than failing.

This is pure SCA intelligence — it only reads public advisory data about CVEs
already found in the target's manifests. It performs no active testing.
"""

import json
import time
import urllib.parse
import urllib.request

_TTL = 6 * 3600  # 6 hours — EPSS refreshes daily, KEV a few times a week
_EPSS_URL = "https://api.first.org/data/v1/epss"
_KEV_URL = "https://www.cisa.gov/sites/default/files/feeds/known_exploited_vulnerabilities.json"

_kev_cache = {"set": None, "ts": 0.0}
_epss_cache: dict[str, dict] = {}  # cve -> {"epss": float, "pct": float, "ts": float}


def _fetch_json(url: str, timeout: int = 12):
    req = urllib.request.Request(url, headers={
        "User-Agent": "SecureScope/1.0 (+https://github.com/OmarRao/secure-scope)",
        "Accept": "application/json",
    })
    with urllib.request.urlopen(req, timeout=timeout) as r:
        return json.loads(r.read().decode("utf-8", "replace"))


def kev_set() -> set:
    """Return the set of CVE IDs on the CISA KEV catalogue (cached)."""
    now = time.time()
    if _kev_cache["set"] is not None and (now - _kev_cache["ts"]) < _TTL:
        return _kev_cache["set"]
    try:
        d = _fetch_json(_KEV_URL)
        s = {v.get("cveID", "").strip().upper()
             for v in d.get("vulnerabilities", []) if v.get("cveID")}
        _kev_cache["set"] = s
        _kev_cache["ts"] = now
        return s
    except Exception:
        # Serve stale data if we have any; otherwise an empty set.
        return _kev_cache["set"] or set()


def epss_scores(cve_ids) -> dict:
    """Return {CVE: {"epss": float, "pct": float}} for the given CVE IDs.

    Uses cached values where fresh and batches the rest into EPSS API queries
    (chunked to keep the URL length sane).
    """
    now = time.time()
    ids = sorted({c.strip().upper() for c in cve_ids if c and c.upper().startswith("CVE-")})
    out: dict[str, dict] = {}
    missing = []
    for c in ids:
        hit = _epss_cache.get(c)
        if hit and (now - hit["ts"]) < _TTL:
            out[c] = {"epss": hit["epss"], "pct": hit["pct"]}
        else:
            missing.append(c)

    for i in range(0, len(missing), 80):
        chunk = missing[i:i + 80]
        try:
            q = urllib.parse.urlencode({"cve": ",".join(chunk)})
            d = _fetch_json(f"{_EPSS_URL}?{q}")
            returned = set()
            for row in d.get("data", []):
                cve = (row.get("cve") or "").strip().upper()
                if not cve:
                    continue
                try:
                    epss = float(row.get("epss", 0) or 0)
                    pct = float(row.get("percentile", 0) or 0)
                except (TypeError, ValueError):
                    epss, pct = 0.0, 0.0
                _epss_cache[cve] = {"epss": epss, "pct": pct, "ts": now}
                out[cve] = {"epss": epss, "pct": pct}
                returned.add(cve)
            # Cache negative results too so we don't re-query CVEs EPSS lacks.
            for cve in chunk:
                if cve not in returned:
                    _epss_cache[cve] = {"epss": 0.0, "pct": 0.0, "ts": now}
        except Exception:
            # Leave this chunk unenriched; caller handles missing entries.
            continue
    return out


def _cves_of(vuln: dict) -> list:
    """All CVE IDs associated with a dependency-vuln dict."""
    cves = set()
    for a in (vuln.get("aliases") or []):
        if a and a.upper().startswith("CVE-"):
            cves.add(a.strip().upper())
    for key in ("primary_cve", "vuln_id"):
        v = vuln.get(key)
        if v and str(v).upper().startswith("CVE-"):
            cves.add(str(v).strip().upper())
    return sorted(cves)


def enrich_deps(deps: dict) -> dict:
    """Enrich a DepScanResult.to_dict() in place with EPSS + KEV, and re-sort.

    Adds to each vulnerability:
      - epss      : float 0..1        (highest EPSS across its CVEs)
      - epss_pct  : float 0..1        (percentile for that CVE)
      - kev       : bool              (any of its CVEs on CISA KEV)
    Adds to the top-level dict:
      - kev_count : int               (vulns with a KEV-listed CVE)
      - max_epss  : float 0..1        (highest EPSS observed)
    Best-effort: on any failure the dict is returned unchanged.
    """
    if not deps or not isinstance(deps, dict):
        return deps
    vulns = deps.get("vulnerabilities") or []
    if not vulns:
        deps.setdefault("kev_count", 0)
        deps.setdefault("max_epss", 0.0)
        return deps

    try:
        all_cves = set()
        for v in vulns:
            all_cves.update(_cves_of(v))
        scores = epss_scores(all_cves)
        kev = kev_set()

        kev_count = 0
        max_epss = 0.0
        for v in vulns:
            cves = _cves_of(v)
            best_epss, best_pct = 0.0, 0.0
            is_kev = False
            for c in cves:
                s = scores.get(c)
                if s and s["epss"] >= best_epss:
                    best_epss, best_pct = s["epss"], s["pct"]
                if c in kev:
                    is_kev = True
            v["epss"] = round(best_epss, 5)
            v["epss_pct"] = round(best_pct, 5)
            v["kev"] = is_kev
            if is_kev:
                kev_count += 1
            max_epss = max(max_epss, best_epss)

        # Prioritise: KEV first, then EPSS desc, then CVSS desc.
        _sev_rank = {"CRITICAL": 4, "HIGH": 3, "MEDIUM": 2, "LOW": 1, "UNKNOWN": 0}
        vulns.sort(key=lambda v: (
            1 if v.get("kev") else 0,
            v.get("epss", 0.0),
            _sev_rank.get(str(v.get("severity", "")).upper(), 0),
            v.get("cvss_score", 0.0),
        ), reverse=True)

        deps["kev_count"] = kev_count
        deps["max_epss"] = round(max_epss, 5)
    except Exception:
        deps.setdefault("kev_count", 0)
        deps.setdefault("max_epss", 0.0)
    return deps