vulncheck
c4abbaa1c1
Identifying a Windows release by its build LINE is not sound, so the whole premise of5703dfafails. Windows 11 24H2 and Windows Server 2025 both sit on 10.0.26100 but keep separate revision sequences: CVE-2026-41089 affects Windows Server 2025 ONLY, fix 10.0.26100.32860 a fully-patched 24H2 client reports 10.0.26100.8655 -> 8655 < 32860, so the client matched a Server-only CVE The same-line floor/fix guard frome19c8d9was necessary but not sufficient: it only rejects the old generic-floor records, not two releases sharing a line. Disabling rather than attempting a third in-place fix: the FPs are in the tester's data now, and I have already been wrong three times here by generalising from too few records. Windows products are no longer indexed at all, so the findings self-heal via the app-scan auto-resolve on the next run. Package scanning is untouched. Re-enabling needs the entry's product NAME in the index, matched against the asset's OS family, verified against CVE-2026-41089 (client must not match) and CVE-2026-47291 (each must match its own line). Written up in scan_asset_os.
591 lines
26 KiB
Python
591 lines
26 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_v4" # v4: pattern products (Windows OS)
|
|
_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|com\.android\.chrome",
|
|
"pairs": [("google", "chrome")]},
|
|
{"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]
|
|
|
|
# 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, so an exact pair list would rot with every new one.
|
|
#
|
|
# DISABLED for Windows (see scan_asset_os): matching the family and letting the
|
|
# build ranges pick the release does NOT work. Windows 11 24H2 and Windows
|
|
# Server 2025 share build line 26100 but keep SEPARATE revision sequences
|
|
# (24H2 fix .8655, Server 2025 fix .32860), so a patched 24H2 host at
|
|
# 26100.8655 fell inside the Server 2025 range and got flagged with a
|
|
# Server-only CVE. Re-enabling needs the entry's PRODUCT NAME kept in the index
|
|
# and matched against the asset's OS family — not the build line alone.
|
|
_PRODUCT_PATTERNS: List[dict] = []
|
|
_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):
|
|
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 = []
|
|
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 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))
|
|
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 [])]
|
|
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,
|
|
"plats": plats, "cvss": cvss, "sev": sev})
|
|
_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 from the asset's build. CURRENTLY UNREACHABLE — the
|
|
caller is disabled and _PRODUCT_PATTERNS is empty, so the index holds no
|
|
'windows' key.
|
|
|
|
Why it is off: identifying a release by its build LINE is not sound.
|
|
Windows 11 24H2 and Windows Server 2025 both live on 10.0.26100 but keep
|
|
separate revision sequences, so CVE-2026-41089 (Server 2025 only, fix
|
|
.32860) matched a fully-patched 24H2 client at .8655. The floor/fix
|
|
same-line guard below is necessary but not sufficient.
|
|
|
|
To re-enable: keep the affected entry's PRODUCT NAME in the index and
|
|
require it to match the asset's OS family (a "Windows 11 …" host may only
|
|
match "Windows 11 …" products, a "Windows Server …" host only
|
|
"Windows Server …"), on top of the same-line guard. Verify against
|
|
CVE-2026-41089 (client must NOT match) and CVE-2026-47291 (client and
|
|
server must each match their own line) before turning it back on.
|
|
"""
|
|
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
|
|
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 _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
|
|
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
|
|
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, version, c, new_ids, touched=touched)
|
|
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")
|