vulncheck
10fa688a91
Tester report: "Network error" on the Exploit Intel button. Backend was iterating every CVE-JSON file in nomi-sec/PoC-in-GitHub (~30k requests across 6 year folders), each ~200ms = 100+ min, guaranteed GitHub rate-limit (60 req/h unauthenticated) AND axios client timeout long before backend finishes. Fix - fetch_pocs_github_cve_map() now REQUIRES known_cves: Set[str] and filters each year's directory listing to filenames matching the set before issuing per-CVE downloads. ~30k → typically 50-500 requests per run. - refresh_all_exploit_intel() loads vulns once up front, builds the CVE-id set, passes it in. Removed the second (duplicate) query. - Cache file path now suffixed with sha1(known_cves) so a smaller DB scope can't serve a stale-larger result and vice versa. - 403/429 from GitHub trigger an early-exit warning instead of swallowing the rate-limit silently. Exploit-DB CSV + Metasploit metadata fetchers are still global (small, one HTTP call each, no per-CVE iteration).
416 lines
15 KiB
Python
416 lines
15 KiB
Python
"""
|
|
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/vulncheck-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/vulncheck-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/vulncheck-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
|