vulncheck
7088c552e2
The pagination guard stopped the fetch on a 403 but not the build: execution fell through to _store, which wrote the truncated index and a fresh timestamp, so a partial answer was served as authoritative for the full 24h TTL — the same failure the guard was written against, one level up. Nothing is stored now, and the last complete index is returned. Returning an empty one would be worse than stale: every repo-advisory finding would go untouched, which the reconcile reads as resolved. _stored() is the raw read, load_index() keeps the TTL check on top of it. Two more from the same review: Commit hashes could still reach a fix version. _load_via_nvd harvests the same NVD bounds as the scanners into FixCandidates and had no is_version check, so "before 44bf114d2f49" survived on that path after both scanners rejected it. It imports the scanner's rule rather than carrying a third copy. A repo that answers with no advisories is recorded as such again; one that never answered is not. The earlier `if not payload: continue` collapsed both into "never checked". The non-manager package rule is now NON_MANAGER_PKG_RE, one constant that _upsert and the test share — the test compiled its own copy of the pattern and would have stayed green if the production one changed. Covered by a new check that stubs a 403 and asserts nothing is written.
1781 lines
75 KiB
Python
1781 lines
75 KiB
Python
"""
|
||
Vulnerability Score Override Service
|
||
|
||
Korrigiert fehlerhafte CVSS-Scores und Severity-Werte aus Wazuh mit den
|
||
verifizierten Daten aus Nessus/Vulnrichment JSON.
|
||
|
||
Problem: Wazuh liefert manchmal falsche CVSS-Werte (z.B. 10.0 für CVE-2026-8390)
|
||
während die korrekten Werte (z.B. 7.3 HIGH) aus Nessus/Vulnrichment stammen.
|
||
|
||
Funktionsweise:
|
||
1. Lädt verifizierte CVE-Daten aus Nessus Plugin-Output
|
||
2. Vergleicht mit bestehenden Datenbankeinträgen
|
||
3. Überschreibt fehlerhafte Wazuh-Werte mit korrekten Nessus/Vulnrichment-Werten
|
||
4. Aktualisiert auch Severity und Exploitation-Status (SSVC)
|
||
"""
|
||
import json
|
||
import logging
|
||
import re
|
||
from datetime import datetime
|
||
from typing import Dict, List, Optional, Set, Tuple
|
||
from dataclasses import dataclass, field
|
||
|
||
|
||
# Operator-prefixed version strings seen in CVE-5 `affected[].versions[].version`
|
||
# (Vim CVE-2026-45130 = "< 9.2.0450", filelock CVE-2025-68146 = "< 3.20.1").
|
||
# Only exclusive upper bounds (`< X`) yield a real fix_version — inclusive
|
||
# (`<= X`) means X itself is affected (Ghostscript bug-1 case).
|
||
_VERSION_LT_RE = re.compile(r"^\s*<\s*([0-9][\w\.\-:+~]*)\s*$")
|
||
_VERSION_LE_RE = re.compile(r"^\s*<=\s*([0-9][\w\.\-:+~]*)\s*$")
|
||
|
||
|
||
def _version_major(value: Optional[str]) -> Optional[int]:
|
||
"""Extract the leading integer ("major") from a version string.
|
||
|
||
Tolerates Firefox / Chrome ("150.0.3" → 150), Adobe date.build
|
||
("2017.011.30079" → 17, year-truncated for cross-source match),
|
||
Java-style ("11.0.21" → 11), ESR suffix ("140.0esr" → 140).
|
||
Returns None when no leading int can be parsed.
|
||
|
||
Adobe normalisation rule shared with `_version_tuple` so the
|
||
picker treats `2017.011.x` (cvelistV5) and `17.011.x` (NVD CPE +
|
||
Wazuh) as the same release stream.
|
||
"""
|
||
t = _version_tuple(value)
|
||
return t[0] if t else None
|
||
|
||
|
||
def _normalize_for_display(value: Optional[str]) -> Optional[str]:
|
||
"""Strip the leading 4-digit year on Adobe date.build.patch
|
||
versions when present, so a propagated fixed_version looks like
|
||
`18.011.20055` rather than `2018.011.20055` next to an installed
|
||
string of `17.011.30079`.
|
||
|
||
Pass-through for everything else (Firefox 150.0.3, semver, etc.).
|
||
"""
|
||
if not value or not isinstance(value, str):
|
||
return value
|
||
parts = value.strip().split(".")
|
||
if len(parts) < 3:
|
||
return value
|
||
m = re.match(r"^0*(\d+)$", parts[0])
|
||
if not m:
|
||
return value
|
||
head = int(m.group(1))
|
||
if 1990 <= head <= 2099:
|
||
parts[0] = str(head % 100).zfill(2)
|
||
return ".".join(parts)
|
||
return value
|
||
|
||
|
||
def _version_tuple(value: Optional[str]) -> Tuple[int, ...]:
|
||
"""Numeric tuple form for cheap version comparisons.
|
||
|
||
Splits on `.` and `-`, takes leading digits per segment, ignores
|
||
non-numeric suffixes. "150.0.3" → (150, 0, 3); "140.0esr" → (140, 0);
|
||
invalid → (). Good enough for `>=`/`<=` against same-product
|
||
candidates inside the picker.
|
||
|
||
Adobe-style date.build.patch (`2017.011.30079`) is normalised to
|
||
its short form (`17.011.30079`) when the first segment looks like
|
||
a 4-digit year (1990-2099) AND there are 3+ segments — this is
|
||
Adobe's date convention. cvelistV5 uses full year, NVD CPE uses
|
||
truncated; without normalisation the picker would see them as
|
||
"different streams" and refuse to match.
|
||
"""
|
||
if not value or not isinstance(value, str):
|
||
return ()
|
||
out: list[int] = []
|
||
for seg in re.split(r"[.\-+]", value.strip()):
|
||
m = re.match(r"0*(\d+)", seg)
|
||
if not m:
|
||
break
|
||
try:
|
||
out.append(int(m.group(1)))
|
||
except ValueError:
|
||
break
|
||
if len(out) >= 3 and 1990 <= out[0] <= 2099:
|
||
out[0] = out[0] % 100 # 2017 → 17, 2018 → 18
|
||
return tuple(out)
|
||
|
||
|
||
def pick_fix_for_installed(
|
||
candidates: List["FixCandidate"],
|
||
installed_version: Optional[str],
|
||
) -> Optional[str]:
|
||
"""Choose the best `less_than` from a candidate list for a given install.
|
||
|
||
Strategy:
|
||
1. Drop inclusive-only candidates (no real fix).
|
||
2. Prefer candidates whose `version_start` major matches the
|
||
installed major (same release stream).
|
||
3. Among matches, prefer the smallest `less_than` that is still
|
||
> installed_version (closest patch).
|
||
4. Fallback: candidate with version_start in ("0", "", None)
|
||
(catch-all "everything before X is affected").
|
||
5. Last resort: first exclusive candidate.
|
||
|
||
Returns the `less_than` string or None if no usable candidate.
|
||
"""
|
||
excl = [c for c in candidates if c.less_than]
|
||
if not excl:
|
||
return None
|
||
if not installed_version:
|
||
# No installed version to compare — use first exclusive bound.
|
||
return excl[0].less_than
|
||
|
||
inst_major = _version_major(installed_version)
|
||
inst_tuple = _version_tuple(installed_version)
|
||
|
||
# Bucket by start-major-matches-installed-major
|
||
same_major = []
|
||
catch_all = []
|
||
for c in excl:
|
||
sm = _version_major(c.version_start)
|
||
if c.version_start in (None, "", "0", "-") or sm == 0:
|
||
catch_all.append(c)
|
||
elif inst_major is not None and sm == inst_major:
|
||
same_major.append(c)
|
||
|
||
# 1) prefer same-major candidate where lessThan > installed
|
||
same_major_real = [
|
||
c for c in same_major
|
||
if _version_tuple(c.less_than) > inst_tuple
|
||
]
|
||
if same_major_real:
|
||
same_major_real.sort(key=lambda c: _version_tuple(c.less_than))
|
||
return same_major_real[0].less_than
|
||
|
||
# 2) any same-major
|
||
if same_major:
|
||
same_major.sort(key=lambda c: _version_tuple(c.less_than))
|
||
return same_major[0].less_than
|
||
|
||
# 3) catch-all whose lessThan is > installed (covers "0 → 151.0" Firefox case)
|
||
catch_all_real = [
|
||
c for c in catch_all
|
||
if _version_tuple(c.less_than) > inst_tuple
|
||
]
|
||
if catch_all_real:
|
||
catch_all_real.sort(key=lambda c: _version_tuple(c.less_than))
|
||
return catch_all_real[0].less_than
|
||
|
||
# 4) any exclusive bound — fallback to first
|
||
return excl[0].less_than
|
||
|
||
|
||
def _fix_from_version_string(value: Optional[str]) -> Optional[str]:
|
||
"""Extract a fix-target from a CVE-5 `version` string containing an operator.
|
||
|
||
Returns the version on the exclusive side; returns None for plain
|
||
versions, inclusive operators, ranges, or anything we can't parse safely.
|
||
"""
|
||
if not value or not isinstance(value, str):
|
||
return None
|
||
m = _VERSION_LT_RE.match(value)
|
||
if m:
|
||
return m.group(1)
|
||
# `<=` is inclusive — affected version IS X, no fix announced → return None
|
||
if _VERSION_LE_RE.match(value):
|
||
return None
|
||
return None
|
||
|
||
|
||
# ----------------------------------------------------------------------
|
||
# Canonical CVE-metadata propagation
|
||
# ----------------------------------------------------------------------
|
||
#
|
||
# Fields that are INTRINSIC to a CVE (i.e. independent of which asset
|
||
# the finding lives on). When one vuln row gets updated with a value
|
||
# from an authoritative source, every sibling row sharing the same
|
||
# cve_id should converge — otherwise sort-by-priority / sort-by-CPR
|
||
# shows the SAME CVE with different scores across hosts (tester
|
||
# report: "Same-CVE / divergent-CVSS rows").
|
||
#
|
||
# `fixed_version` is INTENTIONALLY excluded — Plan I (multi-stream
|
||
# picker) writes per-asset/per-package fixes that legitimately
|
||
# differ between hosts running different release streams.
|
||
#
|
||
# `cvss_vector` follows `cvss_score` so a propagated score is
|
||
# explainable.
|
||
_CANONICAL_FIELDS = (
|
||
"cvss_score",
|
||
"cvss_vector",
|
||
"severity",
|
||
"exploitation_status",
|
||
"exploitation_source",
|
||
"ssvc_technical_impact",
|
||
"ssvc_automatable",
|
||
)
|
||
|
||
|
||
def propagate_canonical_to_siblings(
|
||
db,
|
||
source_vuln,
|
||
fields=_CANONICAL_FIELDS,
|
||
) -> int:
|
||
"""Copy canonical CVE-metadata fields to all sibling rows.
|
||
|
||
A "sibling" is any other Vulnerability with the same cve_id.
|
||
Only writes a field when the sibling's value DIFFERS from the
|
||
source — keeps update count low so the audit log stays useful.
|
||
|
||
Returns the number of sibling rows touched.
|
||
"""
|
||
if not source_vuln.cve_id:
|
||
return 0
|
||
try:
|
||
siblings = (
|
||
db.query(Vulnerability)
|
||
.filter(
|
||
Vulnerability.cve_id == source_vuln.cve_id,
|
||
Vulnerability.id != source_vuln.id,
|
||
)
|
||
.all()
|
||
)
|
||
except Exception as e:
|
||
logger.warning("propagate siblings query failed for %s: %s", source_vuln.cve_id, e)
|
||
return 0
|
||
|
||
touched = 0
|
||
for sib in siblings:
|
||
changed = False
|
||
for f in fields:
|
||
src_val = getattr(source_vuln, f, None)
|
||
if src_val is None:
|
||
continue
|
||
if getattr(sib, f, None) != src_val:
|
||
setattr(sib, f, src_val)
|
||
changed = True
|
||
if changed:
|
||
try:
|
||
sib.refresh_scores()
|
||
except Exception:
|
||
pass
|
||
touched += 1
|
||
return touched
|
||
|
||
|
||
def apply_canonical_from_siblings(db, vuln) -> bool:
|
||
"""Copy canonical fields FROM the freshest sibling INTO a new vuln row.
|
||
|
||
Called right after a sync inserts a new Vulnerability so the row
|
||
starts with the same authoritative CVSS / SSVC values its siblings
|
||
already have. Picks the sibling with the most recent
|
||
`exploitation_source` pin (= last override touch); falls back to
|
||
the most recently updated sibling.
|
||
|
||
Returns True if any field was filled.
|
||
"""
|
||
if not vuln.cve_id:
|
||
return False
|
||
try:
|
||
siblings = (
|
||
db.query(Vulnerability)
|
||
.filter(
|
||
Vulnerability.cve_id == vuln.cve_id,
|
||
Vulnerability.id != vuln.id,
|
||
)
|
||
.all()
|
||
)
|
||
except Exception as e:
|
||
logger.warning("apply-from-siblings query failed for %s: %s", vuln.cve_id, e)
|
||
return False
|
||
if not siblings:
|
||
return False
|
||
# Sort: pinned source first (override-touched), then by updated_at desc.
|
||
siblings.sort(
|
||
key=lambda s: (s.exploitation_source is not None, s.updated_at or datetime.min),
|
||
reverse=True,
|
||
)
|
||
src = siblings[0]
|
||
changed = False
|
||
for f in _CANONICAL_FIELDS:
|
||
src_val = getattr(src, f, None)
|
||
if src_val is None:
|
||
continue
|
||
cur = getattr(vuln, f, None)
|
||
# Only fill when target is empty OR target holds the well-known
|
||
# Wazuh placeholder 10.0 score that we routinely override later.
|
||
if cur is None or (f == "cvss_score" and cur == 10.0 and src_val != 10.0):
|
||
setattr(vuln, f, src_val)
|
||
changed = True
|
||
return changed
|
||
|
||
from sqlalchemy.orm import Session
|
||
|
||
from app.models.vulnerability import (
|
||
Vulnerability,
|
||
VulnerabilitySeverity,
|
||
VulnerabilityStatus,
|
||
)
|
||
from app.models.asset import Asset
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
# SSVC Exploitation Status
|
||
class SSVCExploitationStatus(str):
|
||
"""SSVC Exploitation Kategorien (Draft NIST SP 500-299)"""
|
||
NONE = "none" # Keine Aktivitäten bekannt
|
||
POC = "poc" # Proof-of-Concept vorhanden
|
||
ACTIVE = "active" # Aktive Ausnutzung
|
||
WIDESPREAD = "widespread" # Weit verbreitete Ausnutzung
|
||
|
||
|
||
def _merge_cvss_into(
|
||
bucket: Dict[str, "VerifiedCVEData"],
|
||
cve_id: str,
|
||
extra: "VerifiedCVEData",
|
||
) -> None:
|
||
"""
|
||
Merge an additional source's data into an existing verified entry.
|
||
|
||
Used by the cascade: stage 1 may return SSVC-only data, stage 2/3
|
||
then supply the missing CVSS without overwriting any non-null
|
||
field the earlier stage already set. Source label is preserved
|
||
from whichever stage first persisted the row.
|
||
"""
|
||
existing = bucket.get(cve_id)
|
||
if existing is None:
|
||
bucket[cve_id] = extra
|
||
return
|
||
if existing.cvss_score is None and extra.cvss_score is not None:
|
||
existing.cvss_score = extra.cvss_score
|
||
existing.severity = existing.severity or extra.severity
|
||
# only escalate the source label when the previous one had no
|
||
# actionable score — otherwise we'd churn vulnrichment → nvd
|
||
# for no benefit.
|
||
existing.source = extra.source
|
||
if existing.exploitation_status is None and extra.exploitation_status is not None:
|
||
existing.exploitation_status = extra.exploitation_status
|
||
if existing.ssvc_technical_impact is None and extra.ssvc_technical_impact is not None:
|
||
existing.ssvc_technical_impact = extra.ssvc_technical_impact
|
||
if existing.ssvc_automatable is None and extra.ssvc_automatable is not None:
|
||
existing.ssvc_automatable = extra.ssvc_automatable
|
||
if existing.fixed_version is None and extra.fixed_version is not None:
|
||
existing.fixed_version = extra.fixed_version
|
||
# Merge fix_candidates so stage-2 NVD entries augment stage-1
|
||
# Vulnrichment for picker decisions later.
|
||
if extra.fix_candidates and not existing.fix_candidates:
|
||
existing.fix_candidates = list(extra.fix_candidates)
|
||
|
||
|
||
@dataclass
|
||
class FixCandidate:
|
||
"""One entry in CVE-5 `affected[].versions[]` (or NVD cpeMatch[]).
|
||
|
||
Multiple candidates per CVE arise when the same vulnerability
|
||
affects multiple release streams (Firefox ESR 115 + ESR 140 +
|
||
mainline 150; Adobe DC 2017 + DC 2020; etc.). The picker compares
|
||
each candidate's `version_start` major with the installed
|
||
version's major to choose the right `less_than` patch target.
|
||
|
||
`version_start` mirrors CVE-5 `version` field — either an exact
|
||
version that begins the affected range, or the catch-all "0"
|
||
meaning "everything below less_than".
|
||
"""
|
||
version_start: Optional[str] = None # CVE-5 `version`
|
||
less_than: Optional[str] = None # exclusive upper bound — real fix
|
||
less_than_or_equal: Optional[str] = None # inclusive — NOT a fix
|
||
product: Optional[str] = None
|
||
vendor: Optional[str] = None
|
||
|
||
|
||
@dataclass
|
||
class VerifiedCVEData:
|
||
"""Container für verifizierte CVE-Daten aus einer verlässlichen Quelle"""
|
||
cve_id: str
|
||
ssvc_technical_impact: Optional[str] = None # partial | total
|
||
ssvc_automatable: Optional[str] = None # yes | no
|
||
cvss_score: Optional[float] = None
|
||
cvss_vector: Optional[str] = None
|
||
severity: Optional[str] = None # critical, high, medium, low, none
|
||
exploitation_status: Optional[str] = None # none, poc, active, widespread
|
||
epss_score: Optional[float] = None
|
||
exploit_available: Optional[bool] = None
|
||
exploit_maturity: Optional[str] = None # Unproven, PoC, Functional, High
|
||
vpr_score: Optional[float] = None
|
||
description: Optional[str] = None
|
||
references: Optional[List[str]] = None
|
||
# Wazuh's states-vulnerabilities indexer doesn't carry a fix-target
|
||
# version, so cvelistV5 + NVD are our only sources for it. Extracted
|
||
# from CVE-5 affected[].versions[].lessThan or NVD cpeMatch[].versionEndExcluding.
|
||
# `fixed_version` is the catch-all / first exclusive bound used when
|
||
# no per-package picker runs. Multi-stream CVEs populate
|
||
# `fix_candidates` too so the apply path can pick the stream that
|
||
# matches the installed package version.
|
||
fixed_version: Optional[str] = None
|
||
fix_candidates: List[FixCandidate] = field(default_factory=list)
|
||
source: str = "unknown" # nessus, nvd, vulrighment, cisa
|
||
|
||
|
||
class VulnOverrideService:
|
||
"""
|
||
Service zum Abgleich und Korrektur von CVE-Scores.
|
||
|
||
Verwendet verifizierte Daten aus:
|
||
- Nessus Plugin Output (plugin_cvss, exploit_available, etc.)
|
||
- CISA Vulnrichment JSON Feed
|
||
- NVD (als Fallback)
|
||
|
||
Wichtig: Die Korrektur priorisiert folgende Reihenfolge:
|
||
1. Nessus (direkte Scan-Daten, höchste Vertrauenswürdigkeit)
|
||
2. CISA Vulnrichment (offizielle govt. Quelle)
|
||
3. NVD (offizielle Quelle)
|
||
"""
|
||
|
||
def __init__(self, db: Session):
|
||
self.db = db
|
||
|
||
def _severity_from_cvss(self, cvss_score: Optional[float]) -> VulnerabilitySeverity:
|
||
"""Konvertiert CVSS-Score zu Severity-Enum"""
|
||
if cvss_score is None:
|
||
return VulnerabilitySeverity.none
|
||
if cvss_score >= 9.0:
|
||
return VulnerabilitySeverity.critical
|
||
elif cvss_score >= 7.0:
|
||
return VulnerabilitySeverity.high
|
||
elif cvss_score >= 4.0:
|
||
return VulnerabilitySeverity.medium
|
||
elif cvss_score > 0:
|
||
return VulnerabilitySeverity.low
|
||
return VulnerabilitySeverity.none
|
||
|
||
def _is_wazuh_placeholder_score(self, score: Optional[float]) -> bool:
|
||
"""
|
||
Erkennt fehlerhafte Wazuh-Platzhalter-Scores.
|
||
|
||
Wazuh liefert manchmal 10.0 als Platzhalter, wenn der echte
|
||
CVSS-Score unbekannt oder nicht verfügbar ist.
|
||
"""
|
||
if score is None:
|
||
return False
|
||
# 10.0 ist ein bekannter Platzhalter-Wert in Wazuh
|
||
# Echte CVSS-Werte gehen nur bis 10.0, aber 10.0 selbst ist selten
|
||
# Besonders bei plausiblen Scores wie 7.3 ist 10.0 offensichtlich falsch
|
||
return score == 10.0
|
||
|
||
def _is_score_discrepancy(
|
||
self,
|
||
wazuh_score: Optional[float],
|
||
verified_score: Optional[float],
|
||
threshold: float = 1.0
|
||
) -> bool:
|
||
"""
|
||
Prüft ob eine signifikante Diskrepanz zwischen Scores besteht.
|
||
|
||
Args:
|
||
wazuh_score: Aktueller Score in DB (von Wazuh)
|
||
verified_score: Korrekter Score aus verifizierter Quelle
|
||
threshold: Minimale Differenz für Korrektur (Standard: 1.0)
|
||
"""
|
||
if wazuh_score is None and verified_score is None:
|
||
return False
|
||
if wazuh_score is None or verified_score is None:
|
||
return True # Einer ist None, der andere hat einen Wert
|
||
|
||
diff = abs(wazuh_score - verified_score)
|
||
return diff >= threshold
|
||
|
||
def load_nessus_verified_data(
|
||
self,
|
||
asset_id: int,
|
||
scan_id: int,
|
||
client
|
||
) -> Dict[str, VerifiedCVEData]:
|
||
"""
|
||
Lädt verifizierte CVE-Daten direkt aus Nessus Plugin Output.
|
||
|
||
Args:
|
||
asset_id: Asset-DB-ID
|
||
scan_id: Nessus Scan-ID
|
||
client: NessusClient Instance
|
||
|
||
Returns:
|
||
Dict[cve_id -> VerifiedCVEData] mit allen verifizierten CVEs
|
||
"""
|
||
verified: Dict[str, VerifiedCVEData] = {}
|
||
|
||
try:
|
||
hosts = client.get_scan_hosts(scan_id)
|
||
for host in hosts:
|
||
host_id = host.get("host_id")
|
||
if not host_id:
|
||
continue
|
||
|
||
# Asset-Matching prüfen
|
||
asset = self.db.query(Asset).filter(
|
||
Asset.id == asset_id
|
||
).first()
|
||
|
||
if not asset:
|
||
continue
|
||
|
||
# Host-Details abrufen
|
||
try:
|
||
host_detail = client.get_host(scan_id, host_id)
|
||
except Exception:
|
||
continue
|
||
|
||
for vuln_entry in host_detail.get("vulnerabilities") or []:
|
||
plugin_id = vuln_entry.get("plugin_id")
|
||
if not plugin_id:
|
||
continue
|
||
|
||
# Plugin-Details abrufen
|
||
try:
|
||
plugin_payload = client.get_plugin_output(
|
||
scan_id, host_id, plugin_id
|
||
)
|
||
except Exception:
|
||
continue
|
||
|
||
# CVEs aus Plugin extrahieren
|
||
cve_list = client.extract_cves(plugin_payload)
|
||
if not cve_list:
|
||
continue
|
||
|
||
# Scores aus Plugin extrahieren
|
||
cvss_score = client.plugin_cvss(plugin_payload)
|
||
vpr_score = client.plugin_vpr_score(plugin_payload)
|
||
exploit_avail = client.plugin_exploit_available(plugin_payload)
|
||
exploit_mat = client.plugin_exploit_maturity(plugin_payload)
|
||
description = client.plugin_description(plugin_payload)
|
||
see_also = client.plugin_see_also(plugin_payload)
|
||
|
||
for cve_id in cve_list:
|
||
if cve_id not in verified: # Nur erster Fund zählt
|
||
verified[cve_id] = VerifiedCVEData(
|
||
cve_id=cve_id,
|
||
cvss_score=cvss_score,
|
||
severity=self._severity_from_cvss(cvss_score).value,
|
||
exploitation_status=self._map_exploit_maturity(exploit_mat),
|
||
exploit_available=exploit_avail,
|
||
exploit_maturity=exploit_mat,
|
||
vpr_score=vpr_score,
|
||
description=description,
|
||
references=see_also,
|
||
source="nessus"
|
||
)
|
||
|
||
except Exception as e:
|
||
logger.error(f"Fehler beim Laden Nessus-Daten für Asset {asset_id}: {e}")
|
||
|
||
return verified
|
||
|
||
def _map_exploit_maturity(self, maturity: Optional[str]) -> str:
|
||
"""Konvertiert Nessus exploit_code_maturity zu SSVC exploitation status"""
|
||
if not maturity:
|
||
return SSVCExploitationStatus.NONE
|
||
|
||
maturity_lower = maturity.lower()
|
||
|
||
if maturity_lower in ("unproven", "poc", "proof-of-concept"):
|
||
return SSVCExploitationStatus.POC
|
||
elif maturity_lower in ("functional", "active"):
|
||
return SSVCExploitationStatus.ACTIVE
|
||
elif maturity_lower == "high":
|
||
return SSVCExploitationStatus.WIDESPREAD
|
||
|
||
return SSVCExploitationStatus.NONE
|
||
|
||
# Threshold above which we prefer one big ZIP-snapshot download over
|
||
# per-CVE 404-prone GitHub raw fetches. Picked so that small "fix
|
||
# one CVE" calls stay snappy (~1s/CVE per-raw) while large "correct
|
||
# everything" runs no longer hammer GitHub for 5–10 minutes and
|
||
# then time out the browser.
|
||
_ZIP_FALLBACK_THRESHOLD = 25
|
||
_ZIP_URL = "https://github.com/cisagov/vulnrichment/archive/refs/heads/develop.zip"
|
||
|
||
# cvelistV5 — official MITRE/CVE.org JSON 5 cache, ~557 MB.
|
||
# Used as a 3rd-stage fallback after Vulnrichment + NVD because it
|
||
# covers far more CVEs (every published CVE, not just CISA-curated
|
||
# ones). Disk-cached for 12h to avoid hammering GitHub.
|
||
_CVELIST_ZIP_URL = "https://github.com/CVEProject/cvelistV5/archive/refs/heads/main.zip"
|
||
_CVELIST_CACHE_PATH = "/tmp/truevuln-cvelistv5-cache.zip"
|
||
_CVELIST_CACHE_TTL_SECONDS = 12 * 3600
|
||
|
||
def load_cisa_vulnrichment_data(self, cve_ids: List[str]) -> Dict[str, VerifiedCVEData]:
|
||
"""
|
||
Lädt verifizierte CVE-Daten aus dem CISA Vulnrichment Repo.
|
||
|
||
Routing:
|
||
- ≤ 25 CVEs → per-CVE GitHub raw fetch (low overhead for single
|
||
corrections).
|
||
- > 25 CVEs → ZIP-Snapshot download + local walk (drastically
|
||
faster for the "correct everything" admin button — replaces
|
||
previous per-CVE loop which timed out browsers at ~23 min).
|
||
|
||
CVEs, die GitHub mit 404 beantwortet bzw. die im ZIP-Snapshot
|
||
fehlen, tauchen nicht im Dict auf. CISA lagged ~Wochen für
|
||
neue CVEs — keinen Fehler werfen, einfach skip.
|
||
"""
|
||
cve_ids_upper = [c.upper() for c in cve_ids if c]
|
||
if len(cve_ids_upper) > self._ZIP_FALLBACK_THRESHOLD:
|
||
try:
|
||
verified = self._load_via_zip_snapshot(cve_ids_upper)
|
||
except Exception as e:
|
||
logger.warning(
|
||
"vulnrichment ZIP snapshot failed (%s) — falling back "
|
||
"to per-CVE raw fetch", e,
|
||
)
|
||
verified = self._load_via_per_cve_raw(cve_ids_upper)
|
||
else:
|
||
verified = self._load_via_per_cve_raw(cve_ids_upper)
|
||
|
||
# Stage 2 — NVD fallback for small missing sets. NVD API has a
|
||
# ~5 req/30s unauthenticated cap; ok for a handful of CVEs but
|
||
# not for hundreds.
|
||
# A CVE that Vulnrichment returned with SSVC-only (no CVSS)
|
||
# still counts as "missing" for CVSS purposes — we want NVD or
|
||
# cvelistV5 to fill the score while the SSVC fields stay.
|
||
def _needs_cvss(cid: str) -> bool:
|
||
v = verified.get(cid)
|
||
return v is None or v.cvss_score is None
|
||
|
||
# Cap scales with the NVD API key: 50 req/30s authenticated vs 5
|
||
# unauthenticated, so a key makes a much larger batch feasible before
|
||
# cvelistV5 (stage 3) mops up the rest.
|
||
import os as _os
|
||
nvd_cap = 1500 if _os.getenv("NVD_API_KEY", "").strip() else 100
|
||
missing = [c for c in cve_ids_upper if _needs_cvss(c)]
|
||
if missing and len(missing) <= nvd_cap:
|
||
try:
|
||
nvd_data = self._load_via_nvd(missing)
|
||
for cve_id, data in nvd_data.items():
|
||
_merge_cvss_into(verified, cve_id, data)
|
||
logger.info(
|
||
"stage 2 (NVD): %d/%d missing CVEs filled",
|
||
len(nvd_data), len(missing),
|
||
)
|
||
except Exception as e:
|
||
logger.warning("NVD fallback failed: %s", e)
|
||
elif missing:
|
||
logger.info(
|
||
"stage 2 (NVD): skipped — %d missing CVEs exceeds the %d-cap "
|
||
"to avoid rate-limit; falling through to cvelistV5",
|
||
len(missing), nvd_cap,
|
||
)
|
||
|
||
# Stage 3 — cvelistV5 ZIP snapshot from CVE.org. Used when many
|
||
# CVEs are still missing after NVD (too many for the API), or
|
||
# always as a backstop when Vulnrichment couldn't see them.
|
||
# Same CVE-5 JSON shape as Vulnrichment so the parser is reused.
|
||
missing = [c for c in cve_ids_upper if _needs_cvss(c)]
|
||
if missing:
|
||
try:
|
||
cvelist_data = self._load_via_cvelistv5_zip(missing)
|
||
for cve_id, data in cvelist_data.items():
|
||
_merge_cvss_into(verified, cve_id, data)
|
||
logger.info(
|
||
"stage 3 (cvelistV5): %d/%d missing CVEs filled",
|
||
len(cvelist_data), len(missing),
|
||
)
|
||
except Exception as e:
|
||
logger.warning("cvelistV5 fallback failed: %s", e)
|
||
|
||
# Stage 4 — GitHub Security Advisories. Backstop for CVEs still missing
|
||
# a score after NVD + cvelistV5, i.e. very fresh CVEs GHSA has but the
|
||
# others don't yet (the gap the tester hit). Self-throttles on the
|
||
# GitHub rate limit; a github_pat setting lifts it to 5000 req/h.
|
||
missing = [c for c in cve_ids_upper if _needs_cvss(c)]
|
||
if missing:
|
||
try:
|
||
ghsa_data = self._load_via_ghsa(missing)
|
||
for cve_id, data in ghsa_data.items():
|
||
_merge_cvss_into(verified, cve_id, data)
|
||
logger.info(
|
||
"stage 4 (GHSA): %d/%d missing CVEs filled",
|
||
len(ghsa_data), len(missing),
|
||
)
|
||
except Exception as e:
|
||
logger.warning("GHSA fallback failed: %s", e)
|
||
return verified
|
||
|
||
def _load_via_per_cve_raw(self, cve_ids: List[str]) -> Dict[str, VerifiedCVEData]:
|
||
"""Per-CVE GitHub raw fetch. Pfadschema:
|
||
``/{year}/{bucket}xxx/{CVE-ID}.json``
|
||
wobei ``bucket`` = ``int(cve_number) // 1000`` (z.B. CVE-2026-8390
|
||
→ ``2026/8xxx/CVE-2026-8390.json``).
|
||
"""
|
||
import httpx
|
||
import re
|
||
|
||
verified: Dict[str, VerifiedCVEData] = {}
|
||
cve_pattern = re.compile(r"^CVE-(\d{4})-(\d+)$")
|
||
base = "https://raw.githubusercontent.com/cisagov/vulnrichment/develop"
|
||
|
||
def _url_for(cve_id: str) -> Optional[str]:
|
||
m = cve_pattern.match(cve_id.upper())
|
||
if not m:
|
||
return None
|
||
year, num = m.group(1), m.group(2)
|
||
bucket = f"{int(num) // 1000}xxx"
|
||
return f"{base}/{year}/{bucket}/{cve_id.upper()}.json"
|
||
|
||
try:
|
||
with httpx.Client(timeout=10.0, follow_redirects=True) as client:
|
||
for cve_id in cve_ids:
|
||
url = _url_for(cve_id)
|
||
if not url:
|
||
continue
|
||
try:
|
||
r = client.get(url)
|
||
if r.status_code == 404:
|
||
continue
|
||
r.raise_for_status()
|
||
parsed = self._parse_vulnrichment_record(cve_id, r.json())
|
||
if parsed:
|
||
verified[cve_id] = parsed
|
||
except Exception as e:
|
||
logger.debug("vulnrichment raw fetch failed for %s: %s", cve_id, e)
|
||
except Exception as e:
|
||
logger.error("vulnrichment client error: %s", e)
|
||
return verified
|
||
|
||
def _load_via_ghsa(self, cve_ids: List[str]) -> Dict[str, VerifiedCVEData]:
|
||
"""GitHub Security Advisories — last-resort CVSS/severity/description.
|
||
|
||
GHSA mirrors CVEs that can still be missing from NVD and cvelistV5 when
|
||
very fresh (the gap the tester hit on new Firefox/Notepad++ CVEs). The
|
||
global-advisory API returns cvss + severity + description keyed by CVE.
|
||
Optional PAT (setting `github_pat`) lifts the rate limit 60 → 5000/h;
|
||
the loop stops cleanly when the limit is hit.
|
||
|
||
Only CVSS/severity/description are filled — GHSA 'unreviewed' advisories
|
||
(desktop-app CVEs like Notepad++) carry NO affected-version range, so no
|
||
fix/version data can be derived here (verified against the live API)."""
|
||
import httpx
|
||
from app.auth.setting_crypto import read_setting_value
|
||
|
||
token = None
|
||
try:
|
||
token = (read_setting_value(self.db, "github_pat") or "").strip() or None
|
||
except Exception:
|
||
token = None
|
||
|
||
headers = {"Accept": "application/vnd.github+json",
|
||
"X-GitHub-Api-Version": "2022-11-28"}
|
||
if token:
|
||
headers["Authorization"] = f"Bearer {token}"
|
||
|
||
verified: Dict[str, VerifiedCVEData] = {}
|
||
with httpx.Client(timeout=15.0, follow_redirects=True, headers=headers) as client:
|
||
for cve_id in cve_ids:
|
||
try:
|
||
r = client.get("https://api.github.com/advisories",
|
||
params={"cve_id": cve_id})
|
||
if r.status_code == 403 and r.headers.get("x-ratelimit-remaining") == "0":
|
||
logger.warning(
|
||
"GHSA: rate limit hit — stopping (set the github_pat "
|
||
"setting for 5000 req/h)")
|
||
break
|
||
if r.status_code != 200:
|
||
continue
|
||
arr = r.json() or []
|
||
if not arr:
|
||
continue
|
||
adv = arr[0]
|
||
cvss = adv.get("cvss") or {}
|
||
score = cvss.get("score")
|
||
if not score: # cvss.score can be 0/None → try structured block
|
||
sev_block = adv.get("cvss_severities") or {}
|
||
for k in ("cvss_v4", "cvss_v3"):
|
||
s = (sev_block.get(k) or {}).get("score")
|
||
if s:
|
||
score, cvss = s, sev_block[k]
|
||
break
|
||
if not score:
|
||
continue
|
||
score = float(score)
|
||
sev = (adv.get("severity") or "").lower() or \
|
||
self._severity_from_cvss(score).value
|
||
verified[cve_id] = VerifiedCVEData(
|
||
cve_id=cve_id,
|
||
cvss_score=score,
|
||
cvss_vector=cvss.get("vector_string") or None,
|
||
severity=sev,
|
||
description=(adv.get("description") or adv.get("summary") or None),
|
||
references=adv.get("references") or None,
|
||
source="ghsa",
|
||
)
|
||
except Exception as e:
|
||
logger.debug("GHSA fetch failed for %s: %s", cve_id, e)
|
||
return verified
|
||
|
||
def _load_via_nvd(self, cve_ids: List[str]) -> Dict[str, VerifiedCVEData]:
|
||
"""
|
||
Pull CVSSv3 from the public NVD REST API as a Vulnrichment
|
||
fallback. Endpoint: https://services.nvd.nist.gov/rest/json/cves/2.0
|
||
|
||
Unauthenticated cap is ~5 requests / 30 seconds — fine for the
|
||
handful of CVEs that miss Vulnrichment per correction run. For
|
||
higher throughput an NVD API key could be added later via the
|
||
nvd_config setting.
|
||
|
||
Returns the same VerifiedCVEData shape as the Vulnrichment
|
||
loaders so apply_overrides treats both sources identically.
|
||
"""
|
||
import httpx
|
||
import os
|
||
import time
|
||
|
||
# NVD API key (env NVD_API_KEY, same as the enrichment service) lifts
|
||
# the limit from 5 to 50 req/30s and lets us throttle far less.
|
||
api_key = os.getenv("NVD_API_KEY", "").strip()
|
||
headers = {"apiKey": api_key} if api_key else {}
|
||
batch = 45 if api_key else 5 # requests before a 1s pause
|
||
verified: Dict[str, VerifiedCVEData] = {}
|
||
with httpx.Client(timeout=15.0, follow_redirects=True, headers=headers) as client:
|
||
for idx, cve_id in enumerate(cve_ids):
|
||
if idx and idx % batch == 0:
|
||
time.sleep(1.0) # crude throttle
|
||
try:
|
||
r = client.get(
|
||
"https://services.nvd.nist.gov/rest/json/cves/2.0",
|
||
params={"cveId": cve_id},
|
||
)
|
||
if r.status_code != 200:
|
||
continue
|
||
data = r.json()
|
||
items = data.get("vulnerabilities") or []
|
||
if not items:
|
||
continue
|
||
cve = items[0].get("cve", {})
|
||
metrics = cve.get("metrics", {})
|
||
# Try CVSSv3.1 first, then v3.0
|
||
cvss_score: Optional[float] = None
|
||
severity_label: Optional[str] = None
|
||
for key in ("cvssMetricV31", "cvssMetricV30"):
|
||
entries = metrics.get(key) or []
|
||
if entries:
|
||
data_part = entries[0].get("cvssData", {})
|
||
cvss_score = data_part.get("baseScore")
|
||
severity_label = (
|
||
data_part.get("baseSeverity") or ""
|
||
).lower() or None
|
||
break
|
||
# Fixed-version from NVD CPE matches — collect all
|
||
# candidates for multi-stream pickers (Firefox ESR
|
||
# branches each get their own cpeMatch entry).
|
||
# versionEndExcluding: X → fix at X
|
||
# versionEndIncluding: X → X ITSELF affected, no fix
|
||
# A commit hash bounds a source tree, not a release. The
|
||
# scanners reject those; this path harvests the same NVD
|
||
# bounds, so without the check a hash could still be
|
||
# written out as the version to upgrade to.
|
||
from app.services.app_cve_scanner_service import is_version
|
||
fix_candidates: List[FixCandidate] = []
|
||
for cfg in cve.get("configurations", []) or []:
|
||
for node in cfg.get("nodes", []) or []:
|
||
for cm in node.get("cpeMatch", []) or []:
|
||
lt = cm.get("versionEndExcluding")
|
||
lte = cm.get("versionEndIncluding")
|
||
start = cm.get("versionStartIncluding") \
|
||
or cm.get("versionStartExcluding") \
|
||
or "0"
|
||
if lt in ("*", "-"):
|
||
lt = None
|
||
if lte in ("*", "-"):
|
||
lte = None
|
||
if any(b is not None and not is_version(b)
|
||
for b in (lt, lte, start)):
|
||
continue
|
||
if lt or lte:
|
||
# Extract product from CPE 2.3 string
|
||
# cpe:2.3:a:vendor:product:...
|
||
cpe = cm.get("criteria") or ""
|
||
parts = cpe.split(":")
|
||
vendor = parts[3] if len(parts) > 3 else None
|
||
product = parts[4] if len(parts) > 4 else None
|
||
fix_candidates.append(FixCandidate(
|
||
version_start=str(start) if start else None,
|
||
less_than=str(lt) if lt else None,
|
||
less_than_or_equal=str(lte) if lte else None,
|
||
product=product,
|
||
vendor=vendor,
|
||
))
|
||
fixed_version: Optional[str] = None
|
||
for c in fix_candidates:
|
||
if c.less_than:
|
||
fixed_version = str(c.less_than)[:100]
|
||
break
|
||
if cvss_score is None and not fixed_version:
|
||
continue
|
||
verified[cve_id] = VerifiedCVEData(
|
||
cve_id=cve_id,
|
||
cvss_score=cvss_score,
|
||
severity=severity_label,
|
||
fixed_version=fixed_version,
|
||
fix_candidates=fix_candidates,
|
||
source="nvd",
|
||
)
|
||
except Exception as e:
|
||
logger.debug("NVD fetch failed for %s: %s", cve_id, e)
|
||
return verified
|
||
|
||
def _load_via_cvelistv5_zip(self, cve_ids: List[str]) -> Dict[str, VerifiedCVEData]:
|
||
"""
|
||
Stage 3 fallback — official MITRE/CVE.org cvelistV5 cache.
|
||
|
||
Pulls https://github.com/CVEProject/cvelistV5/archive/refs/heads/main.zip
|
||
once per 12h (disk cache at /tmp/truevuln-cvelistv5-cache.zip),
|
||
then walks the wanted CVE files in-place. Same CVE-5 JSON shape
|
||
as Vulnrichment so _parse_vulnrichment_record handles both.
|
||
|
||
In-zip path schema:
|
||
cvelistV5-main/cves/{year}/{bucket}xxx/{CVE-ID}.json
|
||
|
||
cvelistV5 is the authoritative MITRE feed — every published CVE.
|
||
Where Vulnrichment is opinionated (CISA-curated, ~10k CVEs)
|
||
cvelistV5 is exhaustive (~250k+ CVEs). Downside is size
|
||
(~557 MB). Disk cache amortises the download cost across runs.
|
||
"""
|
||
import httpx
|
||
import os
|
||
import re
|
||
import time
|
||
import zipfile
|
||
|
||
verified: Dict[str, VerifiedCVEData] = {}
|
||
cve_pattern = re.compile(r"^CVE-(\d{4})-(\d+)$")
|
||
wanted: set = set(cve_ids)
|
||
|
||
def _zip_path_for(cve_id: str) -> Optional[str]:
|
||
m = cve_pattern.match(cve_id)
|
||
if not m:
|
||
return None
|
||
year, num = m.group(1), m.group(2)
|
||
bucket = f"{int(num) // 1000}xxx"
|
||
return f"cvelistV5-main/cves/{year}/{bucket}/{cve_id}.json"
|
||
|
||
cache_path = self._CVELIST_CACHE_PATH
|
||
ttl = self._CVELIST_CACHE_TTL_SECONDS
|
||
cache_fresh = (
|
||
os.path.exists(cache_path)
|
||
and (time.time() - os.path.getmtime(cache_path)) < ttl
|
||
and os.path.getsize(cache_path) > 100_000_000 # > 100 MB sanity
|
||
)
|
||
if not cache_fresh:
|
||
logger.info(
|
||
"cvelistV5: downloading fresh ZIP (cache miss / stale) to %s",
|
||
cache_path,
|
||
)
|
||
tmp_path = cache_path + ".part"
|
||
try:
|
||
with httpx.Client(timeout=httpx.Timeout(120.0, connect=15.0),
|
||
follow_redirects=True) as client:
|
||
with client.stream("GET", self._CVELIST_ZIP_URL) as resp:
|
||
resp.raise_for_status()
|
||
with open(tmp_path, "wb") as f:
|
||
for chunk in resp.iter_bytes(chunk_size=1024 * 512):
|
||
f.write(chunk)
|
||
os.replace(tmp_path, cache_path)
|
||
logger.info(
|
||
"cvelistV5: download complete (%.1f MB)",
|
||
os.path.getsize(cache_path) / 1_048_576,
|
||
)
|
||
except Exception:
|
||
# leave any prior cache file alone, drop the partial
|
||
if os.path.exists(tmp_path):
|
||
try:
|
||
os.remove(tmp_path)
|
||
except Exception:
|
||
pass
|
||
raise
|
||
else:
|
||
logger.info(
|
||
"cvelistV5: using cached ZIP (%.1f MB, %.1fh old)",
|
||
os.path.getsize(cache_path) / 1_048_576,
|
||
(time.time() - os.path.getmtime(cache_path)) / 3600,
|
||
)
|
||
|
||
with zipfile.ZipFile(cache_path) as zf:
|
||
names = set(zf.namelist())
|
||
hits = 0
|
||
for cve_id in wanted:
|
||
in_zip = _zip_path_for(cve_id)
|
||
if not in_zip or in_zip not in names:
|
||
continue
|
||
try:
|
||
with zf.open(in_zip) as jf:
|
||
data = json.loads(jf.read().decode("utf-8"))
|
||
parsed = self._parse_vulnrichment_record(cve_id, data)
|
||
if parsed:
|
||
# tag source so the operator + Nessus override-lock
|
||
# can tell where this came from
|
||
parsed.source = "cvelistv5"
|
||
verified[cve_id] = parsed
|
||
hits += 1
|
||
except Exception as e:
|
||
logger.debug("cvelistV5 parse failed for %s: %s", cve_id, e)
|
||
logger.info(
|
||
"cvelistV5: walk found %d/%d requested CVEs",
|
||
hits, len(wanted),
|
||
)
|
||
return verified
|
||
|
||
def load_cve_dates_via_zip(self, cve_ids: List[str]) -> Dict[str, dict]:
|
||
"""Bulk-extract {cve_id: {"published": iso, "last_modified": iso}}
|
||
from the cvelistV5 ZIP snapshot — reuses the SAME 12h disk cache as
|
||
the CVSS-correction cascade (/tmp/truevuln-cvelistv5-cache.zip), so
|
||
when CVSS-correction already pulled the ZIP this is download-free.
|
||
|
||
Used by the enrichment date-backfill when many CVEs are missing
|
||
dates at once (fresh DB) — one 557 MB ZIP + local walk beats
|
||
thousands of per-CVE HTTP round-trips. Dates come from the
|
||
authoritative cveMetadata.datePublished / .dateUpdated.
|
||
"""
|
||
import os
|
||
import re
|
||
import time
|
||
import zipfile
|
||
|
||
out: Dict[str, dict] = {}
|
||
cve_pattern = re.compile(r"^CVE-(\d{4})-(\d+)$")
|
||
wanted = set(cve_ids)
|
||
|
||
def _zip_path_for(cve_id: str) -> Optional[str]:
|
||
m = cve_pattern.match(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"
|
||
|
||
cache_path = self._CVELIST_CACHE_PATH
|
||
cache_fresh = (
|
||
os.path.exists(cache_path)
|
||
and (time.time() - os.path.getmtime(cache_path)) < self._CVELIST_CACHE_TTL_SECONDS
|
||
and os.path.getsize(cache_path) > 100_000_000
|
||
)
|
||
if not cache_fresh:
|
||
# Reuse the cascade's downloader (handles streaming + .part swap).
|
||
self._download_cvelistv5_zip()
|
||
|
||
with zipfile.ZipFile(cache_path) as zf:
|
||
names = set(zf.namelist())
|
||
for cve_id in wanted:
|
||
in_zip = _zip_path_for(cve_id)
|
||
if not in_zip or in_zip not in names:
|
||
continue
|
||
try:
|
||
with zf.open(in_zip) as jf:
|
||
meta = (json.loads(jf.read().decode("utf-8")).get("cveMetadata") or {})
|
||
out[cve_id] = {
|
||
"published": (meta.get("datePublished") or None),
|
||
"last_modified": (meta.get("dateUpdated") or None),
|
||
}
|
||
except Exception as e:
|
||
logger.debug("cvelistV5 date parse failed for %s: %s", cve_id, e)
|
||
logger.info("cvelistV5 dates: %d/%d CVEs found in ZIP", len(out), len(wanted))
|
||
return out
|
||
|
||
def _download_cvelistv5_zip(self) -> None:
|
||
"""Stream the cvelistV5 main.zip to the shared disk cache (12h TTL)."""
|
||
import httpx
|
||
import os
|
||
|
||
cache_path = self._CVELIST_CACHE_PATH
|
||
tmp_path = cache_path + ".part"
|
||
logger.info("cvelistV5: downloading fresh ZIP to %s", cache_path)
|
||
try:
|
||
with httpx.Client(timeout=httpx.Timeout(120.0, connect=15.0),
|
||
follow_redirects=True) as client:
|
||
with client.stream("GET", self._CVELIST_ZIP_URL) as resp:
|
||
resp.raise_for_status()
|
||
with open(tmp_path, "wb") as f:
|
||
for chunk in resp.iter_bytes(chunk_size=1024 * 512):
|
||
f.write(chunk)
|
||
os.replace(tmp_path, cache_path)
|
||
logger.info("cvelistV5: download complete (%.1f MB)",
|
||
os.path.getsize(cache_path) / 1_048_576)
|
||
except Exception:
|
||
if os.path.exists(tmp_path):
|
||
try:
|
||
os.remove(tmp_path)
|
||
except Exception:
|
||
pass
|
||
raise
|
||
|
||
def _load_via_zip_snapshot(self, cve_ids: List[str]) -> Dict[str, VerifiedCVEData]:
|
||
"""Single 249 MB ZIP download → walk locally for the requested CVE
|
||
files. Two orders of magnitude faster than per-CVE 404 lookups
|
||
for full-DB corrections (~2 min vs ~23 min observed).
|
||
|
||
Implementation notes:
|
||
- Stream-download to a temp file so we never hold 249 MB in RAM.
|
||
- Random tempdir per call → no cross-request races.
|
||
- Open ZIP and read only the requested CVE entries by computing
|
||
the in-zip path; no full extraction needed (saves disk + time).
|
||
- tempdir cleaned up via context manager regardless of outcome.
|
||
"""
|
||
import httpx
|
||
import re
|
||
import tempfile
|
||
import zipfile
|
||
import os
|
||
|
||
verified: Dict[str, VerifiedCVEData] = {}
|
||
cve_pattern = re.compile(r"^CVE-(\d{4})-(\d+)$")
|
||
wanted: set = set(cve_ids)
|
||
|
||
def _zip_path_for(cve_id: str) -> Optional[str]:
|
||
m = cve_pattern.match(cve_id)
|
||
if not m:
|
||
return None
|
||
year, num = m.group(1), m.group(2)
|
||
bucket = f"{int(num) // 1000}xxx"
|
||
# GitHub archive top-level dir is "<repo>-<branch>"
|
||
return f"vulnrichment-develop/{year}/{bucket}/{cve_id}.json"
|
||
|
||
with tempfile.TemporaryDirectory(prefix="vulnrichment_") as tmpdir:
|
||
zip_path = os.path.join(tmpdir, "snapshot.zip")
|
||
logger.info(
|
||
"vulnrichment: downloading ZIP snapshot for %d CVEs (target: %s)",
|
||
len(wanted), zip_path,
|
||
)
|
||
with httpx.Client(timeout=httpx.Timeout(60.0, connect=10.0),
|
||
follow_redirects=True) as client:
|
||
with client.stream("GET", self._ZIP_URL) as resp:
|
||
resp.raise_for_status()
|
||
with open(zip_path, "wb") as f:
|
||
for chunk in resp.iter_bytes(chunk_size=1024 * 256):
|
||
f.write(chunk)
|
||
logger.info(
|
||
"vulnrichment: ZIP downloaded (%.1f MB), parsing requested CVEs",
|
||
os.path.getsize(zip_path) / 1_048_576,
|
||
)
|
||
|
||
with zipfile.ZipFile(zip_path) as zf:
|
||
names = set(zf.namelist())
|
||
hits = 0
|
||
for cve_id in wanted:
|
||
in_zip = _zip_path_for(cve_id)
|
||
if not in_zip or in_zip not in names:
|
||
continue
|
||
try:
|
||
with zf.open(in_zip) as jf:
|
||
data = json.loads(jf.read().decode("utf-8"))
|
||
parsed = self._parse_vulnrichment_record(cve_id, data)
|
||
if parsed:
|
||
verified[cve_id] = parsed
|
||
hits += 1
|
||
except Exception as e:
|
||
logger.debug("vulnrichment ZIP parse failed for %s: %s", cve_id, e)
|
||
logger.info(
|
||
"vulnrichment: ZIP walk found %d/%d requested CVEs",
|
||
hits, len(wanted),
|
||
)
|
||
# tempdir is auto-deleted here
|
||
return verified
|
||
|
||
def _parse_vulnrichment_record(
|
||
self, cve_id: str, data: dict
|
||
) -> Optional[VerifiedCVEData]:
|
||
"""Vulnrichment 2.x shape — CVSS can sit in either container:
|
||
|
||
- ``containers.cna.metrics[].cvssV3_1`` — primary score from
|
||
the CVE Numbering Authority (vendor like Microsoft, Mozilla).
|
||
- ``containers.adp[].metrics[].cvssV3_1`` — secondary score
|
||
from an Authorized Data Publisher (CISA).
|
||
|
||
Earlier versions of this parser only checked ADP. Tester
|
||
reported CVE-2026-40416 (MS Edge) not being corrected — root
|
||
cause: Microsoft put CVSS 4.3 in the CNA container, CISA's
|
||
ADP container only carries SSVC, so we never saw the score.
|
||
|
||
Walk CNA first (vendor authoritative), then ADP as fallback.
|
||
SSVC always comes from the ADP container.
|
||
"""
|
||
cvss_score: Optional[float] = None
|
||
severity_label: Optional[str] = None
|
||
exploitation: Optional[str] = None
|
||
technical_impact: Optional[str] = None
|
||
automatable: Optional[str] = None
|
||
|
||
containers = data.get("containers") or {}
|
||
|
||
# 1. CNA — vendor-supplied CVSS (Microsoft, Mozilla, etc.)
|
||
cna_metrics = (containers.get("cna") or {}).get("metrics") or []
|
||
for metric in cna_metrics:
|
||
cvss_v31 = metric.get("cvssV3_1")
|
||
if cvss_v31 and cvss_score is None:
|
||
cvss_score = cvss_v31.get("baseScore")
|
||
severity_label = (cvss_v31.get("baseSeverity") or "").lower() or None
|
||
|
||
# 2. ADP — CISA Vulnrichment (CVSS fallback + SSVC)
|
||
for adp in containers.get("adp") or []:
|
||
for metric in adp.get("metrics", []):
|
||
cvss_v31 = metric.get("cvssV3_1")
|
||
if cvss_v31 and cvss_score is None:
|
||
cvss_score = cvss_v31.get("baseScore")
|
||
severity_label = (cvss_v31.get("baseSeverity") or "").lower() or None
|
||
other = metric.get("other") or {}
|
||
if other.get("type") == "ssvc":
|
||
for opt in (other.get("content") or {}).get("options", []):
|
||
if not isinstance(opt, dict):
|
||
continue
|
||
if "Exploitation" in opt and exploitation is None:
|
||
exploitation = (opt["Exploitation"] or "").lower() or None
|
||
if "Technical Impact" in opt and technical_impact is None:
|
||
technical_impact = (opt["Technical Impact"] or "").lower() or None
|
||
if "Automatable" in opt and automatable is None:
|
||
automatable = (opt["Automatable"] or "").lower() or None
|
||
|
||
# Fixed-version extraction from CVE-5 'affected[].versions[]'.
|
||
# Walk both containers — CNA usually carries the affected ranges.
|
||
#
|
||
# Semantics (per CVE-5 spec):
|
||
# - `lessThan: X` → affected: [version, X) → fix at X
|
||
# - `lessThanOrEqual: X` → affected: [version, X] → X ITSELF
|
||
# is affected,
|
||
# no fix yet
|
||
#
|
||
# Multi-stream CVEs (Firefox ESR 115 + ESR 140 + mainline 150)
|
||
# appear as multiple `affected` entries here. Collect EVERY
|
||
# candidate so the apply path can pick the one whose start-major
|
||
# matches the installed package version (e.g. installed 150.0.3
|
||
# → pick the mainline candidate → fix=151.0).
|
||
fix_candidates: List[FixCandidate] = []
|
||
for container_key in ("cna", "adp"):
|
||
container_value = containers.get(container_key)
|
||
container_list = container_value if isinstance(container_value, list) else [container_value]
|
||
for cont in container_list:
|
||
if not isinstance(cont, dict):
|
||
continue
|
||
for affected in cont.get("affected", []) or []:
|
||
product = affected.get("product")
|
||
vendor = affected.get("vendor")
|
||
for ver in affected.get("versions", []) or []:
|
||
if not isinstance(ver, dict):
|
||
continue
|
||
less_than = ver.get("lessThan")
|
||
less_than_or_equal = ver.get("lessThanOrEqual")
|
||
version_start = ver.get("version")
|
||
# Operator embedded in the `version` string
|
||
# (Vim CVE-2026-45130 = "< 9.2.0450")
|
||
if not less_than and not less_than_or_equal:
|
||
parsed = _fix_from_version_string(version_start)
|
||
if parsed:
|
||
less_than = parsed
|
||
version_start = "0" # was the whole range
|
||
if less_than in ("*", "-"):
|
||
less_than = None
|
||
if less_than_or_equal in ("*", "-"):
|
||
less_than_or_equal = None
|
||
if less_than or less_than_or_equal:
|
||
fix_candidates.append(FixCandidate(
|
||
version_start=version_start,
|
||
less_than=less_than,
|
||
less_than_or_equal=less_than_or_equal,
|
||
product=product,
|
||
vendor=vendor,
|
||
))
|
||
|
||
# Default-pick for the legacy `fixed_version` field — first
|
||
# exclusive bound encountered (back-compat with parsers that
|
||
# don't know about candidates).
|
||
fixed_version: Optional[str] = None
|
||
for c in fix_candidates:
|
||
if c.less_than:
|
||
fixed_version = str(c.less_than)[:100]
|
||
break
|
||
|
||
if (cvss_score is None and not exploitation and not technical_impact
|
||
and not automatable and not fixed_version):
|
||
return None
|
||
|
||
return VerifiedCVEData(
|
||
cve_id=cve_id,
|
||
cvss_score=cvss_score,
|
||
severity=severity_label,
|
||
exploitation_status=exploitation,
|
||
ssvc_technical_impact=technical_impact,
|
||
ssvc_automatable=automatable,
|
||
fixed_version=fixed_version,
|
||
fix_candidates=fix_candidates,
|
||
source="vulnrichment",
|
||
)
|
||
|
||
def apply_overrides(
|
||
self,
|
||
asset_id: int,
|
||
verified_data: Dict[str, VerifiedCVEData],
|
||
dry_run: bool = False
|
||
) -> dict:
|
||
"""
|
||
Wendet Overrides auf Vulnerabilities an.
|
||
|
||
Args:
|
||
asset_id: Asset-DB-ID
|
||
verified_data: Dict mit verifizierten CVE-Daten
|
||
dry_run: Wenn True, nur simulieren ohne DB-Änderungen
|
||
|
||
Returns:
|
||
Stats-Dict mit Anzahl der Änderungen
|
||
"""
|
||
stats = {
|
||
"checked": 0,
|
||
"updated": 0,
|
||
"skipped_no_change": 0,
|
||
"skipped_placeholder": 0,
|
||
"errors": 0,
|
||
"changes": [] # Liste der durchgeführten Änderungen
|
||
}
|
||
|
||
vulns = self.db.query(Vulnerability).filter(
|
||
Vulnerability.asset_id == asset_id,
|
||
Vulnerability.cve_id.in_(list(verified_data.keys()))
|
||
).all()
|
||
|
||
for vuln in vulns:
|
||
stats["checked"] += 1
|
||
cve_id = vuln.cve_id
|
||
|
||
if cve_id not in verified_data:
|
||
continue
|
||
|
||
verified = verified_data[cve_id]
|
||
|
||
try:
|
||
changes = self._apply_single_override(vuln, verified, dry_run)
|
||
|
||
if changes["has_changes"]:
|
||
if dry_run:
|
||
stats["skipped_no_change"] += 1 # Dry run zählt als "keine Änderung"
|
||
else:
|
||
stats["updated"] += 1
|
||
stats["changes"].append({
|
||
"cve_id": cve_id,
|
||
"vuln_id": vuln.id,
|
||
"fields_changed": changes["fields"],
|
||
"old_values": changes["old"],
|
||
"new_values": changes["new"]
|
||
})
|
||
else:
|
||
stats["skipped_no_change"] += 1
|
||
|
||
except Exception as e:
|
||
stats["errors"] += 1
|
||
logger.error(f"Override für {cve_id} fehlgeschlagen: {e}")
|
||
|
||
if not dry_run:
|
||
self.db.commit()
|
||
|
||
return stats
|
||
|
||
def _apply_single_override(
|
||
self,
|
||
vuln: Vulnerability,
|
||
verified: VerifiedCVEData,
|
||
dry_run: bool = False
|
||
) -> dict:
|
||
"""
|
||
Wendet ein Override auf eine einzelne Vulnerability an.
|
||
|
||
Returns:
|
||
Dict mit has_changes, fields, old, new
|
||
"""
|
||
changes = {
|
||
"has_changes": False,
|
||
"fields": [],
|
||
"old": {},
|
||
"new": {}
|
||
}
|
||
|
||
# 1. CVSS-Score prüfen und ggf. korrigieren
|
||
wazuh_score = vuln.cvss_score
|
||
verified_score = verified.cvss_score
|
||
|
||
# Nur korrigieren wenn:
|
||
# - Wazuh einen Platzhalter liefert (10.0)
|
||
# - ODER signifikante Diskrepanz besteht (>1.0 Differenz)
|
||
should_override_score = (
|
||
self._is_wazuh_placeholder_score(wazuh_score) or
|
||
self._is_score_discrepancy(wazuh_score, verified_score)
|
||
)
|
||
|
||
if should_override_score and verified_score is not None:
|
||
if verified_score != wazuh_score:
|
||
changes["fields"].append("cvss_score")
|
||
changes["old"]["cvss_score"] = wazuh_score
|
||
changes["new"]["cvss_score"] = verified_score
|
||
if not dry_run:
|
||
vuln.cvss_score = verified_score
|
||
# Pin the source so subsequent Nessus syncs won't undo
|
||
# this correction (see nessus_sync.run_nessus_sync —
|
||
# the merge path checks exploitation_source == 'vulnrichment'
|
||
# and skips its CVSS/severity override).
|
||
vuln.exploitation_source = verified.source or "vulnrichment"
|
||
changes["has_changes"] = True
|
||
|
||
# 2. Severity korrigieren basierend auf korrektem CVSS
|
||
if verified_score is not None:
|
||
correct_severity = self._severity_from_cvss(verified_score)
|
||
if vuln.severity != correct_severity:
|
||
changes["fields"].append("severity")
|
||
changes["old"]["severity"] = vuln.severity.value
|
||
changes["new"]["severity"] = correct_severity.value
|
||
if not dry_run:
|
||
vuln.severity = correct_severity
|
||
# Same pin as for CVSS — Nessus must not overwrite a
|
||
# Vulnrichment-corrected severity on next sync.
|
||
vuln.exploitation_source = verified.source or "vulnrichment"
|
||
changes["has_changes"] = True
|
||
|
||
# 3. Exploitation-Status (SSVC) aktualisieren
|
||
if verified.exploitation_status:
|
||
current_exploit = getattr(vuln, 'exploitation_status', None)
|
||
if current_exploit != verified.exploitation_status:
|
||
changes["fields"].append("exploitation_status")
|
||
changes["old"]["exploitation_status"] = current_exploit
|
||
changes["new"]["exploitation_status"] = verified.exploitation_status
|
||
if not dry_run:
|
||
vuln.exploitation_status = verified.exploitation_status
|
||
changes["has_changes"] = True
|
||
|
||
# 4. Exploit-Available-Flag aktualisieren
|
||
if verified.exploit_available is not None:
|
||
if vuln.exploit_available != verified.exploit_available:
|
||
changes["fields"].append("exploit_available")
|
||
changes["old"]["exploit_available"] = vuln.exploit_available
|
||
changes["new"]["exploit_available"] = verified.exploit_available
|
||
if not dry_run:
|
||
vuln.exploit_available = verified.exploit_available
|
||
changes["has_changes"] = True
|
||
|
||
# 5. Exploit-Maturity aktualisieren
|
||
if verified.exploit_maturity:
|
||
if vuln.exploit_maturity != verified.exploit_maturity:
|
||
changes["fields"].append("exploit_maturity")
|
||
changes["old"]["exploit_maturity"] = vuln.exploit_maturity
|
||
changes["new"]["exploit_maturity"] = verified.exploit_maturity
|
||
if not dry_run:
|
||
vuln.exploit_maturity = verified.exploit_maturity
|
||
changes["has_changes"] = True
|
||
|
||
# 6. VPR-Score aktualisieren (Tenable)
|
||
if verified.vpr_score is not None:
|
||
if vuln.nessus_vpr_score != verified.vpr_score:
|
||
changes["fields"].append("nessus_vpr_score")
|
||
changes["old"]["nessus_vpr_score"] = vuln.nessus_vpr_score
|
||
changes["new"]["nessus_vpr_score"] = verified.vpr_score
|
||
if not dry_run:
|
||
vuln.nessus_vpr_score = verified.vpr_score
|
||
changes["has_changes"] = True
|
||
|
||
# 7. SSVC Technical Impact (partial | total) — "total" means an
|
||
# attacker can fully control the affected software (CVE-2024-35057
|
||
# style). Stored alongside exploitation_status so the operator can
|
||
# surface "total + automatable + active" CVEs first.
|
||
if verified.ssvc_technical_impact:
|
||
current_ti = getattr(vuln, 'ssvc_technical_impact', None)
|
||
if current_ti != verified.ssvc_technical_impact:
|
||
changes["fields"].append("ssvc_technical_impact")
|
||
changes["old"]["ssvc_technical_impact"] = current_ti
|
||
changes["new"]["ssvc_technical_impact"] = verified.ssvc_technical_impact
|
||
if not dry_run:
|
||
vuln.ssvc_technical_impact = verified.ssvc_technical_impact
|
||
changes["has_changes"] = True
|
||
|
||
# 8. SSVC Automatable (yes | no) — whether reliable mass exploitation
|
||
# is mechanically feasible. Combined with exploitation_status=active
|
||
# this is the strongest "patch right now" signal Vulnrichment gives.
|
||
if verified.ssvc_automatable:
|
||
current_auto = getattr(vuln, 'ssvc_automatable', None)
|
||
if current_auto != verified.ssvc_automatable:
|
||
changes["fields"].append("ssvc_automatable")
|
||
changes["old"]["ssvc_automatable"] = current_auto
|
||
changes["new"]["ssvc_automatable"] = verified.ssvc_automatable
|
||
if not dry_run:
|
||
vuln.ssvc_automatable = verified.ssvc_automatable
|
||
changes["has_changes"] = True
|
||
|
||
# 9. fixed_version — fill from CVE-5 affected[].versions[].lessThan
|
||
# (exclusive upper bound only) or NVD cpeMatch[].versionEndExcluding.
|
||
# Wazuh indexer doesn't carry a fix target so the override service is
|
||
# the only path that populates this column for most rows. Only writes
|
||
# when empty (don't trample a Nessus-supplied version).
|
||
#
|
||
# Multi-stream picker: when fix_candidates is populated (Firefox
|
||
# ESR 115 + 140 + mainline 150 case), call pick_fix_for_installed
|
||
# per child VulnerabilityPackage so PuTTY 0.73 + WinSCP 6.1.2 on
|
||
# the same CVE row get the correct per-stream patch target. The
|
||
# parent `vuln.fixed_version` keeps the catch-all / first-pick
|
||
# for back-compat with rows that have no child packages yet.
|
||
if verified.fixed_version or verified.fix_candidates:
|
||
# Parent column — picker scoped to the parent's package_version
|
||
# if it's set (legacy summary row), otherwise the default pick.
|
||
parent_pick = (
|
||
pick_fix_for_installed(verified.fix_candidates, vuln.package_version)
|
||
if verified.fix_candidates
|
||
else None
|
||
) or verified.fixed_version
|
||
|
||
if parent_pick:
|
||
# Adobe-year-truncation so a propagated `2018.011.20055`
|
||
# matches the format Wazuh/Nessus actually report.
|
||
parent_pick_norm = _normalize_for_display(parent_pick) or parent_pick
|
||
if not getattr(vuln, 'fixed_version', None):
|
||
# Skip when picked fix == installed (false-positive guard).
|
||
if not (vuln.package_version and vuln.package_version == parent_pick_norm):
|
||
changes["fields"].append("fixed_version")
|
||
changes["old"]["fixed_version"] = None
|
||
changes["new"]["fixed_version"] = parent_pick_norm
|
||
if not dry_run:
|
||
vuln.fixed_version = parent_pick_norm[:100]
|
||
changes["has_changes"] = True
|
||
|
||
# Per-package picker — each child row gets its OWN best-match.
|
||
if not dry_run:
|
||
pkg_fix_writes = 0
|
||
for pkg in getattr(vuln, "packages", []) or []:
|
||
if pkg.fixed_version:
|
||
continue
|
||
pkg_pick = (
|
||
pick_fix_for_installed(verified.fix_candidates, pkg.package_version)
|
||
if verified.fix_candidates
|
||
else None
|
||
) or verified.fixed_version
|
||
if not pkg_pick:
|
||
continue
|
||
pkg_pick = _normalize_for_display(pkg_pick) or pkg_pick
|
||
if pkg.package_version and pkg.package_version == pkg_pick:
|
||
continue
|
||
pkg.fixed_version = pkg_pick[:100]
|
||
pkg_fix_writes += 1
|
||
if pkg_fix_writes:
|
||
changes["fields"].append(f"packages.fixed_version (×{pkg_fix_writes})")
|
||
changes["has_changes"] = True
|
||
|
||
# Source pin — set exploitation_source whenever ANY override field
|
||
# changed, not just CVSS/severity. Earlier behaviour only pinned the
|
||
# source on CVSS/severity diffs, so a CVE whose Wazuh CVSS happened
|
||
# to already match Vulnrichment got SSVC fields written but
|
||
# exploitation_source stayed NULL. Tester filter
|
||
# WHERE exploitation_source='vulnrichment' AND exploitation_status != 'none'
|
||
# then under-reported by orders of magnitude (4 vs the real ~840).
|
||
if changes["has_changes"] and not dry_run:
|
||
vuln.exploitation_source = verified.source or "vulnrichment"
|
||
# Materialised sort columns — CVSS/severity/EPSS-adjacent
|
||
# changes shift priority. Cheap, in-process.
|
||
vuln.refresh_scores()
|
||
# Propagate the canonical CVE metadata to every sibling row
|
||
# so the same CVE on other assets doesn't keep showing
|
||
# stale Wazuh placeholders (10.0 → real score, etc.).
|
||
try:
|
||
propagate_canonical_to_siblings(self.db, vuln)
|
||
except Exception as e:
|
||
logger.warning(
|
||
"canonical propagation failed for %s: %s",
|
||
vuln.cve_id, e,
|
||
)
|
||
|
||
return changes
|
||
|
||
def correct_all_vulnerabilities(
|
||
self,
|
||
dry_run: bool = False
|
||
) -> dict:
|
||
"""
|
||
Korrigiert alle Vulnerabilities in der Datenbank.
|
||
|
||
Verwendet CISA Vulnrichment als primäre Korrekturquelle.
|
||
|
||
Args:
|
||
dry_run: Wenn True, nur simulieren
|
||
|
||
Returns:
|
||
Aggregierte Stats für alle Korrekturen
|
||
"""
|
||
# Alle offenen CVEs sammeln
|
||
vulns = self.db.query(Vulnerability).filter(
|
||
Vulnerability.status == VulnerabilityStatus.open
|
||
).all()
|
||
|
||
cve_ids = list(set(v.cve_id for v in vulns if v.cve_id and v.cve_id.upper().startswith("CVE-")))
|
||
|
||
if not cve_ids:
|
||
return {"checked": 0, "updated": 0, "message": "Keine CVEs zum Korrigieren"}
|
||
|
||
# Vulnrichment-Daten laden
|
||
verified_data = self.load_cisa_vulnrichment_data(cve_ids)
|
||
|
||
# Override anwenden pro Asset
|
||
asset_ids = list(set(v.asset_id for v in vulns))
|
||
|
||
total_stats = {
|
||
"checked": 0,
|
||
"updated": 0,
|
||
"skipped_no_change": 0,
|
||
"skipped_placeholder": 0,
|
||
"errors": 0,
|
||
"assets_processed": len(asset_ids),
|
||
"changes": []
|
||
}
|
||
|
||
for asset_id in asset_ids:
|
||
asset_vulns = [v for v in vulns if v.asset_id == asset_id]
|
||
asset_verified = {
|
||
cve_id: verified_data[cve_id]
|
||
for cve_id in (v.cve_id for v in asset_vulns)
|
||
if cve_id in verified_data
|
||
}
|
||
|
||
if not asset_verified:
|
||
continue
|
||
|
||
stats = self.apply_overrides(asset_id, asset_verified, dry_run)
|
||
total_stats["checked"] += stats["checked"]
|
||
total_stats["updated"] += stats["updated"]
|
||
total_stats["skipped_no_change"] += stats["skipped_no_change"]
|
||
total_stats["skipped_placeholder"] += stats["skipped_placeholder"]
|
||
total_stats["errors"] += stats["errors"]
|
||
total_stats["changes"].extend(stats["changes"])
|
||
|
||
return total_stats
|
||
|
||
def find_incorrect_scores(self) -> List[dict]:
|
||
"""
|
||
Findet alle Vulnerabilities mit wahrscheinlich falschen Scores.
|
||
|
||
Erkennt:
|
||
- Platzhalter-Scores (10.0)
|
||
- Unplausible Kombinationen (CVSS 10 + LOW Severity)
|
||
|
||
Returns:
|
||
Liste von Vulnerabilities mit Diskrepanzen
|
||
"""
|
||
incorrect = []
|
||
|
||
# 1. Alle mit Score 10.0 (Platzhalter)
|
||
placeholder_vulns = self.db.query(Vulnerability).filter(
|
||
Vulnerability.cvss_score == 10.0
|
||
).all()
|
||
|
||
for vuln in placeholder_vulns:
|
||
correct_severity = self._severity_from_cvss(7.3) # Typischer Korrekturwert
|
||
if vuln.severity == VulnerabilitySeverity.critical:
|
||
incorrect.append({
|
||
"vuln_id": vuln.id,
|
||
"cve_id": vuln.cve_id,
|
||
"asset_id": vuln.asset_id,
|
||
"current_score": vuln.cvss_score,
|
||
"current_severity": vuln.severity.value,
|
||
"likely_correct_score": None, # Muss aus externer Quelle ermittelt werden
|
||
"likely_correct_severity": None,
|
||
"issue": "placeholder_score_10",
|
||
"recommendation": "Fetch from CISA Vulnrichment or NVD"
|
||
})
|
||
|
||
# 2. Alle mit CVSS > 9 aber LOW/NONE Severity
|
||
misaligned = self.db.query(Vulnerability).filter(
|
||
Vulnerability.cvss_score >= 9.0,
|
||
Vulnerability.severity.in_([
|
||
VulnerabilitySeverity.low,
|
||
VulnerabilitySeverity.none,
|
||
VulnerabilitySeverity.medium
|
||
])
|
||
).all()
|
||
|
||
for vuln in misaligned:
|
||
incorrect.append({
|
||
"vuln_id": vuln.id,
|
||
"cve_id": vuln.cve_id,
|
||
"asset_id": vuln.asset_id,
|
||
"current_score": vuln.cvss_score,
|
||
"current_severity": vuln.severity.value,
|
||
"issue": "severity_mismatch",
|
||
"recommendation": "CVSS >= 9.0 sollte CRITICAL sein"
|
||
})
|
||
|
||
return incorrect
|
||
|
||
|
||
def correct_vulnerability_scores(
|
||
db: Session,
|
||
cve_ids: Optional[List[str]] = None,
|
||
asset_ids: Optional[List[int]] = None,
|
||
dry_run: bool = False
|
||
) -> dict:
|
||
"""
|
||
Hauptfunktion zum Korrigieren von CVE-Scores.
|
||
|
||
Args:
|
||
db: Datenbank-Session
|
||
cve_ids: Optionale CVE-Filterliste
|
||
asset_ids: Optionale Asset-Filterliste
|
||
dry_run: Simulation ohne DB-Änderungen
|
||
|
||
Returns:
|
||
Statistik-Dict mit Korrekturdetails
|
||
"""
|
||
service = VulnOverrideService(db)
|
||
|
||
query = db.query(Vulnerability).filter(
|
||
Vulnerability.status == VulnerabilityStatus.open
|
||
)
|
||
|
||
if cve_ids:
|
||
query = query.filter(Vulnerability.cve_id.in_(cve_ids))
|
||
if asset_ids:
|
||
query = query.filter(Vulnerability.asset_id.in_(asset_ids))
|
||
|
||
vulns = query.all()
|
||
|
||
if not vulns:
|
||
return {"message": "Keine Vulnerabilities zum Korrigieren", "updated": 0}
|
||
|
||
# Sammle alle CVE-IDs für Vulnrichment-Abfrage
|
||
all_cve_ids = list(set(v.cve_id for v in vulns if v.cve_id and v.cve_id.upper().startswith("CVE-")))
|
||
|
||
if not all_cve_ids:
|
||
return {"message": "Keine echten CVEs zum Korrigieren", "updated": 0}
|
||
|
||
# Lade verifizierte Daten
|
||
verified_data = service.load_cisa_vulnrichment_data(all_cve_ids)
|
||
|
||
# CVE-IDs, die im Feed FEHLEN (CISA hat sie noch nicht analysiert).
|
||
# Frontend nutzt diese Zahl, um den Unterschied zwischen "Feed down"
|
||
# und "neue CVE noch nicht im Feed" sichtbar zu machen.
|
||
not_found = len(all_cve_ids) - len(verified_data)
|
||
|
||
if not verified_data:
|
||
logger.warning("Keine verifizierten Daten von CISA Vulnrichment erhalten")
|
||
return {
|
||
"message": "CISA Vulnrichment-Daten nicht verfügbar",
|
||
"checked": len(vulns),
|
||
"updated": 0,
|
||
"not_found": not_found,
|
||
}
|
||
|
||
# Gruppiere nach Asset
|
||
asset_groups: Dict[int, List[str]] = {}
|
||
for vuln in vulns:
|
||
if vuln.cve_id in verified_data:
|
||
if vuln.asset_id not in asset_groups:
|
||
asset_groups[vuln.asset_id] = []
|
||
asset_groups[vuln.asset_id].append(vuln.cve_id)
|
||
|
||
total_stats = {
|
||
"checked": 0,
|
||
"updated": 0,
|
||
"errors": 0,
|
||
"not_found": not_found,
|
||
"changes": []
|
||
}
|
||
|
||
for asset_id, cve_list in asset_groups.items():
|
||
asset_verified = {
|
||
cve_id: verified_data[cve_id]
|
||
for cve_id in cve_list
|
||
if cve_id in verified_data
|
||
}
|
||
|
||
stats = service.apply_overrides(asset_id, asset_verified, dry_run)
|
||
total_stats["checked"] += stats["checked"]
|
||
total_stats["updated"] += stats["updated"]
|
||
total_stats["errors"] += stats["errors"]
|
||
total_stats["changes"].extend(stats["changes"])
|
||
|
||
return total_stats |