Tester: CVE-2026-16395 flagged on 'Mozilla Firefox 52.2.1 ESR' — the cvelistV5 record carries only 'unaffected 153, lte *' (release train), and per mfsa2026-69/-70 no ESR branch is affected at all. Both matchers (cvelistV5 registry + curated CPE registry) resolved the ESR package to the plain firefox key, so ESR builds were judged against release-train ranges — structurally wrong (NVD tracks ESR as its own firefox_esr CPE; Mozilla ships separate ESR advisories). Guard both resolve() paths: a package whose name carries the ESR word never resolves to the firefox key. ESR patch state would need the MFSA fixed_in data (already stored in the mozilla_advisory index) — until that's wired, no match beats a false positive. Existing FP rows auto-close on the next app-scan via the stale reconcile.
710 lines
33 KiB
Python
710 lines
33 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/truevuln-cvelistv5-cache.zip"
|
|
_ZIP_URL = "https://github.com/CVEProject/cvelistV5/archive/refs/heads/main.zip"
|
|
_ZIP_TTL = 12 * 3600
|
|
_INDEX_SETTING = "cvelistv5_product_index_v7" # v7: modern .NET per-release keys
|
|
_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")]},
|
|
# Require the vendor word: match "Mozilla Firefox" (and "Mozilla Firefox
|
|
# ESR"), never a bare "Firefox" — a stray "…Firefox…" in some other
|
|
# product's name must not resolve here (tester: "nur 'Mozilla' UND
|
|
# 'Firefox', nicht 'Firefox' alleine"). Windows ARP / Wazuh always carry the
|
|
# "Mozilla" prefix, so this loses no real install.
|
|
{"key": "firefox", "re": r"mozilla firefox",
|
|
"pairs": [("mozilla", "firefox")]},
|
|
{"key": "chrome", "re": r"google chrome|com\.android\.chrome",
|
|
"pairs": [("google", "chrome")]},
|
|
# SharePoint — ONE KEY PER RELEASE, deliberately. Every MS SharePoint range
|
|
# uses a generic 16.0.0 floor (checked across 82 recent CVEs), and 2016,
|
|
# 2019 and Subscription Edition all report 16.0.x, so a single shared key
|
|
# would let a 2016 install (16.0.5456) fall inside the 2019 range
|
|
# (16.0.0 .. 16.0.10417.20153) — the exact cross-release false positive the
|
|
# Windows OS scan just had. The release comes from the NAME; the range then
|
|
# only ever decides "patched or not" within that one release.
|
|
# 2013 IS here: it is EOL (2023-04-11) and absent from recent MSRC docs, but
|
|
# the CVE records from its supported years carry real fix builds
|
|
# (CVE-2023-23395: 15.0.0 .. 15.0.5537.1000), and an unpatched 2013 farm is
|
|
# behind all of them. Checking only recent MSRC docs is what made this look
|
|
# undetectable — the tester's Nessus finds these, and so should we.
|
|
{"key": "sharepoint-2013", "re": r"sharepoint.*\b2013\b",
|
|
"pairs": [("microsoft", "microsoft sharepoint foundation 2013 service pack 1"),
|
|
("microsoft", "microsoft sharepoint enterprise server 2013 service pack 1"),
|
|
("microsoft", "microsoft sharepoint server 2013 service pack 1"),
|
|
("microsoft", "microsoft sharepoint foundation 2013"),
|
|
("microsoft", "microsoft sharepoint enterprise server 2013"),
|
|
("microsoft", "microsoft sharepoint server 2013")]},
|
|
{"key": "sharepoint-se", "re": r"sharepoint.*subscription",
|
|
"pairs": [("microsoft", "microsoft sharepoint server subscription edition")]},
|
|
{"key": "sharepoint-2019", "re": r"sharepoint.*\b2019\b",
|
|
"pairs": [("microsoft", "microsoft sharepoint server 2019")]},
|
|
{"key": "sharepoint-2016", "re": r"sharepoint.*\b2016\b",
|
|
"pairs": [("microsoft", "microsoft sharepoint enterprise server 2016"),
|
|
("microsoft", "microsoft sharepoint server 2016")]},
|
|
# Modern .NET (8/9/10) — one key PER RELEASE (same reasoning as SharePoint:
|
|
# the record floors are generic release floors, so a shared key would
|
|
# cross-match releases). The real semantic version lives in the NAME
|
|
# ("Microsoft .NET Runtime - 8.0.16 (x64)"), which bumps with every monthly
|
|
# patch — the version FIELD is an MSI build → name_ver. SDKs are excluded:
|
|
# their numbering (8.0.1xx) never falls inside the runtime fix range and
|
|
# the runtime is installed alongside anyway.
|
|
# .NET FRAMEWORK is deliberately NOT here: its ARP version (4.8.04084) is
|
|
# STATIC across monthly patches (only file versions change), so comparing
|
|
# it against fix builds like 4.8.4803.0 would flag every install forever.
|
|
# Framework patch state is file/KB-based — Defender TVM covers it.
|
|
{"key": "dotnet-10", "name_ver": True,
|
|
"re": r"(?:microsoft \.net (?:desktop )?(?:runtime|host)|windows desktop runtime).*?\b10\.0\.",
|
|
"pairs": [("microsoft", ".net 10.0")]},
|
|
{"key": "dotnet-9", "name_ver": True,
|
|
"re": r"(?:microsoft \.net (?:desktop )?(?:runtime|host)|windows desktop runtime).*?\b9\.0\.",
|
|
"pairs": [("microsoft", ".net 9.0")]},
|
|
{"key": "dotnet-8", "name_ver": True,
|
|
"re": r"(?:microsoft \.net (?:desktop )?(?:runtime|host)|windows desktop runtime).*?\b8\.0\.",
|
|
"pairs": [("microsoft", ".net 8.0")]},
|
|
{"key": "edge", "re": r"microsoft edge(?!.*webview)",
|
|
"pairs": [("microsoft", "microsoft edge (chromium-based)"),
|
|
("microsoft", "edge (chromium-based)"), ("microsoft", "microsoft edge")]},
|
|
{"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")]},
|
|
# Teams itself only — not the Office add-in / VDI / Citrix plugin. And
|
|
# only the Windows/desktop/generic product entries: the "for Mac" / mobile
|
|
# products are dropped so a Mac-only Teams CVE can't land on a Windows host
|
|
# (the per-scan platform filter is the general backstop, but not every CNA
|
|
# fills the platforms field, so scoping the pairs is belt-and-suspenders).
|
|
{"key": "teams", "re": r"microsoft teams(?!.*(machine-wide|add-in|plugin|vdi|citrix))",
|
|
"pairs": [("microsoft", "microsoft teams for desktop"),
|
|
("microsoft", "microsoft teams for windows"),
|
|
("microsoft", "microsoft teams"), ("microsoft", "teams")]},
|
|
]
|
|
|
|
_COMPILED = [(re.compile(e["re"], re.I), e) for e in _REGISTRY]
|
|
_KEY_ENTRY = {e["key"]: 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"]
|
|
|
|
# Pattern products — one curated key for a whole family whose CVE records name
|
|
# a product per RELEASE ("Windows 11 Version 23H2", "Windows Server 2025
|
|
# (Server Core installation)", ...), so an exact pair list would rot with every
|
|
# new release. The entry keeps its product NAME so the scan can tell the
|
|
# releases apart; the build line alone cannot (see _WIN_FAMILIES).
|
|
_PRODUCT_PATTERNS: List[dict] = [
|
|
{"key": "windows", "vendor_re": r"^microsoft$",
|
|
"product_re": r"^windows\s+(10|11|server)\b"},
|
|
]
|
|
|
|
# Asset OS string → which cvelistV5 product names it may match.
|
|
#
|
|
# This is the part that was missing and caused the cross-release FPs: Windows 11
|
|
# 24H2 and Windows Server 2025 both live on build line 10.0.26100 but keep
|
|
# SEPARATE revision sequences (24H2 fix .8655, Server 2025 fix .32860), so
|
|
# CVE-2026-41089 — Server 2025 only — matched a fully-patched 24H2 client.
|
|
# Deciding the family from the OS string first makes the build range a
|
|
# within-release "patched or not" test, which is all it can honestly answer.
|
|
# Order matters: "Windows Server" must be tested before the bare client names.
|
|
_WIN_FAMILIES: List[tuple] = [
|
|
(re.compile(r"windows\s+server", re.I), re.compile(r"^windows\s+server\b", re.I)),
|
|
(re.compile(r"windows\s*11", re.I), re.compile(r"^windows\s+11\b", re.I)),
|
|
(re.compile(r"windows\s*10", re.I), re.compile(r"^windows\s+10\b", re.I)),
|
|
]
|
|
|
|
|
|
def _win_family(os_name: str) -> Optional[re.Pattern]:
|
|
"""Asset OS string → regex the entry's product name must satisfy."""
|
|
n = (os_name or "").strip().lower()
|
|
for os_rx, prod_rx in _WIN_FAMILIES:
|
|
if os_rx.search(n):
|
|
return prod_rx
|
|
return None
|
|
_PATTERNS_COMPILED = [
|
|
(re.compile(p["vendor_re"], re.I), re.compile(p["product_re"], re.I), p["key"])
|
|
for p in _PRODUCT_PATTERNS
|
|
]
|
|
|
|
|
|
def _pair_key(vendor: str, product: str) -> Optional[str]:
|
|
"""(vendor, product) from a CVE record → curated product key."""
|
|
v, p = (vendor or "").strip().lower(), (product or "").strip().lower()
|
|
key = _PAIR_TO_KEY.get((v, p))
|
|
if key:
|
|
return key
|
|
for vrx, prx, k in _PATTERNS_COMPILED:
|
|
if vrx.search(v) and prx.search(p):
|
|
return k
|
|
return None
|
|
|
|
|
|
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):
|
|
if e["key"] == "firefox" and re.search(r"\besr\b", n):
|
|
# ESR install ("Mozilla Firefox 52.2.1 ESR"): the cvelistV5
|
|
# Firefox ranges describe the RELEASE train (fresh records list
|
|
# only "unaffected 153+"), so matching an ESR build against
|
|
# them flags CVEs whose MFSA advisory doesn't touch ESR at all
|
|
# (tester: CVE-2026-16395 on Firefox 52 ESR; mfsa2026-70 lists
|
|
# no ESR fix). ESR patch state needs the MFSA fixed_in data —
|
|
# until that's wired, no match beats a false positive.
|
|
return None
|
|
return e["key"]
|
|
return None
|
|
|
|
|
|
# ---------- CVSS base score ----------
|
|
def _cvss_from_record(data: dict) -> Tuple[Optional[float], Optional[str]]:
|
|
"""Pull (baseScore, baseSeverity) from a CVE-5 record. Prefers CVSS 3.1 >
|
|
3.0 > 4.0, and the CNA's own metrics over the CISA-ADP block. Returns
|
|
(None, None) when the record carries no score (many fresh CVEs) — enrichment
|
|
can still backfill later."""
|
|
conts = data.get("containers") or {}
|
|
metric_blocks = [(conts.get("cna") or {}).get("metrics") or []]
|
|
for adp in (conts.get("adp") or []):
|
|
metric_blocks.append(adp.get("metrics") or [])
|
|
for field in ("cvssV3_1", "cvssV3_0", "cvssV4_0"):
|
|
for metrics in metric_blocks:
|
|
for m in metrics:
|
|
cv = m.get(field) if isinstance(m, dict) else None
|
|
if isinstance(cv, dict) and cv.get("baseScore") is not None:
|
|
try:
|
|
return float(cv["baseScore"]), (cv.get("baseSeverity") or "").lower() or None
|
|
except (TypeError, ValueError):
|
|
pass
|
|
return None, 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 = []
|
|
unaffected_floors: List[str] = []
|
|
unaffected_total = 0
|
|
for v in aff.get("versions", []) or []:
|
|
if not isinstance(v, dict):
|
|
continue
|
|
if (v.get("status") or "affected") != "affected":
|
|
# Some CNAs state the INVERSE — Mozilla ships no affected range at
|
|
# all, only "152.0.6 and up are unaffected" (status=unaffected,
|
|
# lessThanOrEqual="*"). Skipping those meant the newest Firefox CVEs
|
|
# never entered the index. "X and up are fixed" == "below X is
|
|
# affected", so remember X as a fix bound.
|
|
if v.get("status") == "unaffected":
|
|
unaffected_total += 1
|
|
if (not v.get("lessThan")
|
|
and (v.get("lessThanOrEqual") in ("*", None))):
|
|
ver = v.get("version")
|
|
if isinstance(ver, str) and ver.strip() not in ("0", "*", "-", ""):
|
|
unaffected_floors.append(ver.strip())
|
|
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 start is not None and (start == lt or start == lte):
|
|
# Some CNAs emit "version" == "lessThan" (e.g. a few Chrome
|
|
# records) — a literal reading gives a zero-width, impossible
|
|
# range. NVD treats these as an open floor (no lower bound); do
|
|
# the same rather than silently dropping the CVE.
|
|
start = None
|
|
if lt or lte:
|
|
out.append((start, lt, lte))
|
|
# Inverse-only records (see above): derive the fix bound from the single
|
|
# "unaffected" floor — but ONLY when the record has EXACTLY ONE unaffected
|
|
# entry. Firefox ESR CVEs list two ("115.38 lte 115.*" AND "140.13 lte *"),
|
|
# and Mozilla writes the ESR floor as an unbounded "lte *", so a "below X"
|
|
# rule wrongly catches regular Firefox (tester: ESR-only CVE-2026-16361
|
|
# flagged on Firefox 121/152). Multiple unaffected entries = multi-train
|
|
# (ESR + release) record → the inverse heuristic can't tell them apart, so
|
|
# skip it rather than risk the false positive. Distinguishing them reliably
|
|
# needs Mozilla's MFSA advisories (per-product), not cvelistV5 alone.
|
|
if not out and unaffected_total == 1 and len(unaffected_floors) == 1:
|
|
out.append((None, unaffected_floors[0], None))
|
|
return out
|
|
|
|
|
|
# Recognised OS names in cvelistV5 affected[].platforms. Microsoft also uses
|
|
# that field for CPU arch ("x64-based Systems", "ARM64-based Systems") — those
|
|
# are NOT OS names, so when a record lists only arch/hardware we can't judge
|
|
# the OS and must keep the finding (never drop on arch alone).
|
|
_OS_PLATFORMS = {
|
|
"windows": {"windows"},
|
|
"macos": {"macos", "mac os", "mac os x", "os x", "mac"},
|
|
"linux": {"linux"},
|
|
"iphone_os": {"ios"},
|
|
"ipados": {"ipados", "ios"},
|
|
"android": {"android"},
|
|
}
|
|
_ALL_OS_TOKENS = {tok for toks in _OS_PLATFORMS.values() for tok in toks}
|
|
|
|
|
|
def _platform_ok(asset_family: Optional[str], platforms) -> bool:
|
|
"""Keep a finding unless the CVE explicitly lists OS platform(s) and the
|
|
asset's OS isn't among them. Records with no platforms, or only
|
|
arch/hardware platforms (x64/ARM/…), are kept (can't judge the OS)."""
|
|
if not asset_family or not platforms:
|
|
return True
|
|
listed = {str(p).lower().strip() for p in platforms}
|
|
os_listed = listed & _ALL_OS_TOKENS
|
|
if not os_listed:
|
|
return True # arch/hardware only → not an OS signal
|
|
return bool(_OS_PLATFORMS.get(asset_family, set()) & os_listed)
|
|
|
|
|
|
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
|
|
cvss, sev = _cvss_from_record(data)
|
|
for aff in affected:
|
|
key = _pair_key(aff.get("vendor") or "", aff.get("product") or "")
|
|
if not key:
|
|
continue
|
|
plats = [str(p).lower().strip() for p in (aff.get("platforms") or [])]
|
|
prod = (aff.get("product") or "").strip()
|
|
for start, lt, lte in _ranges_from_affected(aff):
|
|
sig = (key, cve_id, start, lt, lte, prod.lower())
|
|
if sig in seen:
|
|
continue
|
|
seen.add(sig)
|
|
index.setdefault(key, []).append(
|
|
{"cve": cve_id, "start": start, "lt": lt, "lte": lte,
|
|
"plats": plats, "cvss": cvss, "sev": sev, "prod": prod})
|
|
_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 _build_line(v: Optional[str]) -> Optional[tuple]:
|
|
"""First three segments of a Windows build — the release line
|
|
(10.0.14393.9234 → (10,0,14393) = Server 2016 / Win10 1607)."""
|
|
t = cpe._vtuple(v or "")
|
|
return t[:3] if t and len(t) >= 3 else None
|
|
|
|
|
|
def _release_bounded(start: Optional[str], lt: Optional[str]) -> bool:
|
|
"""True when a range actually pins ONE Windows release, i.e. its floor and
|
|
its fix sit on the same build line (10.0.22631.0 .. 10.0.22631.7219).
|
|
|
|
Modern MS records do this; older ones (≈pre-2022) use a generic floor —
|
|
CVE-2021-26432 says `version 10.0.0, lessThan 10.0.17763.2114` for Server
|
|
2019, and that range swallows EVERY lower build, so a Server 2016 host
|
|
(14393.9234) got flagged with a 17763 fix. Such an entry carries no release
|
|
information at all, so the OS scan must skip it rather than guess: the OS
|
|
string alone can't name the client release, which is the whole reason we
|
|
lean on the bounds.
|
|
"""
|
|
sl, ll = _build_line(start), _build_line(lt)
|
|
return bool(sl and ll and sl == ll)
|
|
|
|
|
|
def scan_asset_os(db: Session, asset, index: dict,
|
|
new_ids: Optional[list] = None, touched: Optional[set] = None) -> int:
|
|
"""Windows OS CVEs straight from the asset's build (asset.os_version).
|
|
|
|
Two guards, both learned the hard way — either one alone is not enough:
|
|
|
|
1. FAMILY (_win_family): the entry's product name must belong to the same
|
|
family as the asset. Windows 11 24H2 and Windows Server 2025 share build
|
|
line 10.0.26100 but keep separate revision sequences, so CVE-2026-41089
|
|
(Server 2025 only, fix .32860) otherwise matches a fully-patched 24H2
|
|
client at .8655.
|
|
2. RELEASE-BOUNDED (_release_bounded): the range's floor and fix must sit on
|
|
one build line. Older records use a generic 10.0.0 floor, which swallows
|
|
every lower build (CVE-2021-26432 put a 17763 fix on a 14393 host).
|
|
|
|
With both, the range only ever answers "patched or not" inside the host's
|
|
own release — which is all it can honestly answer.
|
|
"""
|
|
if not index:
|
|
return 0
|
|
entries = index.get("windows") or []
|
|
if not entries:
|
|
return 0
|
|
if cpe._os_family(asset.operating_system or "") != "windows":
|
|
return 0
|
|
fam_rx = _win_family(asset.operating_system or "")
|
|
if not fam_rx:
|
|
return 0 # Windows flavour we can't place (e.g. bare "Windows") → skip
|
|
cver = cpe._clean_version(asset.os_version or "")
|
|
if not cver:
|
|
return 0
|
|
if new_ids is None:
|
|
new_ids = []
|
|
label = (asset.operating_system or "Microsoft Windows").strip()
|
|
count = 0
|
|
for entry in entries:
|
|
if not fam_rx.search((entry.get("prod") or "").strip()):
|
|
continue # different Windows family → its revisions are unrelated
|
|
if not _release_bounded(entry.get("start"), entry.get("lt")):
|
|
continue # range can't tell releases apart → would cross-match
|
|
if not _affected(cver, entry.get("start"), entry.get("lt"), entry.get("lte")):
|
|
continue
|
|
c = {"cve": entry["cve"], "cvss": entry.get("cvss"), "severity": entry.get("sev"),
|
|
"fixed": entry.get("lt")}
|
|
try:
|
|
before = len(new_ids)
|
|
cpe._upsert(db, asset, label, asset.os_version or cver, c, new_ids, touched=touched)
|
|
count += 1 if len(new_ids) > before else 0
|
|
except Exception as e:
|
|
logger.debug("cvelistv5 OS upsert failed (%s on %s): %s", entry["cve"], asset.id, e)
|
|
return count
|
|
|
|
|
|
def scan_asset(db: Session, asset, packages: list, index: dict,
|
|
new_ids: Optional[list] = None, touched: Optional[set] = 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()
|
|
fam = cpe._os_family(asset.operating_system or "")
|
|
for pkg in packages or []:
|
|
name = (pkg.get("name") or "").strip()
|
|
version = (pkg.get("version") or "").strip()
|
|
if not name or not version:
|
|
continue
|
|
if cpe._is_citrix_shim(pkg):
|
|
continue # published-app registry stub, software not on the box
|
|
key = resolve(name)
|
|
if not key or key not in index:
|
|
continue
|
|
# name_ver products (modern .NET): the semantic version lives in the
|
|
# display NAME; the version field is an MSI build that never matches.
|
|
eff_ver = cpe._effective_version(name, version, _KEY_ENTRY.get(key) or {})
|
|
cver = cpe._clean_version(eff_ver)
|
|
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
|
|
if not _platform_ok(fam, entry.get("plats")):
|
|
continue # CVE is for a different OS platform (e.g. Teams-for-Mac)
|
|
# Only lessThan is a real fix target; lessThanOrEqual means that
|
|
# version is still affected (no published fix) → leave fixed empty.
|
|
c = {"cve": entry["cve"], "cvss": entry.get("cvss"), "severity": entry.get("sev"),
|
|
"fixed": entry.get("lt")}
|
|
try:
|
|
before = len(new_ids)
|
|
cpe._upsert(db, asset, name, eff_ver, c, new_ids, touched=touched,
|
|
vendor=(pkg.get("vendor") or None))
|
|
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
|
|
|
|
|
|
if __name__ == "__main__":
|
|
# ponytail: one self-check for the metric parser (CNA v3.1 preferred, ADP
|
|
# fallback, missing → None). Run: python -m app.services.cvelistv5_scan_service
|
|
rec = {"containers": {"cna": {"metrics": [{"cvssV3_1": {"baseScore": 4.3, "baseSeverity": "MEDIUM"}}]},
|
|
"adp": [{"metrics": [{"cvssV3_1": {"baseScore": 9.8, "baseSeverity": "CRITICAL"}}]}]}}
|
|
assert _cvss_from_record(rec) == (4.3, "medium"), _cvss_from_record(rec) # CNA wins
|
|
assert _cvss_from_record({"containers": {"adp": [{"metrics": [{"cvssV3_0": {"baseScore": 7.5}}]}]}}) == (7.5, None)
|
|
assert _cvss_from_record({"containers": {"cna": {}}}) == (None, None) # no metrics
|
|
print("cvelistv5 _cvss_from_record self-check OK")
|