vulncheck/app/services/exploit_intel_service.py

"""
Exploit-intel enrichment from third-party catalogs.

Closes the gap noted by tester: "Nutzt Du darüber hinaus auch bereits
alle Infos wie von exploit-db.com? Das ist in der Detailansicht nur
verlinkt — fliessen die Infos in den Score?"

Sources covered by this module:

  1. Exploit-DB (Offensive Security)
     - Source: gitlab.com/exploit-database/exploitdb (raw CSV)
     - File:   files_exploits.csv (~50k rows, ~10 MB)
     - CVE refs in column `codes` (e.g. "CVE-2023-1234;CVE-2023-5678")

  2. PoC-in-GitHub  (nomi-sec)
     - Source: github.com/nomi-sec/PoC-in-GitHub
     - Per-year tree of JSON files (one per CVE)
     - We pull the aggregated JSON dump (mirrored daily) when available.

  3. Metasploit Framework
     - Source: github.com/rapid7/metasploit-framework
     - Modules under modules/exploits/ with `References` field in
       comments — extracted into a flat CVE→[module-path] map.

All three are best-effort: a fetch failure leaves the column as-is,
never wipes a known-good row. Cached on disk + in the `settings`
table for 24h (matches the existing EPSS/KEV cache pattern).

Each enricher returns `dict[cve_id, list[reference]]`. The bulk
runner (`refresh_all_exploit_intel`) walks all open vulns once and
writes the union into the new columns from migration 025.
"""
from __future__ import annotations

import csv
import io
import json
import logging
import os
import re
import tempfile
import time
from datetime import datetime, timedelta
from typing import Dict, List, Optional, Set

import httpx
from sqlalchemy.orm import Session

from app.models.setting import Setting
from app.models.vulnerability import Vulnerability, VulnerabilityStatus

logger = logging.getLogger(__name__)


# ----------------------------------------------------------------------
# Exploit-DB CSV
# ----------------------------------------------------------------------

_EDB_CSV_URL = "https://gitlab.com/exploit-database/exploitdb/-/raw/main/files_exploits.csv"
_EDB_CACHE_PATH = "/tmp/truevuln-exploit-db.csv"
_EDB_CACHE_TTL = 24 * 3600

# Lower-case header → index lookup (the GitLab CSV occasionally
# shuffles columns; resolve by name).
_EDB_CVE_RE = re.compile(r"CVE-\d{4}-\d{4,7}", re.IGNORECASE)


def _cache_fresh(path: str, ttl_seconds: int) -> bool:
    try:
        mtime = os.path.getmtime(path)
        return (time.time() - mtime) < ttl_seconds
    except OSError:
        return False


def _download_to(path: str, url: str, timeout: float = 60.0) -> bool:
    try:
        with httpx.Client(timeout=timeout, follow_redirects=True) as client:
            r = client.get(url)
            r.raise_for_status()
            tmp = path + ".tmp"
            with open(tmp, "wb") as f:
                f.write(r.content)
            os.replace(tmp, path)
            logger.info("downloaded %s → %s (%d bytes)", url, path, len(r.content))
            return True
    except Exception as e:
        logger.warning("download failed for %s: %s", url, e)
        return False


def fetch_exploit_db_cve_map() -> Dict[str, List[str]]:
    """Parse Exploit-DB CSV → {cve_id: [edb_id, ...]}.

    Returns {} on fetch / parse failure — caller skips enrichment for
    that run rather than wiping previously-stored values.
    """
    if not _cache_fresh(_EDB_CACHE_PATH, _EDB_CACHE_TTL):
        if not _download_to(_EDB_CACHE_PATH, _EDB_CSV_URL):
            return {}
    out: Dict[str, List[str]] = {}
    try:
        with open(_EDB_CACHE_PATH, "r", encoding="utf-8", errors="replace") as f:
            reader = csv.DictReader(f)
            for row in reader:
                edb_id = (row.get("id") or "").strip()
                if not edb_id:
                    continue
                # Field name has changed over time — try common spellings.
                codes_blob = (
                    row.get("codes")
                    or row.get("Codes")
                    or row.get("cve_ref")
                    or row.get("references")
                    or ""
                )
                for m in _EDB_CVE_RE.findall(codes_blob):
                    cve = m.upper()
                    out.setdefault(cve, []).append(edb_id)
    except Exception as e:
        logger.warning("exploit-db CSV parse failed: %s", e)
        return {}
    logger.info("Exploit-DB map: %d CVEs with public exploits", len(out))
    return out


# ----------------------------------------------------------------------
# PoC-in-GitHub
# ----------------------------------------------------------------------

# nomi-sec maintains the aggregated dump that mirrors the per-year
# folders into a single JSON: https://github.com/nomi-sec/PoC-in-GitHub
# Direct raw URL — heavy but cacheable. Pattern: one folder per year
# under the repo, each holding {CVE-ID}.json.
_POC_RAW_BASE = "https://raw.githubusercontent.com/nomi-sec/PoC-in-GitHub/master"
_POC_CACHE_PATH = "/tmp/truevuln-pocs-github.json"
_POC_CACHE_TTL = 24 * 3600


def fetch_pocs_github_cve_map(
    known_cves: Optional[Set[str]] = None,
    years: Optional[List[int]] = None,
) -> Dict[str, List[str]]:
    """Per-CVE PoC URLs from PoC-in-GitHub.

    Strategy
      1. List each year-folder (one cheap HTTP call per year).
      2. Filter the listing to filenames matching `known_cves`
         (uppercase CVE-id set). If known_cves is empty/None we DO
         NOT walk the whole repo — that's 30k+ requests = 100 min and
         hits the unauthenticated 60-req/h GitHub rate-limit fast.
         Caller MUST pass the CVE-id set from the DB.
      3. For each matching CVE, fetch its per-CVE JSON file → repo URLs.

    Cached on disk + keyed by the known_cves hash so a repeat call
    with the same DB scope is instant.
    """
    if not known_cves:
        logger.info(
            "PoC-in-GitHub skipped — no known_cves set passed in. "
            "Walking the full repo would burn the 60 req/h GitHub limit."
        )
        return {}

    if years is None:
        current_year = datetime.now().year
        years = list(range(current_year - 5, current_year + 1))

    # Cache key includes the known-cves digest so we never serve stale
    # results scoped to a smaller DB.
    import hashlib
    digest = hashlib.sha1(
        ",".join(sorted(known_cves)).encode("utf-8")
    ).hexdigest()[:12]
    cache_path = f"{_POC_CACHE_PATH}.{digest}"
    if _cache_fresh(cache_path, _POC_CACHE_TTL):
        try:
            with open(cache_path, "r", encoding="utf-8") as f:
                return json.load(f)
        except Exception:
            pass

    out: Dict[str, List[str]] = {}
    try:
        with httpx.Client(timeout=30.0, follow_redirects=True) as client:
            for year in years:
                api = f"https://api.github.com/repos/nomi-sec/PoC-in-GitHub/contents/{year}"
                try:
                    r = client.get(api, headers={"Accept": "application/vnd.github+json"})
                    if r.status_code != 200:
                        if r.status_code in (403, 429):
                            logger.warning(
                                "PoC-in-GitHub: GitHub rate-limit hit on year %d — stopping.",
                                year,
                            )
                            break
                        continue
                    entries = r.json()
                except Exception:
                    continue
                if not isinstance(entries, list):
                    continue
                # Filter to CVEs we actually have in the DB.
                relevant: list[dict] = []
                for entry in entries:
                    name = entry.get("name") or ""
                    if not name.endswith(".json"):
                        continue
                    cve_id = name[:-5].upper()
                    if cve_id in known_cves:
                        relevant.append(entry)
                logger.info(
                    "PoC-in-GitHub year %d: %d / %d entries match DB",
                    year, len(relevant), len(entries),
                )
                for entry in relevant:
                    cve_id = entry["name"][:-5].upper()
                    download_url = entry.get("download_url")
                    if not download_url:
                        continue
                    try:
                        rr = client.get(download_url)
                        if rr.status_code == 403 or rr.status_code == 429:
                            logger.warning(
                                "PoC-in-GitHub: rate-limit on %s — stopping.",
                                cve_id,
                            )
                            break
                        if rr.status_code != 200:
                            continue
                        pocs = rr.json()
                    except Exception:
                        continue
                    if isinstance(pocs, list):
                        urls = [p.get("html_url") for p in pocs if isinstance(p, dict) and p.get("html_url")]
                        if urls:
                            out[cve_id] = urls
        try:
            with open(cache_path, "w", encoding="utf-8") as f:
                json.dump(out, f)
        except Exception:
            pass
    except Exception as e:
        logger.warning("PoC-in-GitHub fetch failed: %s", e)
        return {}
    logger.info("PoC-in-GitHub map: %d CVEs with public PoC links", len(out))
    return out


# ----------------------------------------------------------------------
# Metasploit Framework modules
# ----------------------------------------------------------------------

# rapid7 ships a `db/modules_metadata_base.json` companion file containing
# parsed metadata for every module. CVE refs live under `references`.
_MSF_META_URL = (
    "https://raw.githubusercontent.com/rapid7/metasploit-framework/master/"
    "db/modules_metadata_base.json"
)
_MSF_CACHE_PATH = "/tmp/truevuln-msf-modules.json"
_MSF_CACHE_TTL = 24 * 3600


def fetch_metasploit_cve_map() -> Dict[str, List[str]]:
    """Per-CVE Metasploit module paths."""
    if not _cache_fresh(_MSF_CACHE_PATH, _MSF_CACHE_TTL):
        if not _download_to(_MSF_CACHE_PATH, _MSF_META_URL, timeout=120.0):
            return {}
    out: Dict[str, List[str]] = {}
    try:
        with open(_MSF_CACHE_PATH, "r", encoding="utf-8") as f:
            data = json.load(f)
    except Exception as e:
        logger.warning("metasploit metadata parse failed: %s", e)
        return {}
    if not isinstance(data, dict):
        return {}
    for path, mod in data.items():
        if not isinstance(mod, dict):
            continue
        refs = mod.get("references") or []
        if not isinstance(refs, list):
            continue
        for ref in refs:
            if not isinstance(ref, str):
                continue
            # References come as "CVE-YYYY-NNNN" plain or "CVE,YYYY-NNNN"
            for m in _EDB_CVE_RE.findall(ref) or (
                [f"CVE-{ref[4:]}"] if ref.startswith("CVE,") and "-" in ref[4:] else []
            ):
                cve = m.upper().replace("CVE,", "CVE-")
                if not cve.startswith("CVE-"):
                    continue
                out.setdefault(cve, []).append(path)
    # Dedup per-cve while preserving order
    for k in list(out.keys()):
        seen: Set[str] = set()
        deduped: List[str] = []
        for p in out[k]:
            if p in seen:
                continue
            seen.add(p)
            deduped.append(p)
        out[k] = deduped
    logger.info("Metasploit map: %d CVEs with weaponised modules", len(out))
    return out


# ----------------------------------------------------------------------
# Bulk apply
# ----------------------------------------------------------------------

def refresh_all_exploit_intel(
    db: Session,
    *,
    only_open: bool = True,
    fetch_pocs: bool = True,
    fetch_msf: bool = True,
) -> dict:
    """Walk vulnerabilities, write exploit-intel columns.

    Stats: {edb_marked, edb_cleared, poc_marked, msf_marked, total}.
    """
    # Load vulns first so we can pass the CVE-id set into the PoC
    # fetcher — otherwise it walks the whole 30k-entry repo and hits
    # GitHub's 60 req/h limit.
    q = db.query(Vulnerability)
    if only_open:
        q = q.filter(Vulnerability.status == VulnerabilityStatus.open)
    vulns = q.all()
    known_cves = {
        (v.cve_id or "").upper() for v in vulns
        if v.cve_id and v.cve_id.upper().startswith("CVE-")
    }

    edb_map = fetch_exploit_db_cve_map()
    poc_map = (
        fetch_pocs_github_cve_map(known_cves=known_cves)
        if fetch_pocs else {}
    )
    msf_map = fetch_metasploit_cve_map() if fetch_msf else {}

    stats = {
        "edb_marked": 0, "edb_cleared": 0,
        "poc_marked": 0, "poc_cleared": 0,
        "msf_marked": 0, "msf_cleared": 0,
        "total": 0,
    }
    if not (edb_map or poc_map or msf_map):
        return stats

    # `vulns` already loaded above for the known_cves set.
    now = datetime.now()
    for vuln in vulns:
        cve = (vuln.cve_id or "").upper()
        if not cve.startswith("CVE-"):
            continue
        stats["total"] += 1

        # Exploit-DB
        edb_ids = edb_map.get(cve, [])
        if edb_ids:
            payload = json.dumps(edb_ids[:50])  # cap for sanity
            if vuln.exploit_db_ids != payload:
                vuln.exploit_db_ids = payload
                vuln.exploit_db_count = len(edb_ids)
                stats["edb_marked"] += 1
        else:
            if vuln.exploit_db_count and vuln.exploit_db_count > 0:
                vuln.exploit_db_ids = None
                vuln.exploit_db_count = 0
                stats["edb_cleared"] += 1

        # PoC-in-GitHub
        if fetch_pocs:
            urls = poc_map.get(cve, [])
            if urls:
                payload = json.dumps(urls[:50])
                if vuln.pocs_github_urls != payload:
                    vuln.pocs_github_urls = payload
                    vuln.pocs_github_count = len(urls)
                    stats["poc_marked"] += 1
            else:
                if vuln.pocs_github_count and vuln.pocs_github_count > 0:
                    vuln.pocs_github_urls = None
                    vuln.pocs_github_count = 0
                    stats["poc_cleared"] += 1

        # Metasploit
        if fetch_msf:
            mods = msf_map.get(cve, [])
            if mods:
                payload = json.dumps(mods[:30])
                if vuln.metasploit_modules != payload:
                    vuln.metasploit_modules = payload
                    vuln.metasploit_module_count = len(mods)
                    stats["msf_marked"] += 1
            else:
                if vuln.metasploit_module_count and vuln.metasploit_module_count > 0:
                    vuln.metasploit_modules = None
                    vuln.metasploit_module_count = 0
                    stats["msf_cleared"] += 1

        vuln.exploit_intel_updated_at = now
        try:
            vuln.refresh_scores()
        except Exception:
            pass

    db.commit()
    logger.info(
        "exploit-intel refresh: total=%d, edb_marked=%d, poc_marked=%d, msf_marked=%d",
        stats["total"], stats["edb_marked"], stats["poc_marked"], stats["msf_marked"],
    )
    return stats