vulncheck
c1fc992285
Why the tester saw nothing: the manual "App CVE Scan" only LOADED the cached index, it never built it — only the nightly did, and that build likely timed out on the 557 MB download (120 s). So 7-Zip CVE-2026-58052 / Notepad++ CVE-2026-52885 (both in-registry, in-range) were never matched. - Manual scan now builds the index when it's missing (same result as the nightly), then caches it. - ZIP download timeout 120 s → 600 s. - Index build pre-filters on raw bytes (only JSON-parse files mentioning a curated vendor) → ~99% fewer json.loads, build drops from minutes to ~a minute. Verified the two CVEs' vendor/product (7-Zip/7-Zip, notepad-plus-plus) match the registry and the installed versions (26.01 ≤ 26.02, 8.9.5 < 8.9.6.4) fall in range — so they will now be detected once the index exists.
405 lines
16 KiB
Python
405 lines
16 KiB
Python
"""
|
|
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 (~557 MB)…")
|
|
with httpx.Client(timeout=httpx.Timeout(600.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
|
|
parsed = 0
|
|
# Pre-filter on raw bytes: only JSON-parse files that mention a curated
|
|
# vendor. ~99% of the 250k CVEs don't, so this skips that many json.loads
|
|
# (the expensive part) — build drops from minutes to ~a minute.
|
|
vendor_bytes = {v.encode() for (v, _p) in _PAIR_TO_KEY.keys()}
|
|
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:
|
|
raw = zf.read(name)
|
|
except Exception:
|
|
continue
|
|
low = raw.lower()
|
|
if not any(vb in low for vb in vendor_bytes):
|
|
continue
|
|
try:
|
|
data = json.loads(raw)
|
|
except Exception:
|
|
continue
|
|
parsed += 1
|
|
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 (%d files scanned, %d parsed) → %d products, %d ranges",
|
|
scanned, parsed, 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 ----------
|
|
# ---------- false-positive suppression ----------
|
|
_FP_STOP = {"setup", "edition", "en", "english", "x64", "x86", "cu", "gdr",
|
|
"for", "based", "systems", "the", "of", "and", "client", "full",
|
|
"host", "version", "core", "server"}
|
|
|
|
|
|
def _sig_tokens(s: str) -> set:
|
|
"""Significant tokens: drop stopwords, 4-digit years, and bare numbers."""
|
|
out = set()
|
|
for t in re.findall(r"[a-z0-9]+", (s or "").lower()):
|
|
if t in _FP_STOP or re.fullmatch(r"(19|20)\d{2}", t) or re.fullmatch(r"\d+", t):
|
|
continue
|
|
out.add(t)
|
|
return out
|
|
|
|
|
|
def _product_matches(installed_name: str, vendor: str, product: str) -> bool:
|
|
"""True when the CVE's affected product is the same product family as the
|
|
installed package. Requires ≥2 shared significant tokens — so only
|
|
multi-word products (SQL Server, Visual Studio, …) where Wazuh's loose
|
|
CPE match over-reports across editions are ever scoped for suppression;
|
|
single-token apps never match → their findings are left untouched."""
|
|
a = _sig_tokens(f"{vendor} {product}")
|
|
b = _sig_tokens(installed_name)
|
|
return len(a & b) >= 2
|
|
|
|
|
|
def _zip_path_for(cve_id: str) -> Optional[str]:
|
|
m = re.match(r"^CVE-(\d{4})-(\d+)$", cve_id)
|
|
if not m:
|
|
return None
|
|
year, num = m.group(1), m.group(2)
|
|
return f"cvelistV5-main/cves/{year}/{int(num) // 1000}xxx/{cve_id}.json"
|
|
|
|
|
|
def suppress_false_positives(db: Session, asset_id: Optional[int] = None) -> dict:
|
|
"""Auto-flag Wazuh findings whose installed version is provably OUTSIDE
|
|
all cvelistV5 affected ranges for the matched product (e.g. a SQL Server
|
|
2019 / 15.x host carrying a CVE that only affects 16.x/17.x because Wazuh
|
|
matched 'Microsoft SQL Server' too loosely).
|
|
|
|
Conservative — only marks when ALL relevant affected entries have clean
|
|
numeric ranges and the install is outside every one. Sets
|
|
status=false_positive (reversible, audit-logged); never auto-unmarks.
|
|
"""
|
|
from app.models.vulnerability import Vulnerability, VulnerabilityStatus
|
|
stats = {"checked": 0, "suppressed": 0, "errors": []}
|
|
if not _ensure_zip():
|
|
stats["errors"].append("cvelistV5 ZIP unavailable")
|
|
return stats
|
|
|
|
q = (db.query(Vulnerability)
|
|
.filter(Vulnerability.status == VulnerabilityStatus.open,
|
|
Vulnerability.cve_id.like("CVE-%"),
|
|
Vulnerability.sources.contains('"wazuh"'),
|
|
Vulnerability.package_name.isnot(None),
|
|
Vulnerability.package_version.isnot(None)))
|
|
if asset_id is not None:
|
|
q = q.filter(Vulnerability.asset_id == asset_id)
|
|
candidates = q.all()
|
|
if not candidates:
|
|
return stats
|
|
|
|
suppressed_ids: list = []
|
|
with zipfile.ZipFile(_ZIP_PATH) as zf:
|
|
names = set(zf.namelist())
|
|
for v in candidates:
|
|
stats["checked"] += 1
|
|
cver = cpe._clean_version(v.package_version or "")
|
|
if not cver:
|
|
continue
|
|
path = _zip_path_for(v.cve_id)
|
|
if not path or path not in names:
|
|
continue # CVE not in snapshot → can't judge → keep
|
|
try:
|
|
data = json.loads(zf.read(path))
|
|
except Exception:
|
|
continue
|
|
affected = ((data.get("containers") or {}).get("cna") or {}).get("affected") or []
|
|
relevant = [a for a in affected
|
|
if _product_matches(v.package_name, a.get("vendor") or "", a.get("product") or "")]
|
|
if not relevant:
|
|
continue # CVE doesn't clearly name this product → keep
|
|
ranges = []
|
|
uncertain = False
|
|
for a in relevant:
|
|
rs = _ranges_from_affected(a)
|
|
if not rs:
|
|
uncertain = True # an entry we can't bound → don't risk it
|
|
break
|
|
ranges.extend(rs)
|
|
if uncertain or not ranges:
|
|
continue
|
|
if any(_affected(cver, s, lt, lte) for s, lt, lte in ranges):
|
|
continue # installed IS in an affected range → real, keep
|
|
# Outside every clean range → false positive.
|
|
v.status = VulnerabilityStatus.false_positive
|
|
v.notification_suppressed = True
|
|
rng = "; ".join(f"[{s or '0'}, {lt or lte})" for s, lt, lte in ranges)
|
|
v.defer_reason = (f"[auto] installed {v.package_version} is outside all "
|
|
f"cvelistV5 affected ranges for this product ({rng})")[:500]
|
|
suppressed_ids.append(v.id)
|
|
stats["suppressed"] += 1
|
|
|
|
if suppressed_ids:
|
|
db.commit()
|
|
logger.info("cvelistV5 FP-suppression: checked %d, suppressed %d",
|
|
stats["checked"], stats["suppressed"])
|
|
return stats
|
|
|
|
|
|
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
|