vulncheck
e0646077d6
Jev could only run over OpenRouter. The AI settings card now also takes a TypeSafe API key (setting typesafe_api_key, encrypted at rest like the OpenRouter key, or env TYPESAFE_API_KEY). When it is set, Jev goes straight to api.typesafe.ai/v1/systemone with the pinned jev-1.13.0. Without it, Jev uses the OpenRouter key and OpenRouter Decisions with typesafe/jev-1.13. Both endpoints take the same body and give the same answer. The card shows "Configured" when either key is set, and the TypeSafe key has its own Remove button. JEV_MODEL is gone. It overrode the model on both endpoints, but the two use different model ids. Adding or removing a TypeSafe key switched the endpoint and left the other endpoint's model id in place. Every CVE then failed as a CVE-specific 4xx and was skipped, up to 500 requests with nothing written and no error reported. The model is now fixed per endpoint. For the same failure shape from any other cause, such as a changed schema, five CVE-specific failures in a row now stop the run with an error. The run no longer walks through every CVE. A test checks the actual request: URL, bearer key, model and question.
640 lines
27 KiB
Python
640 lines
27 KiB
Python
"""
|
||
Vulnerability Model
|
||
"""
|
||
from enum import Enum
|
||
from sqlalchemy import (
|
||
Column, Integer, String, Float, Boolean, DateTime,
|
||
Text, ForeignKey, Enum as SQLEnum, Index
|
||
)
|
||
from sqlalchemy.orm import relationship
|
||
from app.models.base import Base, TimestampMixin
|
||
|
||
|
||
class VulnerabilitySeverity(str, Enum):
|
||
"""CVSS-basierte Severity-Levels"""
|
||
critical = "critical" # CVSS 9.0-10.0
|
||
high = "high" # CVSS 7.0-8.9
|
||
medium = "medium" # CVSS 4.0-6.9
|
||
low = "low" # CVSS 0.1-3.9
|
||
none = "none" # CVSS 0.0
|
||
|
||
|
||
class VulnerabilityStatus(str, Enum):
|
||
"""Status der Vulnerability"""
|
||
open = "open"
|
||
patched = "patched"
|
||
pending_verification = "pending_verification"
|
||
patch_failed = "patch_failed"
|
||
accepted_risk = "accepted_risk"
|
||
false_positive = "false_positive"
|
||
deferred = "deferred"
|
||
|
||
|
||
# Lower bound of each CVSS band. Used only when no score exists at all —
|
||
# some vendors publish none: Chrome states "Chromium security severity:
|
||
# Critical" in prose and NVD frequently never scores those CVEs. Both
|
||
# priority and CPR multiply the CVSS in, so a missing one drove them to zero
|
||
# and a genuinely critical finding sorted below a medium with a score. Taking
|
||
# the band's FLOOR keeps the estimate conservative: it can only ever
|
||
# understate a real score, never inflate one.
|
||
# Every vendor scale seen: FIRST's and Microsoft's (TeamViewer uses it too).
|
||
_VENDOR_RATINGS = {"critical", "important", "high", "moderate", "medium", "low"}
|
||
|
||
_SEVERITY_FLOOR = {"critical": 9.0, "high": 7.0, "medium": 4.0, "low": 0.1}
|
||
|
||
|
||
def cpr_score(cvss_score, severity, epss_score, epss_percentile):
|
||
"""CPR from raw column values — see Vulnerability.calculate_cpr_score.
|
||
|
||
Module-level so bulk scorers can run it over a lightweight column
|
||
query instead of hydrating one ORM object per finding.
|
||
"""
|
||
base = cvss_score
|
||
if base is None:
|
||
sev = getattr(severity, "value", severity)
|
||
base = _SEVERITY_FLOOR.get(str(sev or "").lower())
|
||
if base is None or epss_score is None:
|
||
return None
|
||
cvss_pct = max(0.0, min(base, 10.0)) * 10.0
|
||
if epss_percentile is not None:
|
||
epss_pct = max(0.0, min(epss_percentile, 1.0)) * 100.0
|
||
else:
|
||
# Fallback when only raw EPSS probability is stored — not a
|
||
# true percentile but the best we have. Older data only.
|
||
epss_pct = max(0.0, min(epss_score, 1.0)) * 100.0
|
||
cpr = (cvss_pct * 0.6) + (epss_pct * 0.4)
|
||
return round(min(cpr, 100.0), 2)
|
||
|
||
|
||
class Vulnerability(Base, TimestampMixin):
|
||
"""
|
||
Vulnerability-Tabelle für CVE-Tracking
|
||
|
||
Enthält Daten aus Wazuh sowie Enrichment-Informationen
|
||
"""
|
||
__tablename__ = "vulnerabilities"
|
||
__table_args__ = (
|
||
Index('uq_vuln_cve_asset', 'cve_id', 'asset_id', unique=True),
|
||
)
|
||
|
||
id = Column(Integer, primary_key=True, index=True)
|
||
|
||
# CVE-Identifikation. Widened to 50 chars in migration 010 because Nessus
|
||
# non-CVE findings (EOL, compliance, weak ciphers) are imported with a
|
||
# pseudo-CVE id of the form 'NESSUS-PLUGIN-{plugin_id}'.
|
||
cve_id = Column(String(50), nullable=False, index=True) # CVE-2024-1234 or NESSUS-PLUGIN-12345
|
||
# CASCADE: the ORM already cascades, but the database should not depend on
|
||
# the application getting the order right.
|
||
asset_id = Column(Integer, ForeignKey("assets.id", ondelete="CASCADE"),
|
||
nullable=False, index=True)
|
||
|
||
# CVSS-Bewertung
|
||
cvss_score = Column(Float, nullable=True, index=True)
|
||
cvss_vector = Column(String(100), nullable=True)
|
||
# The vendors' own ratings, JSON {"Microsoft": "Critical", ...}, written
|
||
# through set_vendor_severity. `severity` follows the CVSS, and vendors rate
|
||
# by their own scale — CVE-2026-85921 is CVSS 8.2 (High) but Critical at
|
||
# MSRC. Its own column, not appended to `description` like Mozilla's:
|
||
# M365 and Nessus rewrite that text every night.
|
||
vendor_severity = Column(Text, nullable=True)
|
||
severity = Column(
|
||
SQLEnum(VulnerabilitySeverity),
|
||
nullable=False,
|
||
index=True
|
||
)
|
||
|
||
# Status-Tracking
|
||
status = Column(
|
||
SQLEnum(VulnerabilityStatus),
|
||
default=VulnerabilityStatus.open,
|
||
nullable=False,
|
||
index=True
|
||
)
|
||
|
||
# Details
|
||
title = Column(Text, nullable=True)
|
||
description = Column(Text, nullable=True)
|
||
|
||
# Betroffene Software
|
||
package_name = Column(String(255), nullable=True, index=True)
|
||
package_vendor = Column(String(255), nullable=True) # Wazuh/Intune/Defender vendor|publisher
|
||
package_version = Column(String(100), nullable=True)
|
||
fixed_version = Column(String(100), nullable=True)
|
||
|
||
# Exploitation
|
||
exploitable = Column(Boolean, default=False, nullable=False, index=True)
|
||
exploit_available = Column(Boolean, default=False, nullable=False)
|
||
exploit_maturity = Column(String(50), nullable=True) # poc, functional, high
|
||
# SSVC Exploitation Status (NIST SP 500-299 Draft v2.0, 2024-05)
|
||
# Values: none, poc, active, widespread
|
||
exploitation_status = Column(String(20), nullable=True, index=True)
|
||
exploitation_source = Column(String(20), nullable=True) # nessus, vulrighment, nvd, manual
|
||
# Additional SSVC decision points (CISA Vulnrichment per-CVE)
|
||
ssvc_technical_impact = Column(String(16), nullable=True, index=True) # partial | total
|
||
ssvc_automatable = Column(String(8), nullable=True, index=True) # yes | no
|
||
|
||
# Zeitstempel
|
||
published_date = Column(DateTime, nullable=True, index=True) # NVD `published`
|
||
last_modified_date = Column(DateTime, nullable=True) # NVD `lastModified`
|
||
detected_at = Column(DateTime, nullable=False, index=True)
|
||
patched_at = Column(DateTime, nullable=True)
|
||
|
||
# Zusätzliche Informationen
|
||
references = Column(Text, nullable=True) # JSON-Array mit URLs
|
||
remediation = Column(Text, nullable=True) # Nessus solution / fix guidance
|
||
cwe_id = Column(String(20), nullable=True) # Common Weakness Enumeration
|
||
|
||
# Assignment
|
||
assigned_user_id = Column(Integer, ForeignKey("users.id"), nullable=True, index=True)
|
||
assigned_group_id = Column(Integer, ForeignKey("groups.id"), nullable=True, index=True)
|
||
|
||
# Deferral
|
||
deferred_until = Column(DateTime, nullable=True)
|
||
defer_reason = Column(Text, nullable=True)
|
||
|
||
# Notification Control
|
||
notification_suppressed = Column(Boolean, default=False, nullable=False)
|
||
# Imported with a container image repository's first report: already
|
||
# there before anyone looked, so no new-finding mail (059). Not a mute —
|
||
# SLA and KEV alerts still apply.
|
||
baseline = Column(Boolean, default=False, nullable=False, server_default="false")
|
||
|
||
# Enrichment: EPSS (Exploit Prediction Scoring System - FIRST.org)
|
||
epss_score = Column(Float, nullable=True, index=True) # 0.0 - 1.0 probability of exploitation in next 30 days
|
||
epss_percentile = Column(Float, nullable=True) # 0.0 - 1.0 percentile rank
|
||
epss_updated_at = Column(DateTime, nullable=True)
|
||
|
||
# Enrichment: CISA KEV (Known Exploited Vulnerabilities)
|
||
kev_listed = Column(Boolean, default=False, nullable=False, index=True)
|
||
kev_date_added = Column(DateTime, nullable=True)
|
||
kev_ransomware_use = Column(Boolean, default=False, nullable=False)
|
||
kev_short_description = Column(Text, nullable=True)
|
||
|
||
# Enrichment: ENISA EUVD (EU exploited / critical vulnerabilities)
|
||
euvd_listed = Column(Boolean, default=False, nullable=False, index=True)
|
||
euvd_critical = Column(Boolean, default=False, nullable=False)
|
||
euvd_date_added = Column(DateTime, nullable=True)
|
||
euvd_id = Column(String(50), nullable=True)
|
||
|
||
# Generic enrichment tracking
|
||
enrichment_sources = Column(Text, nullable=True) # JSON array, e.g. ["epss","kev"]
|
||
enrichment_updated_at = Column(DateTime, nullable=True)
|
||
|
||
# Multi-scanner source tracking (Nessus integration, migration 010).
|
||
# `sources` is a JSON list of scanner names that detected this finding
|
||
# (e.g. ["wazuh"], ["nessus"], ["wazuh","nessus"]). Cross-scanner
|
||
# confirmation = highest confidence. Status changes (e.g. PATCHED backfill)
|
||
# only fire when *all* listed scanners stop reporting the CVE.
|
||
sources = Column(Text, nullable=False, default='["wazuh"]', server_default='[]')
|
||
nessus_plugin_id = Column(String(20), nullable=True, index=True)
|
||
nessus_finding_uuid = Column(String(64), nullable=True)
|
||
first_detected_by = Column(String(20), nullable=True) # wazuh|nessus|manual
|
||
# Tenable's proprietary VPR (Vulnerability Priority Rating, 0-10).
|
||
# Independent from our own priority_score — surfaced for users who
|
||
# already rely on Tenable's ranking.
|
||
nessus_vpr_score = Column(Float, nullable=True, index=True)
|
||
|
||
# Materialised priority + CPR scores so the list endpoint can SQL-sort
|
||
# across the entire filter result instead of in-memory sorting the
|
||
# current page only (was a reported bug: "PRIO sort applies only to
|
||
# the 100 rows on screen, not to all 15k CVEs"). Updated by every
|
||
# sync / enrich / override path via `Vulnerability.refresh_scores()`.
|
||
priority_score = Column(Float, nullable=True, index=True)
|
||
cpr_score = Column(Float, nullable=True, index=True)
|
||
|
||
# Plan M — exploit-intel enrichment (Exploit-DB / PoC-in-GitHub /
|
||
# Metasploit). Cve-id mapped to public-exploit references. The
|
||
# `_count` columns are denormalised for fast SQL sort/filter
|
||
# ("show me CVEs with ≥ 1 metasploit module").
|
||
# Lists are JSON strings to stay schema-stable. Counts populated
|
||
# by `enrichment_service.exploit_intel_refresh()` (24h cache).
|
||
exploit_db_ids = Column(Text, nullable=True) # JSON [edb_id, ...]
|
||
exploit_db_count = Column(Integer, nullable=False, default=0, server_default="0")
|
||
pocs_github_urls = Column(Text, nullable=True) # JSON [url, ...]
|
||
pocs_github_count = Column(Integer, nullable=False, default=0, server_default="0")
|
||
metasploit_modules = Column(Text, nullable=True) # JSON [module_path, ...]
|
||
metasploit_module_count = Column(Integer, nullable=False, default=0, server_default="0")
|
||
exploit_intel_updated_at = Column(DateTime, nullable=True)
|
||
|
||
# Jev (TypeSafe, direct or via OpenRouter Decisions) — probability that the CVE is
|
||
# only exploitable with an optional feature / non-default configuration.
|
||
# A HINT for the operator to check, per CVE text: it never closes a
|
||
# finding and never enters priority_score. See jev_triage_service.
|
||
jev_config_dependent = Column(Float, nullable=True)
|
||
jev_checked_at = Column(DateTime, nullable=True)
|
||
|
||
# Relationships
|
||
asset = relationship("Asset", back_populates="vulnerabilities")
|
||
assigned_user = relationship("User", foreign_keys=[assigned_user_id])
|
||
group = relationship("Group", foreign_keys=[assigned_group_id], back_populates="vulnerabilities")
|
||
ai_analysis = relationship(
|
||
"AIAnalysis",
|
||
back_populates="vulnerability",
|
||
uselist=False,
|
||
cascade="all, delete-orphan"
|
||
)
|
||
|
||
def __repr__(self):
|
||
return f"<Vulnerability(cve='{self.cve_id}', severity='{self.severity}')>"
|
||
|
||
# ------------------------------------------------------------------
|
||
# Multi-source helpers (Nessus integration)
|
||
# ------------------------------------------------------------------
|
||
@property
|
||
def source_list(self) -> list:
|
||
"""Parsed `sources` JSON column. Empty list on malformed JSON."""
|
||
import json
|
||
if not self.sources:
|
||
return []
|
||
try:
|
||
value = json.loads(self.sources)
|
||
return value if isinstance(value, list) else []
|
||
except (json.JSONDecodeError, TypeError):
|
||
return []
|
||
|
||
@property
|
||
def cross_confirmed(self) -> bool:
|
||
"""True when the same finding was reported by 2+ independent scanners."""
|
||
return len(self.source_list) >= 2
|
||
|
||
@property
|
||
def is_pseudo_cve(self) -> bool:
|
||
"""True for non-CVE findings rendered as rows in the vulns list.
|
||
NESSUS-PLUGIN-* (compliance/cipher/EOL detected by Nessus), EOL-*
|
||
(endoflife.date scan), ANDROID-PATCH-* (patch-level gap) and TV-*
|
||
(a TeamViewer bulletin the vendor published without requesting a CVE —
|
||
5 of their 30). They stay in the reports, because they are real work
|
||
items; the badge says only that the id is not a CVE number."""
|
||
return (self.cve_id.startswith("NESSUS-PLUGIN-")
|
||
or self.cve_id.startswith("EOL-")
|
||
or self.cve_id.startswith("ANDROID-PATCH-")
|
||
or self.cve_id.startswith("TV-"))
|
||
|
||
@property
|
||
def is_eol_finding(self) -> bool:
|
||
"""True for endoflife.date pseudo-CVEs (`EOL-…`)."""
|
||
return self.cve_id.startswith("EOL-")
|
||
|
||
def add_source(self, scanner: str) -> bool:
|
||
"""Append `scanner` to `sources` if not already present. Returns True
|
||
if the list changed. Caller is responsible for committing."""
|
||
import json
|
||
current = self.source_list
|
||
if scanner in current:
|
||
return False
|
||
current.append(scanner)
|
||
self.sources = json.dumps(current)
|
||
return True
|
||
|
||
@property
|
||
def reference_list(self) -> list:
|
||
"""Parsed `references` JSON column (URLs). Empty on malformed JSON."""
|
||
import json
|
||
if not self.references:
|
||
return []
|
||
try:
|
||
value = json.loads(self.references)
|
||
except (json.JSONDecodeError, TypeError):
|
||
return []
|
||
if not isinstance(value, list):
|
||
return []
|
||
return [u for u in value if isinstance(u, str)]
|
||
|
||
def add_reference(self, url: str) -> bool:
|
||
"""Append `url` to `references` if not already present. Returns True
|
||
if the list changed. Caller is responsible for committing.
|
||
|
||
The column is shared: Nessus writes its see_also here, vendor-bulletin
|
||
scanners their advisory, the M365 check its release-notes section. So
|
||
a source adds its own evidence and never drops another's — the
|
||
cross-source contract holds for evidence, not just for `sources`.
|
||
"""
|
||
import json
|
||
if not url:
|
||
return False
|
||
current = self.reference_list
|
||
if url in current:
|
||
return False
|
||
current.append(url)
|
||
self.references = json.dumps(current)
|
||
return True
|
||
|
||
def remove_source(self, scanner: str) -> bool:
|
||
"""Drop `scanner` from `sources`. Returns True if list changed.
|
||
Caller is responsible for committing."""
|
||
import json
|
||
current = self.source_list
|
||
if scanner not in current:
|
||
return False
|
||
current = [s for s in current if s != scanner]
|
||
self.sources = json.dumps(current)
|
||
return True
|
||
|
||
@property
|
||
def vendor_severities(self) -> dict:
|
||
import json
|
||
try:
|
||
got = json.loads(self.vendor_severity or "{}")
|
||
except (TypeError, ValueError):
|
||
return {}
|
||
return got if isinstance(got, dict) else {}
|
||
|
||
def set_vendor_severity(self, vendor: str, rating) -> bool:
|
||
"""Record `vendor`'s rating, title-cased. Words off the scale
|
||
("Informative") or none at all are not a rating and never remove
|
||
one: a bulletin that says nothing is not a retraction."""
|
||
import json
|
||
word = str(rating or "").strip().title()
|
||
if not vendor or word.lower() not in _VENDOR_RATINGS:
|
||
return False
|
||
cur = self.vendor_severities
|
||
if cur.get(vendor) == word:
|
||
return False
|
||
cur[vendor] = word
|
||
self.vendor_severity = json.dumps(cur, sort_keys=True)
|
||
return True
|
||
|
||
def sync_severity_with_cvss(self) -> bool:
|
||
"""Keep `severity` consistent with `cvss_score`.
|
||
|
||
Sources that carry no score seed a neutral placeholder (MSRC only
|
||
publishes a CVSS for a minority of its CVEs), and whatever fills the
|
||
score later — the correction cascade, sibling inheritance — used to
|
||
leave the placeholder behind: Edge findings were seen sitting at
|
||
MEDIUM with a CVSS of 9.6. Never touches a row whose severity an
|
||
operator pinned via the override path (exploitation_source set).
|
||
"""
|
||
if self.cvss_score is None:
|
||
return False
|
||
if getattr(self, "exploitation_source", None) == "manual":
|
||
return False
|
||
s = float(self.cvss_score)
|
||
want = (VulnerabilitySeverity.critical if s >= 9.0 else
|
||
VulnerabilitySeverity.high if s >= 7.0 else
|
||
VulnerabilitySeverity.medium if s >= 4.0 else
|
||
VulnerabilitySeverity.low if s > 0 else
|
||
VulnerabilitySeverity.none)
|
||
if self.severity != want:
|
||
self.severity = want
|
||
return True
|
||
return False
|
||
|
||
def refresh_scores(self) -> None:
|
||
"""Recompute + persist priority_score + cpr_score on the row.
|
||
|
||
Call after any change that affects the breakdown (sync, enrich,
|
||
override, manual edits). Cheap — purely Python, no I/O. The list
|
||
endpoint sorts on the materialised columns so the order is global,
|
||
not per-page.
|
||
"""
|
||
try:
|
||
self.sync_severity_with_cvss()
|
||
self.priority_score = self.calculate_priority_breakdown().get("total")
|
||
except Exception:
|
||
self.priority_score = None
|
||
try:
|
||
self.cpr_score = self.calculate_cpr_score()
|
||
except Exception:
|
||
self.cpr_score = None
|
||
|
||
def calculate_priority_score(self) -> float:
|
||
"""
|
||
Berechnet einen Priority-Score basierend auf:
|
||
- Technical Severity (CVSS)
|
||
- Exploit-Signal (max aus: CISA KEV, EUVD, EPSS, Wazuh-Flags)
|
||
- Catalog-Bonus (KEV/EUVD oben drauf, mit Stacking)
|
||
- Asset Policy (Criticality) via Gruppen
|
||
- Age der Vulnerability
|
||
|
||
Returns: Score zwischen 0-100
|
||
"""
|
||
return self.calculate_priority_breakdown()["total"]
|
||
|
||
def _severity_floor(self):
|
||
sev = getattr(self.severity, "value", self.severity)
|
||
return _SEVERITY_FLOOR.get(str(sev or "").lower())
|
||
|
||
def calculate_cpr_score(self):
|
||
"""
|
||
Cybersecurity Priority Risk — weighted blend of CVSS + EPSS
|
||
percentiles (Range 0-100). Follows the JacquesKruger/EPSS-Server
|
||
algorithm:
|
||
|
||
CPR = (CVSS_Percentile × 0.6) + (EPSS_Percentile × 0.4)
|
||
|
||
Where:
|
||
- CVSS_Percentile = CVSS × 10 (CVSS is 0-10, map to 0-100)
|
||
- EPSS_Percentile = epss_percentile field × 100, OR fallback to
|
||
epss_score × 100 if the percentile is missing (older syncs).
|
||
|
||
Weights bias towards CVSS (technical severity) at 60% and EPSS
|
||
(exploitation probability) at 40% — the published default.
|
||
|
||
Previous implementation was a pure multiplicative
|
||
``CVSS × EPSS × 10`` which collapsed to near-zero for the 99% of
|
||
CVEs that have EPSS < 1%, making the score useless for triage
|
||
(field report: CVE-2026-8390 with CVSS 9.8 + EPSS 0.04% scored
|
||
0.04 — a 'critical-CVSS, low-real-world-risk' finding now scores
|
||
~60 instead, which actually reflects its triage position).
|
||
|
||
Returns: float 0-100, oder None wenn CVSS oder EPSS fehlt.
|
||
"""
|
||
return cpr_score(
|
||
self.cvss_score, self.severity, self.epss_score, self.epss_percentile
|
||
)
|
||
|
||
def calculate_priority_breakdown(self) -> dict:
|
||
"""
|
||
Wie calculate_priority_score, gibt aber alle Komponenten zurück
|
||
damit das Frontend die Begründung anzeigen kann.
|
||
"""
|
||
from datetime import datetime, timezone
|
||
|
||
# 1. Technical Severity (0-10). Falls back to the severity band's floor
|
||
# when the vendor published no CVSS — see _SEVERITY_FLOOR. Without it a
|
||
# Critical Chrome CVE scored 0 here and sorted below every medium.
|
||
technical_severity = (self.cvss_score if self.cvss_score is not None
|
||
else (self._severity_floor() or 0.0))
|
||
|
||
# 2. Exploit-Signal: max aus KEV/EUVD / EPSS / Wazuh / SSVC / maturity
|
||
# KEV ODER EUVD bedeuten "real-world exploitation bestätigt"
|
||
# (CISA US-Quelle, ENISA EU-Quelle) -> beide gleichwertig
|
||
# EPSS skaliert kontinuierlich (probabilistisch)
|
||
# Wazuh-Flags als Fallback wenn keine externe Daten
|
||
catalog_listed = self.kev_listed or self.euvd_listed
|
||
kev_signal = 4.0 if catalog_listed else 0.0
|
||
epss_signal = 3.0 * (self.epss_score or 0.0) # max 3.0 bei epss=1.0
|
||
wazuh_signal = 0.0
|
||
if self.exploit_available:
|
||
wazuh_signal = 4.0
|
||
elif self.exploitable:
|
||
wazuh_signal = 2.0
|
||
# SSVC exploitation_status: härter als ein bool, weicher als KEV.
|
||
# active/widespread sind reale Beobachtungen — nicht doppelt zählen
|
||
# zu KEV (max() unten), aber als Fallback wenn KEV/EUVD leer sind.
|
||
ssvc_signal = 0.0
|
||
if self.exploitation_status == "widespread":
|
||
ssvc_signal = 5.0
|
||
elif self.exploitation_status == "active":
|
||
ssvc_signal = 4.0
|
||
elif self.exploitation_status == "poc":
|
||
ssvc_signal = 2.0
|
||
# Nessus exploit_code_maturity. Tenable rankt aufsteigend:
|
||
# Unproven (theoretisch) → PoC → Functional → High (waffenfertig).
|
||
# Stärkste Stufen kommen near-KEV. Max() unten verhindert
|
||
# Doppel-Counting mit Wazuh-Exploit-Flag.
|
||
maturity_signal = 0.0
|
||
mat = (self.exploit_maturity or "").lower()
|
||
if mat in ("high", "weaponized"):
|
||
maturity_signal = 4.0
|
||
elif mat == "functional":
|
||
maturity_signal = 3.0
|
||
elif mat in ("poc", "proof-of-concept", "proof of concept"):
|
||
maturity_signal = 1.5
|
||
# Plan M — public-exploit catalogs. Highest weight on
|
||
# Metasploit (weaponised, drag-and-drop), then Exploit-DB
|
||
# (working PoC), then PoC-in-GitHub (raw research code).
|
||
# max() avoids double-count with KEV/SSVC for CVEs that
|
||
# appear in everything.
|
||
msf_signal = 4.5 if (self.metasploit_module_count or 0) > 0 else 0.0
|
||
edb_signal = 4.0 if (self.exploit_db_count or 0) > 0 else 0.0
|
||
poc_signal = 2.5 if (self.pocs_github_count or 0) > 0 else 0.0
|
||
|
||
exploit_bonus = max(
|
||
kev_signal, epss_signal, wazuh_signal,
|
||
ssvc_signal, maturity_signal,
|
||
msf_signal, edb_signal, poc_signal,
|
||
)
|
||
|
||
# Add-on: SSVC Technical Impact + Automatable stack ON TOP of the
|
||
# max-merged exploit_bonus because they describe DIFFERENT
|
||
# dimensions (impact + automation) rather than overlapping
|
||
# exploitation signals. Capped together at +3 so they cannot
|
||
# outweigh the catalog/EPSS/maturity signals.
|
||
ssvc_addon = 0.0
|
||
if self.ssvc_technical_impact == "total":
|
||
ssvc_addon += 2.0 # attacker takeover scope
|
||
if self.ssvc_automatable == "yes":
|
||
ssvc_addon += 1.0 # reliable mass exploitation
|
||
ssvc_addon = min(ssvc_addon, 3.0)
|
||
exploit_bonus += ssvc_addon
|
||
|
||
# 3. Asset Context (Policy Lookup)
|
||
policies = []
|
||
if self.asset and hasattr(self.asset, 'groups') and self.asset.groups:
|
||
policies = [g.policy for g in self.asset.groups if g.policy]
|
||
|
||
sla_days_list = [p.critical_sla_days for p in policies if p.critical_sla_days is not None]
|
||
|
||
if not policies or not sla_days_list:
|
||
min_days = 90 # Fallback
|
||
else:
|
||
min_days = min(sla_days_list)
|
||
|
||
if min_days <= 3:
|
||
asset_factor = 2.0 # Critical
|
||
elif min_days <= 7:
|
||
asset_factor = 1.5 # High
|
||
elif min_days <= 30:
|
||
asset_factor = 1.0 # Medium
|
||
else:
|
||
asset_factor = 0.8 # Low
|
||
|
||
# 4. Age Add-On
|
||
age_bonus = 0.0
|
||
if self.detected_at:
|
||
try:
|
||
days_old = (datetime.now() - self.detected_at).days
|
||
except TypeError:
|
||
days_old = (datetime.now(timezone.utc) - self.detected_at).days
|
||
if days_old > 0:
|
||
age_bonus = min((days_old / 30.0) * 20.0, 20.0)
|
||
|
||
base_score = (technical_severity + exploit_bonus) * asset_factor
|
||
|
||
# 5. Catalog-Bonus on top (Stacking pro bestätigender Quelle)
|
||
# KEV und EUVD sind unabhängige real-world-Bestätigungen.
|
||
# CVE in beiden Catalogs = höchste Confidence = +20.
|
||
kev_bonus = 10.0 if self.kev_listed else 0.0
|
||
if self.kev_ransomware_use:
|
||
kev_bonus += 5.0
|
||
|
||
euvd_bonus = 10.0 if self.euvd_listed else 0.0
|
||
if self.euvd_critical:
|
||
euvd_bonus += 3.0
|
||
|
||
catalog_count = int(self.kev_listed) + int(self.euvd_listed)
|
||
|
||
total_score = base_score + age_bonus + kev_bonus + euvd_bonus
|
||
total_score = min(total_score, 100.0)
|
||
|
||
# CPR Score (CVSS × EPSS × 10)
|
||
cpr = self.calculate_cpr_score()
|
||
|
||
# exploit_source: erkennt welcher Catalog/Quelle dominiert.
|
||
# Now considers SSVC + maturity signals too. Reported as the
|
||
# single strongest source so the breakdown UI stays compact.
|
||
signals_ranked = [
|
||
("kev_and_euvd" if (self.kev_listed and self.euvd_listed)
|
||
else "kev" if self.kev_listed
|
||
else "euvd" if self.euvd_listed
|
||
else None, kev_signal),
|
||
("metasploit", msf_signal),
|
||
("ssvc", ssvc_signal),
|
||
("exploit_db", edb_signal),
|
||
("maturity", maturity_signal),
|
||
("epss", epss_signal),
|
||
("wazuh", wazuh_signal),
|
||
("poc_github", poc_signal),
|
||
]
|
||
# Drop sources with zero signal, keep the highest by signal value.
|
||
active = [(s, v) for s, v in signals_ranked if s and v > 0]
|
||
if active:
|
||
active.sort(key=lambda t: t[1], reverse=True)
|
||
exploit_source = active[0][0]
|
||
else:
|
||
exploit_source = "none"
|
||
|
||
return {
|
||
"total": total_score,
|
||
"cvss": technical_severity,
|
||
"exploit_bonus": exploit_bonus,
|
||
"exploit_source": exploit_source,
|
||
"exploit_status": self.exploitation_status,
|
||
"ssvc_addon": ssvc_addon,
|
||
"ssvc_technical_impact": self.ssvc_technical_impact,
|
||
"ssvc_automatable": self.ssvc_automatable,
|
||
"exploit_maturity": self.exploit_maturity,
|
||
"asset_factor": asset_factor,
|
||
"age_bonus": age_bonus,
|
||
"kev_bonus": kev_bonus,
|
||
"euvd_bonus": euvd_bonus,
|
||
"catalog_count": catalog_count,
|
||
"cpr": cpr,
|
||
"vpr_score": self.nessus_vpr_score,
|
||
"epss_score": self.epss_score,
|
||
"epss_percentile": self.epss_percentile,
|
||
"kev_listed": self.kev_listed,
|
||
"kev_ransomware_use": self.kev_ransomware_use,
|
||
"euvd_listed": self.euvd_listed,
|
||
"euvd_critical": self.euvd_critical,
|
||
# Plan M — public-exploit catalogs
|
||
"exploit_db_count": self.exploit_db_count or 0,
|
||
"pocs_github_count": self.pocs_github_count or 0,
|
||
"metasploit_module_count": self.metasploit_module_count or 0,
|
||
}
|
||
|
||
|
||
# Composite Index für häufige Abfragen
|
||
Index(
|
||
'idx_vuln_status_severity',
|
||
Vulnerability.status,
|
||
Vulnerability.severity
|
||
)
|
||
|
||
Index(
|
||
'idx_vuln_asset_status',
|
||
Vulnerability.asset_id,
|
||
Vulnerability.status
|
||
)
|