feat(app-cve-scan): cvelistV5 range-based detection for installed software
Catches CVEs the NVD-CPE scanner misses: fresh CVEs NVD hasn't CPE'd yet, or
ones filed under a different CPE product string than we curated (TeamViewer
lives under teamviewer:remote, not teamviewer:teamviewer). Matches directly
against cvelistV5 affected[].vendor/product + version ranges — the
authoritative MITRE feed we already cache as a ZIP.
- Curated product registry (name-regex → cvelistV5 vendor/product pairs):
TeamViewer, Notepad++, Devolutions RDM, 7-Zip, Firefox, Chrome, VLC,
PuTTY, WinSCP, Wireshark, FileZilla, Zoom. Unknown software ignored.
- build_product_index: one walk over the cached ZIP → {product_key:
[{cve,start,lt,lte}]} for curated products only; cached in a Setting,
rebuilt by the nightly job (the 557 MB walk happens once, not per scan).
- scan_asset: resolve installed software → indexed CVEs → version-range
check → upsert (source 'app-scan', shared badge/cross-confirm/enrichment).
- Wired into run_app_cve_scan (loads cached index; skipped+logged if not
built yet) and the nightly job (builds index first).
Verified against the real CVE JSON: TeamViewer CVE-2026-23572 (<15.74.5),
Notepad++ CVE-2026-52885 (<8.9.6.4), Devolutions CVE-2026-13372
(2026.2.5–2026.2.11) all detect at affected versions and correctly do NOT
match patched versions.
This commit is contained in:
+7
-1
@@ -646,9 +646,15 @@ def app_cve_scan_nightly():
|
||||
Slotted at 03:25 UTC — after M365 (03:20), before the audit prune (03:30).
|
||||
Cache (TTL 7d) keeps OSV/NVD load bounded; NVD_API_KEY recommended.
|
||||
"""
|
||||
from app.services import app_cve_scanner_service
|
||||
from app.services import app_cve_scanner_service, cvelistv5_scan_service
|
||||
db = SessionLocal()
|
||||
try:
|
||||
# Rebuild the cvelistV5 reverse index first (one ~557 MB ZIP walk) so
|
||||
# the scan below has fresh product→CVE ranges for curated software.
|
||||
try:
|
||||
cvelistv5_scan_service.build_product_index(db)
|
||||
except Exception as e:
|
||||
logger.warning("cvelistV5 index build failed (non-fatal): %s", e)
|
||||
stats = app_cve_scanner_service.run_app_cve_scan(db)
|
||||
logger.info("App CVE scan nightly: %s", {k: v for k, v in stats.items() if k != "errors"})
|
||||
except Exception as e:
|
||||
|
||||
@@ -503,6 +503,18 @@ def run_app_cve_scan(db: Session, asset_id: Optional[int] = None) -> dict:
|
||||
except Exception as e:
|
||||
logger.debug("app-cve: graph client unavailable: %s", e)
|
||||
|
||||
# cvelistV5 reverse index (curated products) — catches CVEs NVD hasn't
|
||||
# CPE'd yet / filed under a different CPE product string. Cached; built by
|
||||
# the nightly job. Absent → that pass is skipped (logged).
|
||||
cve5_index = {}
|
||||
try:
|
||||
from app.services import cvelistv5_scan_service
|
||||
cve5_index = cvelistv5_scan_service.load_index(db) or {}
|
||||
if not cve5_index:
|
||||
logger.info("app-cve: cvelistV5 index not built yet — run the nightly job to enable it")
|
||||
except Exception as e:
|
||||
logger.debug("app-cve: cvelistV5 index load failed: %s", e)
|
||||
|
||||
q = db.query(Asset)
|
||||
if asset_id is not None:
|
||||
q = q.filter(Asset.id == asset_id)
|
||||
@@ -529,6 +541,12 @@ def run_app_cve_scan(db: Session, asset_id: Optional[int] = None) -> dict:
|
||||
packages = []
|
||||
if packages:
|
||||
stats["findings"] += scan_asset_packages(db, asset, packages, new_ids)
|
||||
if cve5_index:
|
||||
try:
|
||||
from app.services import cvelistv5_scan_service
|
||||
stats["findings"] += cvelistv5_scan_service.scan_asset(db, asset, packages, cve5_index, new_ids)
|
||||
except Exception as e:
|
||||
stats["errors"].append(f"asset {asset.id} cvelistv5: {e}")
|
||||
touched = True
|
||||
|
||||
if touched:
|
||||
|
||||
@@ -0,0 +1,280 @@
|
||||
"""
|
||||
cvelistV5-based CVE detection for installed software.
|
||||
|
||||
The NVD-CPE scanner (app_cve_scanner_service) misses CVEs when NVD hasn't
|
||||
published a CPE yet (fresh CVEs) or files them under a different CPE product
|
||||
string than we curated (e.g. TeamViewer CVE lives under teamviewer:remote,
|
||||
not teamviewer:teamviewer). cvelistV5 — the authoritative MITRE feed we
|
||||
already cache as a ZIP — carries clean affected[].vendor/product/version
|
||||
ranges instead, so we match those directly.
|
||||
|
||||
Design (CURATED + PRECISE, same as the CPE scanner):
|
||||
- A curated product registry maps an installed-software name → the
|
||||
cvelistV5 (vendor, product) pairs that identify it. Unknown software is
|
||||
ignored (no fuzzy vendor/product guessing → no FP storm).
|
||||
- One pass over the cached cvelistV5 ZIP builds a reverse index
|
||||
{product_key: [{cve, start, lt, lte}]} for the curated products only.
|
||||
The index is cached in a Setting (refreshed on the nightly job) so the
|
||||
557 MB walk happens once, not per scan.
|
||||
- Per installed package: resolve → look up indexed CVEs → check the
|
||||
installed version falls inside an affected range → upsert (source
|
||||
'app-scan', shared with the CPE scanner so the badge / cross-confirm /
|
||||
enrichment all apply).
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import logging
|
||||
import os
|
||||
import re
|
||||
import time
|
||||
import zipfile
|
||||
from datetime import datetime, timedelta
|
||||
from typing import Dict, List, Optional, Tuple
|
||||
|
||||
from sqlalchemy.orm import Session
|
||||
|
||||
from app.services import app_cve_scanner_service as cpe
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
# Reuse the same ZIP the override service already downloads/caches (12h).
|
||||
_ZIP_PATH = "/tmp/vulncheck-cvelistv5-cache.zip"
|
||||
_ZIP_URL = "https://github.com/CVEProject/cvelistV5/archive/refs/heads/main.zip"
|
||||
_ZIP_TTL = 12 * 3600
|
||||
_INDEX_SETTING = "cvelistv5_product_index"
|
||||
_INDEX_TTL = timedelta(hours=26) # rebuilt nightly; a missed night still serves
|
||||
|
||||
# Curated registry: name-regex (installed software) → cvelistV5 (vendor,
|
||||
# product) pairs, lowercased. First regex match wins. Vendor/product are
|
||||
# matched case-insensitively against affected[].vendor / .product.
|
||||
# ponytail: curated; add a row when a product is missed — unknown names are
|
||||
# skipped, never guessed.
|
||||
_REGISTRY: List[dict] = [
|
||||
{"key": "teamviewer", "re": r"teamviewer",
|
||||
"pairs": [("teamviewer", "remote"), ("teamviewer", "tensor"),
|
||||
("teamviewer", "host"), ("teamviewer", "full client"),
|
||||
("teamviewer", "teamviewer")]},
|
||||
{"key": "notepad++", "re": r"notepad\+\+",
|
||||
"pairs": [("notepad-plus-plus", "notepad-plus-plus"), ("notepad++", "notepad++"),
|
||||
("don ho", "notepad++")]},
|
||||
{"key": "devolutions-rdm", "re": r"remote desktop manager|devolutions",
|
||||
"pairs": [("devolutions", "remote desktop manager")]},
|
||||
{"key": "7-zip", "re": r"7-?zip",
|
||||
"pairs": [("7-zip", "7-zip"), ("igor pavlov", "7-zip")]},
|
||||
{"key": "firefox", "re": r"mozilla firefox|(?<!\w)firefox",
|
||||
"pairs": [("mozilla", "firefox")]},
|
||||
{"key": "chrome", "re": r"google chrome",
|
||||
"pairs": [("google", "chrome")]},
|
||||
{"key": "vlc", "re": r"vlc media player|videolan",
|
||||
"pairs": [("videolan", "vlc media player"), ("videolan", "vlc")]},
|
||||
{"key": "putty", "re": r"(?<!\w)putty",
|
||||
"pairs": [("putty", "putty"), ("simon tatham", "putty")]},
|
||||
{"key": "winscp", "re": r"winscp", "pairs": [("winscp", "winscp"), ("martin prikryl", "winscp")]},
|
||||
{"key": "wireshark", "re": r"wireshark", "pairs": [("wireshark", "wireshark")]},
|
||||
{"key": "filezilla", "re": r"filezilla", "pairs": [("filezilla", "filezilla"), ("filezilla", "filezilla client")]},
|
||||
{"key": "zoom", "re": r"(?<!\w)zoom(?!\w)", "pairs": [("zoom", "zoom"), ("zoom", "meetings"), ("zoom", "zoom client for meetings")]},
|
||||
]
|
||||
|
||||
_COMPILED = [(re.compile(e["re"], re.I), e) for e in _REGISTRY]
|
||||
|
||||
# Reverse map (vendor_lc, product_lc) → product_key, for the index build.
|
||||
_PAIR_TO_KEY: Dict[Tuple[str, str], str] = {}
|
||||
for _e in _REGISTRY:
|
||||
for _v, _p in _e["pairs"]:
|
||||
_PAIR_TO_KEY[(_v.lower(), _p.lower())] = _e["key"]
|
||||
|
||||
|
||||
def resolve(name: str) -> Optional[str]:
|
||||
n = (name or "").strip().lower()
|
||||
if not n:
|
||||
return None
|
||||
for rx, e in _COMPILED:
|
||||
if rx.search(n):
|
||||
return e["key"]
|
||||
return None
|
||||
|
||||
|
||||
# ---------- affected[] range parsing ----------
|
||||
def _ranges_from_affected(aff: dict) -> List[Tuple[Optional[str], Optional[str], Optional[str]]]:
|
||||
"""→ [(version_start, lessThan, lessThanOrEqual)] for the affected entry.
|
||||
Only ranges with a real upper bound are returned (exact-version-only and
|
||||
unbounded entries are skipped → no over-matching)."""
|
||||
out = []
|
||||
for v in aff.get("versions", []) or []:
|
||||
if not isinstance(v, dict):
|
||||
continue
|
||||
if (v.get("status") or "affected") != "affected":
|
||||
continue
|
||||
start = None
|
||||
lt = v.get("lessThan")
|
||||
lte = v.get("lessThanOrEqual")
|
||||
ver = v.get("version")
|
||||
if isinstance(ver, str):
|
||||
vs = ver.strip()
|
||||
if vs.startswith("<="):
|
||||
lte = lte or vs[2:].strip()
|
||||
elif vs.startswith("<"):
|
||||
lt = lt or vs[1:].strip()
|
||||
elif vs not in ("0", "*", "-", ""):
|
||||
start = vs
|
||||
if lt in ("*", "-", ""):
|
||||
lt = None
|
||||
if lte in ("*", "-", ""):
|
||||
lte = None
|
||||
if lt or lte:
|
||||
out.append((start, lt, lte))
|
||||
return out
|
||||
|
||||
|
||||
def _affected(installed: str, start: Optional[str], lt: Optional[str], lte: Optional[str]) -> bool:
|
||||
if start:
|
||||
c = cpe._vcmp(installed, start)
|
||||
if c is None or c < 0:
|
||||
return False
|
||||
if lt:
|
||||
c = cpe._vcmp(installed, lt)
|
||||
return c is not None and c < 0
|
||||
if lte:
|
||||
c = cpe._vcmp(installed, lte)
|
||||
return c is not None and c <= 0
|
||||
return False
|
||||
|
||||
|
||||
# ---------- index build + cache ----------
|
||||
def _ensure_zip() -> bool:
|
||||
fresh = (os.path.exists(_ZIP_PATH)
|
||||
and (time.time() - os.path.getmtime(_ZIP_PATH)) < _ZIP_TTL
|
||||
and os.path.getsize(_ZIP_PATH) > 100_000_000)
|
||||
if fresh:
|
||||
return True
|
||||
import httpx
|
||||
tmp = _ZIP_PATH + ".part"
|
||||
try:
|
||||
logger.info("cvelistv5-scan: downloading ZIP snapshot…")
|
||||
with httpx.Client(timeout=httpx.Timeout(120.0, connect=15.0), follow_redirects=True) as c:
|
||||
with c.stream("GET", _ZIP_URL) as r:
|
||||
r.raise_for_status()
|
||||
with open(tmp, "wb") as f:
|
||||
for chunk in r.iter_bytes(chunk_size=1024 * 512):
|
||||
f.write(chunk)
|
||||
os.replace(tmp, _ZIP_PATH)
|
||||
return True
|
||||
except Exception as e:
|
||||
logger.warning("cvelistv5-scan: ZIP download failed: %s", e)
|
||||
if os.path.exists(tmp):
|
||||
try:
|
||||
os.remove(tmp)
|
||||
except Exception:
|
||||
pass
|
||||
return False
|
||||
|
||||
|
||||
def build_product_index(db: Session) -> dict:
|
||||
"""Walk the cvelistV5 ZIP once, building {product_key: [{cve,start,lt,lte}]}
|
||||
for the curated products. Heavy (~250k files) → call from the nightly job.
|
||||
Caches the result in a Setting. Returns the index."""
|
||||
if not _ensure_zip():
|
||||
return load_index(db) or {}
|
||||
index: Dict[str, list] = {}
|
||||
seen: set = set()
|
||||
scanned = 0
|
||||
with zipfile.ZipFile(_ZIP_PATH) as zf:
|
||||
for name in zf.namelist():
|
||||
if not name.endswith(".json") or "/cves/" not in name:
|
||||
continue
|
||||
scanned += 1
|
||||
try:
|
||||
data = json.loads(zf.read(name))
|
||||
except Exception:
|
||||
continue
|
||||
cna = (data.get("containers") or {}).get("cna") or {}
|
||||
affected = cna.get("affected") or []
|
||||
if not affected:
|
||||
continue
|
||||
cve_id = ((data.get("cveMetadata") or {}).get("cveId") or "").upper()
|
||||
if not cve_id.startswith("CVE-"):
|
||||
continue
|
||||
for aff in affected:
|
||||
key = _PAIR_TO_KEY.get(((aff.get("vendor") or "").lower(),
|
||||
(aff.get("product") or "").lower()))
|
||||
if not key:
|
||||
continue
|
||||
for start, lt, lte in _ranges_from_affected(aff):
|
||||
sig = (key, cve_id, start, lt, lte)
|
||||
if sig in seen:
|
||||
continue
|
||||
seen.add(sig)
|
||||
index.setdefault(key, []).append(
|
||||
{"cve": cve_id, "start": start, "lt": lt, "lte": lte})
|
||||
_store_index(db, index)
|
||||
logger.info("cvelistv5-scan: index built from %d CVE files → %d products, %d ranges",
|
||||
scanned, len(index), sum(len(v) for v in index.values()))
|
||||
return index
|
||||
|
||||
|
||||
def _store_index(db: Session, index: dict) -> None:
|
||||
from app.models.setting import Setting
|
||||
payload = json.dumps({"built_at": datetime.now().isoformat(), "index": index})
|
||||
row = db.query(Setting).filter(Setting.key == _INDEX_SETTING).first()
|
||||
if row:
|
||||
row.value = payload
|
||||
else:
|
||||
db.add(Setting(key=_INDEX_SETTING, value=payload,
|
||||
description="cvelistV5 product→CVE reverse index (curated)"))
|
||||
db.commit()
|
||||
|
||||
|
||||
def load_index(db: Session, allow_stale: bool = True) -> Optional[dict]:
|
||||
from app.models.setting import Setting
|
||||
row = db.query(Setting).filter(Setting.key == _INDEX_SETTING).first()
|
||||
if not row or not row.value:
|
||||
return None
|
||||
try:
|
||||
blob = json.loads(row.value)
|
||||
built = datetime.fromisoformat(blob.get("built_at"))
|
||||
except Exception:
|
||||
return None
|
||||
if not allow_stale and datetime.now() - built > _INDEX_TTL:
|
||||
return None
|
||||
return blob.get("index") or {}
|
||||
|
||||
|
||||
# ---------- scan ----------
|
||||
def scan_asset(db: Session, asset, packages: list, index: dict,
|
||||
new_ids: Optional[list] = None) -> int:
|
||||
"""Match an asset's installed software against the cvelistV5 index.
|
||||
Returns findings upserted. Caller commits."""
|
||||
if not index:
|
||||
return 0
|
||||
if new_ids is None:
|
||||
new_ids = []
|
||||
count = 0
|
||||
seen: set = set()
|
||||
for pkg in packages or []:
|
||||
name = (pkg.get("name") or "").strip()
|
||||
version = (pkg.get("version") or "").strip()
|
||||
if not name or not version:
|
||||
continue
|
||||
key = resolve(name)
|
||||
if not key or key not in index:
|
||||
continue
|
||||
cver = cpe._clean_version(version)
|
||||
if not cver:
|
||||
continue
|
||||
dedup = (key, cver)
|
||||
if dedup in seen:
|
||||
continue
|
||||
seen.add(dedup)
|
||||
for entry in index[key]:
|
||||
if not _affected(cver, entry.get("start"), entry.get("lt"), entry.get("lte")):
|
||||
continue
|
||||
c = {"cve": entry["cve"], "cvss": None, "severity": None,
|
||||
"fixed": entry.get("lt") or entry.get("lte")}
|
||||
try:
|
||||
before = len(new_ids)
|
||||
cpe._upsert(db, asset, name, version, c, new_ids)
|
||||
count += 1 if len(new_ids) > before else 0
|
||||
except Exception as e:
|
||||
logger.debug("cvelistv5 upsert failed (%s on %s): %s", entry["cve"], asset.id, e)
|
||||
return count
|
||||
Reference in New Issue
Block a user