vulncheck/vulncheck
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
373dc9d813dd33688022129e6481d2972c94bb96
vulncheck/app/routers/vulnerabilities.py
T
vulncheck 373dc9d813 feat(eol): endoflife.date EOL detection — pseudo-CVE per finding 
Tester request: close the unsupported-software gap Wazuh has vs
Nessus plugin 64784 (Microsoft SQL Server Unsupported Version
Detection). PowerShell prototype against endoflife.date API
confirmed the approach.

Backend
- New app/services/eol_service.py
  * Hand-curated product slug map (Microsoft / Mozilla / Adobe /
    runtimes / databases / Linux distros).
  * resolve_product_slug() normalises Wazuh/Nessus product names and
    matches against the map.
  * fetch_product() hits api/v1/products/{slug} with 24h cache in
    the settings table (negative-caches 404s).
  * check_eol() picks the longest-prefix release for the installed
    version, evaluates eolFrom / eoasFrom / eoesFrom against today.
  * upsert_eol_vulnerability() writes one pseudo-CVE per (asset,
    product, release) with cve_id of the form
    EOL-{SLUG}-{RELEASE} so re-runs converge.
- Vulnerability.is_pseudo_cve also matches EOL-* now.
- Vulnerability.is_eol_finding new property + API field.

API + scheduler
- POST /vulnerabilities/eol-check (editor) — optional asset_id query,
  walks syscollector packages, upserts EOL pseudo-CVEs. Returns
  stats {assets_scanned, packages_checked, products_unmapped,
  eol_findings_total, eol_findings_new, errors[]}.
- New scheduler job eol_check_nightly at 03:15 UTC.

Frontend
- Vulnerability type gains is_eol_finding.
- List page renders purple "EOL" badge instead of amber "NON-CVE"
  for endoflife rows.

Out of scope (later iteration):
- OS-level EOL (Windows / Ubuntu releases) — would parse Asset.os
  + os_version separately.
- Per-package detail (VulnerabilityPackage child rows) for EOL
  pseudo-CVEs — current row already carries name/version/latest.
- Settings-UI for the product-slug mapping table.
2026-05-27 11:31:27 +02:00

2524 lines
96 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
"""
Vulnerability Management Router
Endpoints for CVE management, prioritization, and AI analysis.
"""
from typing import Optional, List
from datetime import datetime
from fastapi import APIRouter, Depends, HTTPException, status, Query, BackgroundTasks
from sqlalchemy.orm import Session
from sqlalchemy import desc, asc, func
from sqlalchemy.exc import IntegrityError
from pydantic import BaseModel, field_validator
import json
import os
import logging
logger = logging.getLogger(__name__)
from app.database import get_db
from app.models.user import User
from app.models.vulnerability import Vulnerability, VulnerabilitySeverity, VulnerabilityStatus
from app.models.asset import Asset
from app.models.ai_analysis import AIAnalysis
from app.models.audit_log import AuditLog, AuditEventType
from app.auth.dependencies import get_current_user, RequireEditor
from app.integrations.wazuh_client import get_wazuh_client, WazuhClient
from app.models.setting import Setting
from app.integrations.infomaniak_ai_client import InfomaniakAIClient, get_infomaniak_ai_client, AIAnalysisService
from app.models.ai_report import AIReport
from app.services.enrichment_service import (
enrich_vulnerabilities,
enrich_vulnerability_by_id,
enrich_all_open_vulnerabilities,
fetch_kev_catalog,
EnrichmentError,
)
from app.services.vuln_override_service import (
correct_vulnerability_scores,
VulnOverrideService,
)
from app.services.override_jobs import (
start_vulnrichment_job,
get_job as get_override_job,
list_jobs as list_override_jobs,
)
# Settings key for Nessus configuration
SETTING_KEY = "nessus_config"
router = APIRouter(prefix="/api/v1/vulnerabilities", tags=["Vulnerabilities"])
# ============================================
# Pydantic Schemas
# ============================================
class VulnerabilityResponse(BaseModel):
id: int
cve_id: str
asset_id: int
asset_hostname: str
cvss_score: Optional[float]
severity: VulnerabilitySeverity
status: VulnerabilityStatus
title: Optional[str]
package_name: Optional[str]
package_version: Optional[str]
fixed_version: Optional[str]
exploitable: bool
exploit_available: bool
detected_at: datetime
priority_score: float
priority_breakdown: Optional[dict] = None
cpr_score: Optional[float] = None
epss_score: Optional[float] = None
epss_percentile: Optional[float] = None
kev_listed: bool = False
kev_ransomware_use: bool = False
kev_date_added: Optional[datetime] = None
euvd_listed: bool = False
euvd_critical: bool = False
euvd_date_added: Optional[datetime] = None
euvd_id: Optional[str] = None
enrichment_sources: Optional[str] = None
enrichment_updated_at: Optional[datetime] = None
# Multi-scanner attribution (Nessus integration)
sources: List[str] = []
cross_confirmed: bool = False
first_detected_by: Optional[str] = None
nessus_plugin_id: Optional[str] = None
nessus_vpr_score: Optional[float] = None
exploitation_status: Optional[str] = None
exploitation_source: Optional[str] = None # vulnrichment | nvd | cvelistv5 | nessus
ssvc_technical_impact: Optional[str] = None # partial | total
ssvc_automatable: Optional[str] = None # yes | no
is_pseudo_cve: bool = False
is_eol_finding: bool = False # endoflife.date pseudo-CVE
assigned_user_id: Optional[int] = None
assigned_user_name: Optional[str] = None
assigned_group_id: Optional[int] = None
assigned_group_name: Optional[str] = None
deferred_until: Optional[datetime] = None
defer_reason: Optional[str] = None
notification_suppressed: bool = False
# Per-package detail (1:N child rows). Empty for pseudo-CVEs.
packages: List["PackageInfo"] = []
has_fix_any: bool = False # at least one package has a real fix
class Config:
from_attributes = True
class PackageInfo(BaseModel):
package_name: str
package_version: Optional[str] = None
fixed_version: Optional[str] = None
source: Optional[str] = None
has_fix: bool = False
class Config:
from_attributes = True
class AssignRequest(BaseModel):
user_id: Optional[int] = None
group_id: Optional[int] = None
class DeferRequest(BaseModel):
deferred_until: datetime
reason: str
class VulnerabilityDetailResponse(VulnerabilityResponse):
description: Optional[str]
cvss_vector: Optional[str]
exploit_maturity: Optional[str]
published_date: Optional[datetime]
patched_at: Optional[datetime]
references: Optional[str]
cwe_id: Optional[str]
ai_analysis: Optional[dict]
# Resolve forward ref — VulnerabilityResponse declares `packages: List["PackageInfo"]`
# above the PackageInfo class itself.
VulnerabilityResponse.model_rebuild()
VulnerabilityDetailResponse.model_rebuild()
class VulnerabilityUpdateRequest(BaseModel):
status: VulnerabilityStatus
# `reason` is status-agnostic — captures WHO/HOW for the audit trail
# regardless of which status the row moves to (patched via WSUS,
# FP because vendor confirmed N/A, accepted_risk per ITSM ticket).
# `defer_reason` kept for backward compatibility on existing UI.
reason: Optional[str] = None
defer_reason: Optional[str] = None
deferred_until: Optional[datetime] = None
@field_validator('status', mode='before')
@classmethod
def normalize_status(cls, v):
if isinstance(v, str):
return v.lower()
return v
class BulkVulnerabilityUpdateRequest(BaseModel):
vulnerability_ids: List[int]
assigned_user_id: Optional[int] = None
assigned_group_id: Optional[int] = None
status: Optional[VulnerabilityStatus] = None
# Optional reason — when blank, backend auto-fills with
# "Manually set to {status} by {username} on {datetime}" so the
# audit trail always tells WHO + WHEN even on the bulk-action path.
reason: Optional[str] = None
@field_validator('status', mode='before')
@classmethod
def normalize_status(cls, v):
if isinstance(v, str):
return v.lower()
return v
class VulnerabilityFilterParams(BaseModel):
severity: Optional[VulnerabilitySeverity] = None
status: Optional[VulnerabilityStatus] = None
exploitable: Optional[bool] = None
asset_id: Optional[int] = None
search: Optional[str] = None # Suche in CVE-ID, Package-Name, Title
class DashboardStatsResponse(BaseModel):
total_vulnerabilities: int
open_vulnerabilities: int
critical_count: int
high_count: int
medium_count: int
low_count: int
exploitable_count: int
avg_cvss_score: float
affected_assets: int
total_assets: int
scanned_assets: int
severity_history: List[dict]
oldest_vulnerability_days: int
# ============================================
# Helper-Funktionen
# ============================================
def _auto_reason(new_status, user_name: str) -> str:
"""Fallback reason string when the user submitted no comment.
Tester report: bulk + single status changes accepted blank
comments → audit trail couldn't distinguish manual patches from
scanner auto-patches. Auto-stamp ensures every manual change
still carries who/when in the audit log.
"""
ts = datetime.now().strftime("%Y-%m-%d %H:%M UTC")
label = new_status.value if hasattr(new_status, "value") else str(new_status)
return f"Manually set to {label} by {user_name} on {ts}"
def log_vulnerability_change(
db: Session,
user_id: Optional[int],
vuln_id: int,
old_status,
new_status,
reason: Optional[str] = None,
cve_id: Optional[str] = None,
source: Optional[str] = None,
):
"""Audit-Log entry for a vulnerability status change.
`user_id=None` for automated transitions (Nessus / Wazuh sync
auto-patching when a CVE disappears from the scan result).
`reason` is a free-text rationale persisted as JSON in new_value
alongside the new status — readable by the audit-log UI.
`source` tags WHERE the change came from
(`manual` / `nessus_sync` / `wazuh_sync` / `verify_patch_rescan`).
"""
old_val = old_status.value if hasattr(old_status, "value") else str(old_status)
new_val = new_status.value if hasattr(new_status, "value") else str(new_status)
new_payload = json.dumps({
"status": new_val,
"reason": reason,
"source": source or "manual",
"cve_id": cve_id,
})
audit_log = AuditLog(
user_id=user_id,
event_type=AuditEventType.VULNERABILITY_UPDATED,
event_description=(
f"Vulnerability {cve_id or ''} status changed: {old_val} → {new_val}"
+ (f" ({reason})" if reason else "")
)[:500],
resource_type="vulnerability",
resource_id=str(vuln_id),
old_value=old_val,
new_value=new_payload,
timestamp=datetime.now()
)
db.add(audit_log)
db.commit()
def _build_vuln_response(vuln: Vulnerability) -> dict:
"""Build a vulnerability response dict with assigned user info"""
breakdown = vuln.calculate_priority_breakdown()
# Per-package detail (1:N). Cheaper than a JOIN in queries that
# already loaded the parent — SQLAlchemy's default-lazy load is
# one extra round-trip per vuln. Fine for detail; list endpoints
# may want a joinedload upstream later.
pkgs = getattr(vuln, "packages", []) or []
packages_payload = [
{
"package_name": p.package_name,
"package_version": p.package_version,
"fixed_version": p.fixed_version,
"source": p.source,
"has_fix": p.has_fix,
}
for p in pkgs
]
has_fix_any = any(p.has_fix for p in pkgs) if pkgs else bool(
vuln.fixed_version and vuln.fixed_version != vuln.package_version
)
return {
"id": vuln.id,
"cve_id": vuln.cve_id,
"asset_id": vuln.asset_id,
"asset_hostname": vuln.asset.hostname if vuln.asset else "Unknown",
"cvss_score": vuln.cvss_score,
"severity": vuln.severity,
"status": vuln.status,
"title": vuln.title,
"package_name": vuln.package_name,
"package_version": vuln.package_version,
"fixed_version": vuln.fixed_version,
"exploitable": vuln.exploitable,
"exploit_available": vuln.exploit_available,
"detected_at": vuln.detected_at,
"priority_score": breakdown["total"],
"priority_breakdown": breakdown,
"cpr_score": breakdown.get("cpr"),
"epss_score": vuln.epss_score,
"epss_percentile": vuln.epss_percentile,
"kev_listed": vuln.kev_listed,
"kev_ransomware_use": vuln.kev_ransomware_use,
"kev_date_added": vuln.kev_date_added,
"euvd_listed": vuln.euvd_listed,
"euvd_critical": vuln.euvd_critical,
"euvd_date_added": vuln.euvd_date_added,
"euvd_id": vuln.euvd_id,
"enrichment_sources": vuln.enrichment_sources,
"enrichment_updated_at": vuln.enrichment_updated_at,
# Multi-scanner attribution (Nessus integration)
"sources": vuln.source_list,
"cross_confirmed": vuln.cross_confirmed,
"first_detected_by": vuln.first_detected_by,
"nessus_plugin_id": vuln.nessus_plugin_id,
"nessus_vpr_score": vuln.nessus_vpr_score,
# SSVC decision points from CISA Vulnrichment.
# exploitation_status was previously missing from _build_vuln_response
# — Pydantic schema declared the field but the dict never set it,
# so the frontend POC/ACTIVE/WIDESPREAD pill never fired even when
# the score breakdown showed +2.0 from Exploit (ssvc).
"exploitation_status": vuln.exploitation_status,
"exploitation_source": vuln.exploitation_source,
"ssvc_technical_impact": vuln.ssvc_technical_impact,
"ssvc_automatable": vuln.ssvc_automatable,
"is_pseudo_cve": vuln.is_pseudo_cve,
"is_eol_finding": vuln.is_eol_finding,
"assigned_user_id": vuln.assigned_user_id,
"assigned_user_name": vuln.assigned_user.username if vuln.assigned_user else None,
"assigned_group_id": vuln.assigned_group_id,
"assigned_group_name": vuln.group.name if vuln.group else None,
"deferred_until": vuln.deferred_until,
"defer_reason": vuln.defer_reason,
"notification_suppressed": vuln.notification_suppressed,
"packages": packages_payload,
"has_fix_any": has_fix_any,
# Fields for DetailResponse (will be None if not detail)
"description": getattr(vuln, 'description', None),
"cvss_vector": getattr(vuln, 'cvss_vector', None),
"exploit_maturity": getattr(vuln, 'exploit_maturity', None),
"published_date": getattr(vuln, 'published_date', None),
"patched_at": getattr(vuln, 'patched_at', None),
"references": getattr(vuln, 'references', None),
"cwe_id": getattr(vuln, 'cwe_id', None),
}
# ============================================
# Endpoints
# ============================================
@router.post("/bulk-update")
async def bulk_update_vulnerabilities(
update_data: BulkVulnerabilityUpdateRequest,
db: Session = Depends(get_db),
current_user: User = Depends(RequireEditor)
):
"""Updates multiple vulnerabilities at once (assignment, status)"""
vulnerabilities = db.query(Vulnerability).filter(Vulnerability.id.in_(update_data.vulnerability_ids)).all()
if not vulnerabilities:
raise HTTPException(status_code=404, detail="No vulnerabilities found")
update_fields = {}
if update_data.assigned_user_id is not None:
update_fields["assigned_user_id"] = update_data.assigned_user_id if update_data.assigned_user_id != -1 else None
# Clear group when assigning user (mutually exclusive)
update_fields["assigned_group_id"] = None
if update_data.assigned_group_id is not None:
update_fields["assigned_group_id"] = update_data.assigned_group_id if update_data.assigned_group_id != -1 else None
# Clear user when assigning group (mutually exclusive)
update_fields["assigned_user_id"] = None
if update_data.status is not None:
update_fields["status"] = update_data.status
if not update_fields:
raise HTTPException(status_code=400, detail="No update fields specified")
count = 0
now = datetime.now()
# Status change in this bulk action? Used to drive per-vuln audit
# entries so each row carries who/when/source even in bulk flow.
new_status = update_fields.get("status")
audit_reason = (
update_data.reason
or (_auto_reason(new_status, current_user.username) if new_status else None)
)
status_change_log: list[tuple[int, str, str, str]] = [] # (vuln_id, cve_id, old, new)
for vuln in vulnerabilities:
for field, value in update_fields.items():
# If setting status to PATCHED, also set patched_at logic
if field == "status" and value == VulnerabilityStatus.patched:
if vuln.status != VulnerabilityStatus.patched:
vuln.patched_at = now
if field == "status" and vuln.status != value:
old_val = vuln.status.value if hasattr(vuln.status, "value") else str(vuln.status)
new_val = value.value if hasattr(value, "value") else str(value)
status_change_log.append((vuln.id, vuln.cve_id, old_val, new_val))
setattr(vuln, field, value)
count += 1
db.commit()
# Per-vuln audit entries so each row's Change History panel shows
# WHO bulk-changed it + WHY (auto-reason when no comment supplied).
for vuln_id, cve_id, old_val, new_val in status_change_log:
try:
log_vulnerability_change(
db, current_user.id, vuln_id, old_val, new_val,
reason=audit_reason, cve_id=cve_id, source="manual_bulk",
)
except Exception as e:
logger.warning("bulk audit log failed for vuln_id=%s: %s", vuln_id, e)
# Summary audit entry kept for the admin overview page.
audit_log = AuditLog(
user_id=current_user.id,
event_type=AuditEventType.VULNERABILITY_UPDATED,
event_description=(
f"Bulk update performed for {count} vulnerabilities"
+ (f" (status → {new_status.value})" if new_status else "")
+ (f" — reason: {audit_reason}" if audit_reason else "")
)[:500],
resource_type="vulnerability",
resource_id="multiple",
timestamp=datetime.now()
)
db.add(audit_log)
db.commit()
return {"message": f"{count} vulnerabilities successfully updated", "count": count}
@router.get("")
async def list_vulnerabilities(
severity: Optional[VulnerabilitySeverity] = Query(None, description="Filter nach Severity"),
status: Optional[str] = Query(None, description="Filter nach Status (enum value or 'all')"),
exploitable: Optional[bool] = Query(None, description="Nur exploitable CVEs"),
kev_only: Optional[bool] = Query(None, description="Nur CISA-KEV-gelistete CVEs"),
euvd_only: Optional[bool] = Query(None, description="Nur ENISA EUVD-gelistete CVEs"),
eu_critical: Optional[bool] = Query(None, description="Nur ENISA-Critical-CVEs"),
in_any_catalog: Optional[bool] = Query(None, description="CVE in mind. einem Catalog (KEV oder EUVD)"),
in_both_catalogs: Optional[bool] = Query(None, description="CVE in BEIDEN Catalogs (KEV und EUVD)"),
epss_min: Optional[float] = Query(None, description="Minimaler EPSS-Score (0.0-1.0)"),
source: Optional[str] = Query(None, description="Filter by scanner source: wazuh, nessus, manual"),
cross_confirmed: Optional[bool] = Query(None, description="Only CVEs reported by 2+ scanners"),
asset_id: Optional[int] = Query(None, description="Filter nach Asset"),
search: Optional[str] = Query(None, description="Suche in CVE-ID, Package, Title"),
sort_by: str = Query("priority", description="Sortierung: priority, cvss, detected_at"),
sort_order: str = Query("desc", description="Reihenfolge: asc, desc"),
limit: int = Query(100, le=1000),
offset: int = Query(0),
db: Session = Depends(get_db),
current_user: User = Depends(get_current_user)
):
"""
Liste aller Vulnerabilities mit Filterung und Sortierung
Sortier-Optionen:
- priority: Nach berechnetem Priority-Score (CVSS + Exploitability + Age)
- cvss: Nach CVSS-Score
- detected_at: Nach Erkennungsdatum
"""
# Base Query mit OUTER JOIN zu Assets (damit Vulns auch ohne Asset-Match angezeigt werden)
query = db.query(Vulnerability).outerjoin(Asset)
# Filter anwenden
if severity:
query = query.filter(Vulnerability.severity == severity)
# Status filter — supports literal enum values + two convenience groups.
# Default (no value) = active states only (open + pending_verification +
# patch_failed). 'all' returns every status. 'closed' returns operator-
# closed states (false_positive + accepted_risk + deferred + patched).
_ACTIVE_STATES = (
VulnerabilityStatus.open,
VulnerabilityStatus.pending_verification,
VulnerabilityStatus.patch_failed,
)
_CLOSED_STATES = (
VulnerabilityStatus.patched,
VulnerabilityStatus.false_positive,
VulnerabilityStatus.accepted_risk,
VulnerabilityStatus.deferred,
)
if status == 'all':
pass # no filter
elif status == 'active':
query = query.filter(Vulnerability.status.in_(_ACTIVE_STATES))
elif status == 'closed':
query = query.filter(Vulnerability.status.in_(_CLOSED_STATES))
elif status:
# literal enum value (open / patched / false_positive / ...)
query = query.filter(Vulnerability.status == status)
else:
# Default — active states only (SOC workflow default)
query = query.filter(Vulnerability.status.in_(_ACTIVE_STATES))
if exploitable is not None:
query = query.filter(Vulnerability.exploitable == exploitable)
if kev_only:
query = query.filter(Vulnerability.kev_listed == True)
if euvd_only:
query = query.filter(Vulnerability.euvd_listed == True)
if eu_critical:
query = query.filter(Vulnerability.euvd_critical == True)
if in_any_catalog:
query = query.filter(
(Vulnerability.kev_listed == True) | (Vulnerability.euvd_listed == True)
)
if in_both_catalogs:
query = query.filter(
Vulnerability.kev_listed == True,
Vulnerability.euvd_listed == True,
)
if epss_min is not None:
query = query.filter(Vulnerability.epss_score >= epss_min)
# Source filter (sources column is a JSON list stored as text).
# `wazuh_only` would be too brittle, use string contains.
if source:
src = source.strip().lower()
if src in {"wazuh", "nessus", "manual"}:
query = query.filter(Vulnerability.sources.contains(f'"{src}"'))
if cross_confirmed:
# Pragmatic: rows with at least one comma in the JSON list have >=2 sources.
# `["wazuh"]` has no comma, `["wazuh","nessus"]` does.
query = query.filter(Vulnerability.sources.contains(','))
if asset_id:
query = query.filter(Vulnerability.asset_id == asset_id)
if search:
import re
# Parse search terms, respecting quoted strings
# Example: '!"Mozilla Thunderbird" kernel' -> ['!"Mozilla Thunderbird"', 'kernel']
term_pattern = re.compile(r'!?"[^"]+"|[^\s]+')
terms = term_pattern.findall(search.strip())
for term in terms:
is_negation = term.startswith('!')
clean_term = term[1:] if is_negation else term
# Remove surrounding quotes if present
if clean_term.startswith('"') and clean_term.endswith('"'):
clean_term = clean_term[1:-1]
if not clean_term:
continue
search_pattern = f"%{clean_term}%"
if is_negation:
# Negation (Exclusion)
query = query.filter(
~((Vulnerability.cve_id.ilike(search_pattern)) |
(Vulnerability.package_name.ilike(search_pattern)) |
(Vulnerability.title.ilike(search_pattern)))
)
else:
# Normal Search (Inclusion)
query = query.filter(
(Vulnerability.cve_id.ilike(search_pattern)) |
(Vulnerability.package_name.ilike(search_pattern)) |
(Vulnerability.title.ilike(search_pattern))
)
# Get total count before pagination
total_count = query.count()
# Sortierung
# Supported sort_by values map to SQL columns.
# `priority` is special (needs Python-side breakdown), `cpr` is computed,
# `published_date` and `cve_id` fall through to dedicated branches below.
sort_map = {
"cvss": Vulnerability.cvss_score,
"cvss_score": Vulnerability.cvss_score,
"detected_at": Vulnerability.detected_at,
"published_date": Vulnerability.published_date,
"updated_at": Vulnerability.updated_at, # DB-write timestamp — "recently changed risk"
"epss": Vulnerability.epss_score,
"epss_score": Vulnerability.epss_score,
"kev": Vulnerability.kev_listed,
"kev_listed": Vulnerability.kev_listed,
"euvd": Vulnerability.euvd_listed,
"euvd_listed": Vulnerability.euvd_listed,
"cve_id": Vulnerability.cve_id,
"severity": Vulnerability.severity,
"status": Vulnerability.status,
# Source: groups rows by which scanner first reported the finding.
# Cross-confirmed rows fall into the bucket of whichever saw it first.
"source": Vulnerability.first_detected_by,
"first_detected_by": Vulnerability.first_detected_by,
}
order_fn = desc if sort_order == "desc" else asc
def _python_sort_and_build(vulns, key_fn):
ordered = sorted(vulns, key=key_fn, reverse=(sort_order == "desc"))
return [_build_vuln_response(v) for v in ordered]
if sort_by == "priority":
# Sort by materialised priority_score (migration 024). NULLs go
# last on desc, first on asc. Sync/enrich/override paths keep
# the column current via Vulnerability.refresh_scores().
if sort_order == "desc":
query = query.order_by(
Vulnerability.priority_score.is_(None),
desc(Vulnerability.priority_score),
desc(Vulnerability.id),
)
else:
query = query.order_by(
asc(Vulnerability.priority_score),
asc(Vulnerability.id),
)
vulnerabilities = query.offset(offset).limit(limit).all()
return {"items": [_build_vuln_response(v) for v in vulnerabilities],
"total": total_count}
if sort_by == "cpr":
# Sort by materialised cpr_score (migration 024). Old SQL proxy
# `cvss * epss` was wrong vs the real `CVSS×10×0.6 + EPSS%×0.4`
# formula and only sorted the current page anyway.
if sort_order == "desc":
query = query.order_by(
Vulnerability.cpr_score.is_(None),
desc(Vulnerability.cpr_score),
desc(Vulnerability.id),
)
else:
query = query.order_by(
asc(Vulnerability.cpr_score),
asc(Vulnerability.id),
)
vulnerabilities = query.offset(offset).limit(limit).all()
return {"items": [_build_vuln_response(v) for v in vulnerabilities],
"total": total_count}
if sort_by == "published_date":
# CVE-IDs are NOT chronological — CISA/MITRE assign them in batches.
# Order by official published_date with a detected_at fallback so
# CVEs we have not yet enriched still get a reasonable position.
from sqlalchemy import func as sa_func
ordering_expr = sa_func.coalesce(
Vulnerability.published_date, Vulnerability.detected_at
)
if sort_order == "desc":
query = query.order_by(
ordering_expr.is_(None), desc(ordering_expr), desc(Vulnerability.id)
)
else:
query = query.order_by(asc(ordering_expr), asc(Vulnerability.id))
vulnerabilities = query.offset(offset).limit(limit).all()
return {"items": [_build_vuln_response(v) for v in vulnerabilities],
"total": total_count}
if sort_by == "cve_id":
# Natural-numeric sort: lex on "CVE-2026-8401" sorts BEFORE
# "CVE-2026-35440" because "8" > "3" character-wise. Cast the
# numeric suffix to int so 35440 wins.
# Pseudo-CVEs (NESSUS-PLUGIN-*) fall back to lex at the end.
from sqlalchemy import func as sa_func, case
# Postgres: split_part(cve_id, '-', 2)::int as year, ('', 3)::int as num
# Pseudo-CVEs ("NESSUS-PLUGIN-123456") have non-numeric year segment;
# NULLIF + regexp catches the cast safely.
year_part = sa_func.nullif(sa_func.split_part(Vulnerability.cve_id, '-', 2), '')
num_part = sa_func.nullif(sa_func.split_part(Vulnerability.cve_id, '-', 3), '')
# Cast only when both segments are pure digits (defensive vs pseudo-CVEs)
year_int = case(
(year_part.op('~')('^[0-9]+$'), sa_func.cast(year_part, __import__('sqlalchemy').Integer)),
else_=None,
)
num_int = case(
(num_part.op('~')('^[0-9]+$'), sa_func.cast(num_part, __import__('sqlalchemy').Integer)),
else_=None,
)
if sort_order == "desc":
query = query.order_by(
year_int.is_(None), # pseudo-CVEs to the end
desc(year_int),
desc(num_int),
desc(Vulnerability.cve_id), # tiebreak for pseudo
)
else:
query = query.order_by(asc(year_int), asc(num_int), asc(Vulnerability.cve_id))
vulnerabilities = query.offset(offset).limit(limit).all()
return {"items": [_build_vuln_response(v) for v in vulnerabilities],
"total": total_count}
sort_column = sort_map.get(sort_by, Vulnerability.cvss_score)
# NULL handling: put NULLs last on desc, first on asc (sensible UX)
if sort_order == "desc":
query = query.order_by(sort_column.is_(None), desc(sort_column))
else:
query = query.order_by(asc(sort_column))
vulnerabilities = query.offset(offset).limit(limit).all()
results = [_build_vuln_response(v) for v in vulnerabilities]
return {"items": results, "total": total_count}
@router.get("/ai-prioritization")
def get_ai_prioritization(
limit: int = Query(20, le=50),
severity: Optional[VulnerabilitySeverity] = Query(None, description="Nach Severity filtern"),
db: Session = Depends(get_db),
current_user: User = Depends(RequireEditor),
ai_client: InfomaniakAIClient = Depends(get_infomaniak_ai_client)
):
"""
Generiert KI-basierte Patch-Empfehlungen
"""
ai_service = AIAnalysisService(db, ai_client)
try:
recommendations = ai_service.get_priority_recommendations(limit=limit, severity=severity)
# Save Report
if "recommendations" in recommendations and recommendations["recommendations"]:
# Extract data safely
strategy = recommendations.get("global_strategy", "No strategy provided.")
recs_json = json.dumps(recommendations.get("recommendations", []))
report = AIReport(
created_by_id=current_user.id,
global_strategy=strategy,
recommendations=recs_json
)
db.add(report)
db.commit()
return recommendations
except Exception as e:
logger.error(f"AI prioritization failed: {e}")
raise HTTPException(
status_code=500,
detail="AI prioritization failed. Check server logs for details."
)
@router.get("/ai-history")
async def get_ai_history(
limit: int = 10,
db: Session = Depends(get_db),
current_user: User = Depends(get_current_user)
):
"""
Liefert die letzten AI-Scan-Berichte zurück
"""
reports = db.query(AIReport).order_by(desc(AIReport.created_at)).limit(limit).all()
result = []
for r in reports:
try:
recs = json.loads(r.recommendations)
except (json.JSONDecodeError, TypeError):
recs = []
result.append({
"id": r.id,
"created_at": r.created_at,
"global_strategy": r.global_strategy,
"recommendations_count": len(recs),
"recommendations": recs # Optional: full data
})
return result
@router.get("/reports/dashboard", response_model=DashboardStatsResponse)
async def get_dashboard_statistics(
db: Session = Depends(get_db),
current_user: User = Depends(get_current_user)
):
"""
Dashboard-Statistiken für Frontend
Liefert aggregierte Daten:
- Anzahl Vulnerabilities pro Severity
- Offene vs. Geschlossene
- Exploitable-Count
- Durchschnittlicher CVSS-Score
- Betroffene Assets
- Älteste offene Vulnerability
"""
# Base Query: Nur offene Vulnerabilities
open_vulns = db.query(Vulnerability).filter(
Vulnerability.status == VulnerabilityStatus.open
)
# Counts pro Severity
critical_count = open_vulns.filter(
Vulnerability.severity == VulnerabilitySeverity.critical
).count()
high_count = open_vulns.filter(
Vulnerability.severity == VulnerabilitySeverity.high
).count()
medium_count = open_vulns.filter(
Vulnerability.severity == VulnerabilitySeverity.medium
).count()
low_count = open_vulns.filter(
Vulnerability.severity == VulnerabilitySeverity.low
).count()
# Exploitable Count
exploitable_count = open_vulns.filter(
Vulnerability.exploit_available == True
).count()
# Durchschnittlicher CVSS-Score
avg_cvss = db.query(func.avg(Vulnerability.cvss_score)).filter(
Vulnerability.status == VulnerabilityStatus.open
).scalar() or 0.0
# Betroffene Assets
affected_assets = db.query(func.count(func.distinct(Vulnerability.asset_id))).filter(
Vulnerability.status == VulnerabilityStatus.open
).scalar() or 0
# Älteste Vulnerability
oldest_vuln = open_vulns.order_by(asc(Vulnerability.detected_at)).first()
oldest_days = 0
if oldest_vuln:
oldest_days = (datetime.now() - oldest_vuln.detected_at).days
# Verlauf der letzten 7 Tage (Erkennungsdatum)
severity_history = []
from datetime import timedelta, time, datetime as dt_class
now = datetime.now()
for i in range(6, -1, -1):
day_date = (now - timedelta(days=i)).date()
start_of_day = dt_class.combine(day_date, time.min)
end_of_day = dt_class.combine(day_date, time.max)
count = db.query(Vulnerability).filter(
Vulnerability.detected_at >= start_of_day,
Vulnerability.detected_at <= end_of_day,
Vulnerability.status == VulnerabilityStatus.open
).count()
severity_history.append({
"day": day_date.strftime("%a"),
"count": count
})
# Gesamt-Zahlen
total_vulns = db.query(Vulnerability).count()
open_vulns_count = open_vulns.count()
total_assets = db.query(Asset).count()
# Assets, die mindestens einmal gescannt wurden (last_scan is not null)
scanned_assets = db.query(Asset).filter(Asset.last_scan != None).count()
return {
"total_vulnerabilities": total_vulns,
"open_vulnerabilities": open_vulns_count,
"critical_count": critical_count,
"high_count": high_count,
"medium_count": medium_count,
"low_count": low_count,
"exploitable_count": exploitable_count,
"avg_cvss_score": round(avg_cvss, 2),
"affected_assets": affected_assets,
"total_assets": total_assets,
"scanned_assets": scanned_assets,
"severity_history": severity_history,
"oldest_vulnerability_days": oldest_days
}
@router.get("/{vuln_id}", response_model=VulnerabilityDetailResponse)
async def get_vulnerability_detail(
vuln_id: int,
db: Session = Depends(get_db),
current_user: User = Depends(get_current_user)
):
"""
Detail-Ansicht einer Vulnerability inkl. KI-Analyse (falls vorhanden)
"""
vuln = db.query(Vulnerability).filter(Vulnerability.id == vuln_id).first()
if not vuln:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail="Vulnerability not found"
)
# Hole KI-Analyse falls vorhanden
ai_analysis = None
if vuln.ai_analysis:
ai_analysis = {
"analysis_text": vuln.ai_analysis.analysis_text,
"threat_level": vuln.ai_analysis.threat_level,
"exploits_found": vuln.ai_analysis.exploits_found,
"workarounds": vuln.ai_analysis.workarounds,
"remediation_steps": vuln.ai_analysis.remediation_steps,
"confidence_score": vuln.ai_analysis.confidence_score,
"analysis_timestamp": vuln.ai_analysis.analysis_timestamp
}
resp = _build_vuln_response(vuln)
resp["ai_analysis"] = ai_analysis
return resp
@router.patch("/{vuln_id}", response_model=VulnerabilityResponse)
async def update_vulnerability_status(
vuln_id: int,
update_data: VulnerabilityUpdateRequest,
background_tasks: BackgroundTasks,
db: Session = Depends(get_db),
current_user: User = Depends(RequireEditor) # Editor/Admin-Berechtigung erforderlich
):
"""
Ändert Status einer Vulnerability
Patch-Verifizierungs-Workflow:
1. User markiert CVE als "patched"
2. Status wird auf "pending_verification" gesetzt
3. Wazuh Syscollector-Scan wird getriggert (Background-Task)
4. Nach Scan-Completion: Automatische Verifizierung
"""
try:
vuln = db.query(Vulnerability).filter(Vulnerability.id == vuln_id).first()
if not vuln:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail="Vulnerability not found"
)
old_status = vuln.status
new_status = update_data.status
# Status-Änderung
vuln.status = new_status
# Persist reason in the matching column when it makes sense.
# `reason` is the new status-agnostic field; `defer_reason` is
# kept for the deferred-status legacy UI.
if update_data.defer_reason is not None:
vuln.defer_reason = update_data.defer_reason
if update_data.deferred_until is not None:
vuln.deferred_until = update_data.deferred_until
# The audit-trail reason — accept either field for back-compat.
# If the operator left both blank, auto-stamp "Manually set to X
# by Y on Z" so the audit trail clearly tags this as a manual
# action (vs scanner auto-patch) even without a user comment.
audit_reason = (
update_data.reason
or update_data.defer_reason
or _auto_reason(new_status, current_user.username)
)
# Wenn als "patched" markiert: Trigger Rescan
if new_status == VulnerabilityStatus.patched:
# Check if asset has wazuh agent
if vuln.asset and vuln.asset.wazuh_agent_id:
vuln.status = VulnerabilityStatus.pending_verification
vuln.patched_at = datetime.now()
# Background-Task für Rescan
background_tasks.add_task(
verify_patch_with_rescan,
vuln_id,
vuln.asset.wazuh_agent_id
)
else:
# Kein Agent -> Direkt als Patched markieren
vuln.patched_at = datetime.now()
db.commit()
db.refresh(vuln)
# Audit-Log — includes reason + cve_id for the revisionssicher view
log_vulnerability_change(
db, current_user.id, vuln_id, old_status, vuln.status,
reason=audit_reason, cve_id=vuln.cve_id, source="manual",
)
return _build_vuln_response(vuln)
except HTTPException:
raise
except Exception as e:
logger.error(f"Error updating vulnerability {vuln_id}: {e}", exc_info=True)
raise HTTPException(
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
detail=f"Internal Server Error: {str(e)}"
)
def verify_patch_with_rescan(
vuln_id: int,
agent_id: str
):
"""
Snychroner Background-Task für Patch-Verifizierung (läuft in eigenem Thread)
1. Triggert Syscollector-Scan
2. Wartet auf Completion (max. 10 Min)
3. Re-fetcht Vulnerabilities
4. Prüft ob CVE noch vorhanden
5. Aktualisiert Status
"""
from app.database import get_db_context
from app.models.setting import Setting
import json
with get_db_context() as db:
try:
from app.auth.setting_crypto import read_setting_value
raw_wazuh = read_setting_value(db, "wazuh_config")
if not raw_wazuh:
logger.error("Patch verification aborted: Wazuh configuration missing")
return
config = json.loads(raw_wazuh)
wazuh = WazuhClient(
base_url=config.get("api_url"),
username=config.get("username"),
password=config.get("password"),
indexer_url=config.get("indexer_url"),
indexer_username=config.get("indexer_username"),
indexer_password=config.get("indexer_password"),
verify_ssl=config.get("verify_ssl", True)
)
# 1. Trigger Rescan
try:
wazuh.trigger_syscollector_scan(agent_id)
except Exception as e:
logger.error(f"Could not trigger Syscollector scan for agent {agent_id}: {e}")
# Wir machen trotzdem weiter, evtl. lief gerade schon einer
# 2. Warte auf Scan-Completion (blockiert nur diesen Thread, nicht den Server)
scan_completed = wazuh.wait_for_scan_completion(
agent_id,
timeout=600, # 10 Min
check_interval=30
)
if not scan_completed:
# Timeout: Markiere als Patch-Failed
vuln = db.query(Vulnerability).filter(Vulnerability.id == vuln_id).first()
if vuln:
vuln.status = VulnerabilityStatus.patch_failed
db.commit()
return
# 3. Re-fetch Vulnerabilities
current_vulns = wazuh.get_vulnerabilities(agent_id)
vuln = db.query(Vulnerability).filter(Vulnerability.id == vuln_id).first()
if not vuln:
return
# 4. Prüfe ob CVE noch in Wazuh-Daten
cve_still_exists = any(v.get("cve") == vuln.cve_id for v in current_vulns)
if cve_still_exists:
# Patch fehlgeschlagen
vuln.status = VulnerabilityStatus.patch_failed
else:
# Patch erfolgreich verifiziert
vuln.status = VulnerabilityStatus.patched
db.commit()
except Exception as e:
logger.error(f"Error in verify_patch_with_rescan: {e}")
# Bei Fehler: Markiere als Patch-Failed
vuln = db.query(Vulnerability).filter(Vulnerability.id == vuln_id).first()
if vuln:
vuln.status = VulnerabilityStatus.patch_failed
db.commit()
@router.post("/{vuln_id}/analyze")
def analyze_vulnerability_with_ai(
vuln_id: int,
force_refresh: bool = Query(False, description="Ignore cache"),
db: Session = Depends(get_db),
current_user: User = Depends(RequireEditor),
ai_client: InfomaniakAIClient = Depends(get_infomaniak_ai_client)
):
"""
Triggert KI-Analyse für eine Vulnerability
Nutzt Infomaniak AI API mit Web-Search für:
- Exploit-Detection
- Threat-Intelligence
- Workarounds
- Remediation-Steps
"""
vuln = db.query(Vulnerability).filter(Vulnerability.id == vuln_id).first()
if not vuln:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail="Vulnerability not found"
)
# Service-Layer für DB-Integration
ai_service = AIAnalysisService(db, ai_client)
try:
analysis = ai_service.get_or_create_analysis(vuln_id, force_refresh)
# Audit-Log
audit_log = AuditLog(
user_id=current_user.id,
event_type=AuditEventType.AI_ANALYSIS_REQUESTED,
event_description=f"AI analysis requested for {vuln.cve_id}",
resource_type="vulnerability",
resource_id=str(vuln_id),
timestamp=datetime.now()
)
db.add(audit_log)
db.commit()
return {
"analysis_text": analysis.analysis_text,
"threat_level": analysis.threat_level,
"exploits_found": analysis.exploits_found,
"workarounds": analysis.workarounds,
"remediation_steps": analysis.remediation_steps,
"threat_intel_sources": analysis.threat_intel_sources,
"confidence_score": analysis.confidence_score,
"model_version": analysis.model_version,
"analysis_timestamp": analysis.analysis_timestamp,
"cached": not force_refresh
}
except Exception as e:
logger.error(f"AI analysis failed: {e}")
raise HTTPException(
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
detail="AI analysis failed. Check server logs for details."
)
@router.post("/{vuln_id}/enrich")
async def enrich_single_vulnerability(
vuln_id: int,
db: Session = Depends(get_db),
current_user: User = Depends(RequireEditor)
):
"""Trigger EPSS + CISA KEV enrichment for one vulnerability."""
try:
stats = enrich_vulnerability_by_id(db, vuln_id)
except EnrichmentError as e:
raise HTTPException(status_code=404, detail=str(e))
vuln = db.query(Vulnerability).filter(Vulnerability.id == vuln_id).first()
audit_log = AuditLog(
user_id=current_user.id,
event_type=AuditEventType.VULNERABILITY_UPDATED,
event_description=f"Threat intel enrichment for {vuln.cve_id if vuln else vuln_id}",
resource_type="vulnerability",
resource_id=str(vuln_id),
timestamp=datetime.now()
)
db.add(audit_log)
db.commit()
return {"stats": stats, "vulnerability": _build_vuln_response(vuln) if vuln else None}
class BulkEnrichRequest(BaseModel):
vulnerability_ids: Optional[List[int]] = None
only_open: bool = True
@router.post("/enrich/bulk")
async def bulk_enrich_vulnerabilities(
payload: BulkEnrichRequest,
db: Session = Depends(get_db),
current_user: User = Depends(RequireEditor)
):
"""
Bulk enrich vulnerabilities with EPSS + CISA KEV.
- If `vulnerability_ids` is provided, only those are enriched.
- Otherwise enriches every open vulnerability (or all if only_open=false).
"""
if payload.vulnerability_ids:
vulns = db.query(Vulnerability).filter(
Vulnerability.id.in_(payload.vulnerability_ids)
).all()
else:
q = db.query(Vulnerability)
if payload.only_open:
q = q.filter(Vulnerability.status == VulnerabilityStatus.open)
vulns = q.all()
if not vulns:
return {"stats": {"epss_updated": 0, "kev_marked": 0, "kev_cleared": 0, "total": 0}}
stats = enrich_vulnerabilities(db, vulns)
audit_log = AuditLog(
user_id=current_user.id,
event_type=AuditEventType.VULNERABILITY_UPDATED,
event_description=f"Bulk threat intel enrichment for {stats['total']} CVEs",
resource_type="vulnerability",
resource_id="multiple",
timestamp=datetime.now()
)
db.add(audit_log)
db.commit()
return {"stats": stats}
@router.post("/enrich/kev/refresh")
async def refresh_kev_catalog(
db: Session = Depends(get_db),
current_user: User = Depends(RequireEditor)
):
"""Force-refresh the cached CISA KEV catalog from CISA feed."""
try:
kev_map = fetch_kev_catalog(db, force_refresh=True)
except EnrichmentError as e:
raise HTTPException(status_code=502, detail=str(e))
return {"kev_entries": len(kev_map), "refreshed_at": datetime.now().isoformat()}
@router.patch("/{vuln_id}/assign")
async def assign_vulnerability(
vuln_id: int,
assign_data: AssignRequest,
db: Session = Depends(get_db),
current_user: User = Depends(RequireEditor)
):
"""Assigns a user to a vulnerability"""
vuln = db.query(Vulnerability).filter(Vulnerability.id == vuln_id).first()
if not vuln:
raise HTTPException(status_code=404, detail="Vulnerability not found")
if assign_data.user_id is not None:
user = db.query(User).filter(User.id == assign_data.user_id).first()
if not user:
raise HTTPException(status_code=404, detail="User not found")
if assign_data.group_id is not None:
from app.models.group import Group
group = db.query(Group).filter(Group.id == assign_data.group_id).first()
if not group:
raise HTTPException(status_code=404, detail="Group not found")
vuln.assigned_user_id = assign_data.user_id
vuln.assigned_group_id = assign_data.group_id
db.commit()
db.refresh(vuln)
return _build_vuln_response(vuln)
@router.patch("/{vuln_id}/defer")
async def defer_vulnerability(
vuln_id: int,
defer_data: DeferRequest,
db: Session = Depends(get_db),
current_user: User = Depends(RequireEditor)
):
"""Defers a vulnerability with a target date and reason"""
vuln = db.query(Vulnerability).filter(Vulnerability.id == vuln_id).first()
if not vuln:
raise HTTPException(status_code=404, detail="Vulnerability not found")
old_status = vuln.status
vuln.status = VulnerabilityStatus.deferred
vuln.deferred_until = defer_data.deferred_until
vuln.defer_reason = defer_data.reason
vuln.notification_suppressed = True
db.commit()
db.refresh(vuln)
log_vulnerability_change(db, current_user.id, vuln_id, old_status, vuln.status)
return _build_vuln_response(vuln)
@router.patch("/{vuln_id}/undefer")
async def undefer_vulnerability(
vuln_id: int,
db: Session = Depends(get_db),
current_user: User = Depends(RequireEditor)
):
"""Removes deferral, resets to OPEN"""
vuln = db.query(Vulnerability).filter(Vulnerability.id == vuln_id).first()
if not vuln:
raise HTTPException(status_code=404, detail="Vulnerability nicht gefunden")
old_status = vuln.status
vuln.status = VulnerabilityStatus.open
vuln.deferred_until = None
vuln.defer_reason = None
vuln.notification_suppressed = False
db.commit()
db.refresh(vuln)
log_vulnerability_change(db, current_user.id, vuln_id, old_status, vuln.status)
return _build_vuln_response(vuln)
@router.patch("/{vuln_id}/suppress-notifications")
async def toggle_notification_suppression(
vuln_id: int,
db: Session = Depends(get_db),
current_user: User = Depends(RequireEditor)
):
"""Toggles notification suppression for a vulnerability"""
vuln = db.query(Vulnerability).filter(Vulnerability.id == vuln_id).first()
if not vuln:
raise HTTPException(status_code=404, detail="Vulnerability nicht gefunden")
vuln.notification_suppressed = not vuln.notification_suppressed
db.commit()
db.refresh(vuln)
return _build_vuln_response(vuln)
class FalsePositiveRequest(BaseModel):
reason: Optional[str] = None
@router.patch("/{vuln_id}/false-positive")
async def mark_false_positive(
vuln_id: int,
payload: FalsePositiveRequest,
db: Session = Depends(get_db),
current_user: User = Depends(RequireEditor),
):
"""
Mark a vulnerability as false positive. Bookmark used after manual
review (e.g. confirming a Nessus finding is a false alarm). Sets
status=false_positive and suppresses further notifications.
"""
vuln = db.query(Vulnerability).filter(Vulnerability.id == vuln_id).first()
if not vuln:
raise HTTPException(status_code=404, detail="Vulnerability not found")
old_status = vuln.status
vuln.status = VulnerabilityStatus.false_positive
vuln.notification_suppressed = True
if payload.reason:
vuln.defer_reason = payload.reason # reuse existing field for context
db.commit()
db.refresh(vuln)
log_vulnerability_change(db, current_user.id, vuln_id, old_status, vuln.status)
return _build_vuln_response(vuln)
@router.patch("/{vuln_id}/unmark-false-positive")
async def unmark_false_positive(
vuln_id: int,
db: Session = Depends(get_db),
current_user: User = Depends(RequireEditor),
):
"""Revert a false-positive marking back to open."""
vuln = db.query(Vulnerability).filter(Vulnerability.id == vuln_id).first()
if not vuln:
raise HTTPException(status_code=404, detail="Vulnerability not found")
if vuln.status != VulnerabilityStatus.false_positive:
raise HTTPException(status_code=400, detail="Not currently marked as false positive")
old_status = vuln.status
vuln.status = VulnerabilityStatus.open
vuln.notification_suppressed = False
vuln.defer_reason = None
db.commit()
db.refresh(vuln)
log_vulnerability_change(db, current_user.id, vuln_id, old_status, vuln.status)
return _build_vuln_response(vuln)
@router.post("/sync/wazuh")
async def sync_vulnerabilities_from_wazuh(
db: Session = Depends(get_db),
current_user: User = Depends(RequireEditor)
):
"""
Synchronizes vulnerabilities from Wazuh (synchronous)
Executes sync immediately and returns statistics.
"""
try:
result = run_wazuh_vulnerability_sync(db)
return result
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))
def run_wazuh_vulnerability_sync(db: Session) -> dict:
"""Synchronous Wazuh vulnerability sync with statistics return"""
# 1. Get Wazuh Config from DB (transparently decrypted)
from app.auth.setting_crypto import read_setting_value
raw_wazuh = read_setting_value(db, "wazuh_config")
if not raw_wazuh:
raise HTTPException(status_code=400, detail="Wazuh configuration not found. Please configure Wazuh in Settings.")
try:
config = json.loads(raw_wazuh)
api_url = config.get("api_url")
username = config.get("username")
password = config.get("password")
indexer_url = config.get("indexer_url")
indexer_username = config.get("indexer_username")
indexer_password = config.get("indexer_password")
except json.JSONDecodeError:
raise HTTPException(status_code=400, detail="Invalid Wazuh configuration format.")
if not all([api_url, username, password]):
raise HTTPException(status_code=400, detail="Incomplete Wazuh configuration. Please configure API URL, username and password.")
wazuh = WazuhClient(
base_url=api_url,
username=username,
password=password,
indexer_url=indexer_url,
indexer_username=indexer_username,
indexer_password=indexer_password,
verify_ssl=config.get("verify_ssl", False)
)
stats = {
"agents_synced": 0,
"vulns_created": 0,
"vulns_updated": 0,
"vulns_patched": 0,
"total_vulns_from_wazuh": 0
}
# IDs of vulns created during this sync — handed to the digest dispatcher
# after the full sync completes (one summary email per recipient instead
# of one email per CVE).
newly_created_vuln_ids: list[int] = []
try:
# Hole alle aktiven Agents
agents = wazuh.get_agents(status="active")
for agent in agents:
agent_id = agent["id"]
# Hole oder erstelle Asset
asset = db.query(Asset).filter(
Asset.wazuh_agent_id == agent_id
).first()
if not asset:
continue # Skip if asset not in DB
# Record a Scan-Jobs entry per agent so the manual "Sync
# Data (Wazuh)" button shows up in /scans the same way
# the Nessus sync does (commit 4bca41e). Closed out at
# the bottom of the agent loop.
from app.models.scan import Scan, ScanType, ScanStatus
scan_row = Scan(
asset_id=asset.id,
scan_type=ScanType.WAZUH,
status=ScanStatus.RUNNING,
started_at=datetime.now(),
)
db.add(scan_row)
# Get vulns from Wazuh
wazuh_vulns = wazuh.get_vulnerabilities(agent_id)
stats["total_vulns_from_wazuh"] += len(wazuh_vulns)
# 1. Deduplicate Wazuh vulns by CVE ID for this asset
unique_cves = {}
for vuln_data in wazuh_vulns:
cve_id = vuln_data.get("cve")
if not cve_id:
continue
if cve_id not in unique_cves:
unique_cves[cve_id] = {
"data": vuln_data,
"packages": set(),
"versions": set()
}
pkg_name = vuln_data.get("name")
pkg_version = vuln_data.get("version")
if pkg_name:
unique_cves[cve_id]["packages"].add(pkg_name)
if pkg_version:
unique_cves[cve_id]["versions"].add(pkg_version)
active_cves = set(unique_cves.keys())
# 2. Process unique CVEs
for cve_id, cve_info in unique_cves.items():
vuln_data = cve_info["data"]
# Merge package strings
merged_packages = ", ".join(sorted(cve_info["packages"]))
merged_versions = ", ".join(sorted(cve_info["versions"]))
# Truncate to fit DB columns
if len(merged_packages) > 250:
merged_packages = merged_packages[:247] + "..."
if len(merged_versions) > 95:
merged_versions = merged_versions[:92] + "..."
cvss_data = vuln_data.get("cvss", {})
cvss3 = cvss_data.get("cvss3", {})
score = cvss3.get("base_score")
wazuh_severity = vuln_data.get("severity")
# Check if exists
existing = db.query(Vulnerability).filter(
Vulnerability.cve_id == cve_id,
Vulnerability.asset_id == asset.id
).first()
if existing:
# Track wazuh as a source; reopen if Wazuh sees it again
# after it was marked patched.
existing.add_source("wazuh")
if existing.status == VulnerabilityStatus.patched:
existing.status = VulnerabilityStatus.open
existing.patched_at = None
elif existing.status not in (
VulnerabilityStatus.pending_verification,
VulnerabilityStatus.accepted_risk,
VulnerabilityStatus.false_positive,
VulnerabilityStatus.deferred,
):
existing.status = VulnerabilityStatus.open
# Update package info (merged)
existing.package_name = merged_packages
existing.package_version = merged_versions
# Backfill fixed_version when missing (older syncs
# didn't extract it; new syncs do).
fv = vuln_data.get("fixed_version")
if fv and not existing.fixed_version:
existing.fixed_version = fv
# Update score/severity if changed
if score and existing.cvss_score != score:
existing.cvss_score = score
existing.severity = map_severity(score, wazuh_severity)
stats["vulns_updated"] += 1
else:
# Create new — savepoint so IntegrityError (race duplicate)
# only rolls back this insert, not the whole sync.
try:
severity = map_severity(score, wazuh_severity)
vuln = Vulnerability(
cve_id=cve_id,
asset_id=asset.id,
cvss_score=score,
severity=severity,
status=VulnerabilityStatus.open,
title=vuln_data.get("title"),
package_name=merged_packages,
package_version=merged_versions,
fixed_version=vuln_data.get("fixed_version"),
detected_at=datetime.now(),
sources='["wazuh"]',
first_detected_by="wazuh",
)
with db.begin_nested():
db.add(vuln)
db.flush()
stats["vulns_created"] += 1
newly_created_vuln_ids.append(vuln.id)
except IntegrityError:
# Savepoint rolled back; outer transaction intact.
logger.debug(f"Skipping duplicate CVE {cve_id} for asset {asset.id}")
continue
# Source-aware backfill: drop "wazuh" from vulns Wazuh no longer
# reports. Only mark patched when ALL sources are gone so Nessus-
# only findings are not accidentally closed here.
#
# Safety: when Wazuh returns 0 vulns for an agent (transient API
# glitch, indexer warming up after a manager restart, etc.) the
# naive backfill below would mark every existing finding patched.
# Skip in that case — a real "everything fixed" state is
# vanishingly rare and not worth the risk of wiping the inventory.
if not active_cves and not wazuh_vulns:
logger.warning(
"Wazuh returned 0 vulns for asset %s — skipping backfill "
"to avoid mass-patching from a transient empty response.",
asset.hostname,
)
stats["agents_synced"] += 1
scan_row.status = ScanStatus.COMPLETED
scan_row.completed_at = datetime.now()
scan_row.vulnerabilities_found = 0
scan_row.error_message = "Wazuh returned 0 vulns (skipped backfill)"
continue
stale_wazuh = (
db.query(Vulnerability)
.filter(
Vulnerability.asset_id == asset.id,
Vulnerability.status == VulnerabilityStatus.open,
Vulnerability.sources.contains('"wazuh"'),
)
.all()
)
for v in stale_wazuh:
if v.cve_id in active_cves:
continue
v.remove_source("wazuh")
if not v.source_list:
old_st = v.status
v.status = VulnerabilityStatus.patched
v.patched_at = datetime.now()
stats["vulns_patched"] += 1
try:
log_vulnerability_change(
db, None, v.id, old_st, v.status,
reason=f"Wazuh full sync: agent {agent_id} no longer reports this CVE on {asset.hostname}",
cve_id=v.cve_id,
source="wazuh_sync",
)
except Exception as e:
logger.warning("audit log for wazuh full-sync auto-patch failed (vuln_id=%s): %s", v.id, e)
stats["agents_synced"] += 1
# Close out the per-agent Scan row so the Scan-Jobs page
# shows the manual sync run alongside Nessus runs.
scan_row.status = ScanStatus.COMPLETED
scan_row.completed_at = datetime.now()
scan_row.vulnerabilities_found = len(active_cves)
db.commit()
wazuh.close()
# Best-effort enrichment of newly created vulns (by ID, not a
# broad scan of all un-enriched rows which could be very slow).
if newly_created_vuln_ids:
try:
new_vulns = db.query(Vulnerability).filter(
Vulnerability.id.in_(newly_created_vuln_ids)
).all()
if new_vulns:
enrich_stats = enrich_vulnerabilities(db, new_vulns)
stats["enrichment"] = enrich_stats
except Exception as e:
logger.warning(f"Wazuh sync: enrichment of new vulns failed: {e}")
# Dispatch ONE digest email per recipient containing all new vulns
# above the configured severity threshold. Replaces the per-CVE
# inline send that used to spam SMTP at the rate of 1 mail per CVE
# per recipient. Mode override via setting `notification_mode`.
if newly_created_vuln_ids:
try:
from app.services.email_service import dispatch_new_vuln_notifications
fresh_vulns = db.query(Vulnerability).filter(
Vulnerability.id.in_(newly_created_vuln_ids)
).all()
notif_stats = dispatch_new_vuln_notifications(db, fresh_vulns)
stats["notifications"] = notif_stats
logger.info(
f"Wazuh sync notifications: {notif_stats.get('emails_sent', 0)} sent, "
f"{notif_stats.get('emails_failed', 0)} failed, "
f"{notif_stats.get('recipients', 0)} recipients"
)
except Exception as e:
logger.warning(f"Wazuh sync: new-vuln notification dispatch failed: {e}")
return {
"message": f"Synced {stats['agents_synced']} agents. Created: {stats['vulns_created']}, Updated: {stats['vulns_updated']}, Patched: {stats['vulns_patched']}",
**stats
}
except Exception as e:
db.rollback()
wazuh.close()
logger.error(f"Wazuh sync failed: {e}")
raise HTTPException(status_code=500, detail="Wazuh sync failed. Check server logs for details.")
async def sync_wazuh_vulnerabilities(db_session: Optional[Session] = None):
"""Background-Task for Wazuh-Sync (for Scheduler)"""
from app.database import SessionLocal
db = db_session if db_session is not None else SessionLocal()
try:
result = run_wazuh_vulnerability_sync(db)
logger.info(f"Scheduled Sync: {result['message']}")
except Exception as e:
logger.error(f"Scheduled Sync Error: {e}")
finally:
if db_session is None:
db.close()
def sync_agent_vulnerabilities(db: Session, wazuh: WazuhClient, agent_id: str, asset: Asset):
"""Sync vulnerabilities for a single agent from Wazuh"""
from app.models.vulnerability import Vulnerability, VulnerabilityStatus
from app.models.vulnerability_package import VulnerabilityPackage
from datetime import datetime
logger.info(f"Sync: Fetching vulnerabilities for agent {agent_id} ({asset.hostname})...")
raw_vulns = wazuh.get_vulnerabilities(agent_id)
logger.info(f"Sync: Agent {agent_id} has {len(raw_vulns)} raw vulnerability entries.")
# 1. Deduplicate by CVE ID + collect per-package detail.
# `pkg_rows` is a dict keyed by package_name so we keep one row per
# (cve, package) pair across the agent. fixed_version comes straight
# from Wazuh (often empty — backfill via override service).
unique_cves: dict = {}
for vuln_data in raw_vulns:
cve_id = vuln_data.get("cve")
if not cve_id:
continue
if cve_id not in unique_cves:
unique_cves[cve_id] = {
"data": vuln_data,
"pkg_rows": {}, # name → {version, fixed}
"packages": set(), # legacy summary
"versions": set(), # legacy summary
}
pkg_name = vuln_data.get("name")
pkg_version = vuln_data.get("version")
pkg_fix = vuln_data.get("fixed_version")
if pkg_name:
unique_cves[cve_id]["packages"].add(pkg_name)
existing_pkg = unique_cves[cve_id]["pkg_rows"].get(pkg_name)
if existing_pkg is None:
unique_cves[cve_id]["pkg_rows"][pkg_name] = {
"version": pkg_version,
"fixed": pkg_fix,
}
else:
# Prefer first non-empty fix
if pkg_fix and not existing_pkg.get("fixed"):
existing_pkg["fixed"] = pkg_fix
if pkg_version:
unique_cves[cve_id]["versions"].add(pkg_version)
active_cves = set(unique_cves.keys())
logger.info(f"Sync: Agent {agent_id} has {len(active_cves)} unique CVEs after dedup.")
# Log severity breakdown from Wazuh data
severity_counts = {}
for cve_id_tmp, cve_info_tmp in unique_cves.items():
sev = (cve_info_tmp["data"].get("severity") or "unknown").lower()
severity_counts[sev] = severity_counts.get(sev, 0) + 1
logger.info(f"Sync: Agent {agent_id} Wazuh severity breakdown: {severity_counts}")
new_count = 0
updated_count = 0
# Track new vuln IDs so we can send a single digest mail per recipient
# at the end of this agent sync instead of one mail per CVE.
new_vuln_ids: list[int] = []
def _upsert_packages(vuln_id: int, rows: dict, sync_time: datetime) -> None:
"""Upsert one VulnerabilityPackage row per (vuln, package).
Updates last_seen_at for existing entries so the next stale-
package pass can prune ones Wazuh stopped reporting."""
if not rows:
return
existing_pkgs = {
p.package_name: p
for p in db.query(VulnerabilityPackage)
.filter(VulnerabilityPackage.vulnerability_id == vuln_id)
.all()
}
for pkg_name, info in rows.items():
row = existing_pkgs.get(pkg_name)
if row is None:
row = VulnerabilityPackage(
vulnerability_id=vuln_id,
package_name=pkg_name[:255],
package_version=(info.get("version") or None),
fixed_version=(info.get("fixed") or None),
source="wazuh",
first_detected_at=sync_time,
last_seen_at=sync_time,
)
db.add(row)
else:
row.last_seen_at = sync_time
# Only fill fixed_version when empty (override service
# may have written a better value from Vulnrichment).
if info.get("fixed") and not row.fixed_version:
row.fixed_version = info["fixed"][:100]
if info.get("version") and info["version"] != row.package_version:
row.package_version = info["version"][:100]
if not row.source:
row.source = "wazuh"
# 2. Process each unique CVE
for cve_id, cve_info in unique_cves.items():
vuln_data = cve_info["data"]
# Merge package strings (legacy parent-row summary)
merged_packages = ", ".join(sorted(cve_info["packages"]))
merged_versions = ", ".join(sorted(cve_info["versions"]))
# Truncate to fit DB columns
if len(merged_packages) > 250:
merged_packages = merged_packages[:247] + "..."
if len(merged_versions) > 95:
merged_versions = merged_versions[:92] + "..."
score = vuln_data.get("cvss", {}).get("cvss3", {}).get("base_score")
wazuh_severity = vuln_data.get("severity")
# Check if already exists
existing = db.query(Vulnerability).filter(
Vulnerability.cve_id == cve_id,
Vulnerability.asset_id == asset.id
).first()
if existing:
# Update existing entry — track wazuh as a source and reopen
# if Wazuh sees a previously-patched vuln again.
existing.add_source("wazuh")
if existing.status == VulnerabilityStatus.patched:
existing.status = VulnerabilityStatus.open
existing.patched_at = None
elif existing.status not in (
VulnerabilityStatus.pending_verification,
VulnerabilityStatus.accepted_risk,
VulnerabilityStatus.false_positive,
VulnerabilityStatus.deferred,
):
existing.status = VulnerabilityStatus.open
existing.package_name = merged_packages
existing.package_version = merged_versions
fv = vuln_data.get("fixed_version")
if fv and not existing.fixed_version:
existing.fixed_version = fv
if score:
existing.cvss_score = score
existing.severity = map_severity(score, wazuh_severity)
_upsert_packages(existing.id, cve_info["pkg_rows"], datetime.now())
existing.refresh_scores()
updated_count += 1
else:
# Create new — use savepoint so an IntegrityError (duplicate race)
# only rolls back this single insert, not the entire sync transaction.
try:
severity = map_severity(score, wazuh_severity)
vuln = Vulnerability(
cve_id=cve_id,
asset_id=asset.id,
cvss_score=score,
severity=severity,
status=VulnerabilityStatus.open,
title=vuln_data.get("title"),
package_name=merged_packages,
package_version=merged_versions,
fixed_version=vuln_data.get("fixed_version"),
detected_at=datetime.now(),
sources='["wazuh"]',
first_detected_by="wazuh",
)
with db.begin_nested():
db.add(vuln)
db.flush()
_upsert_packages(vuln.id, cve_info["pkg_rows"], datetime.now())
# Inherit canonical CVE metadata from existing siblings
# on other assets so the new row doesn't keep Wazuh's
# placeholder 10.0 when an override-pinned value exists.
try:
from app.services.vuln_override_service import apply_canonical_from_siblings
apply_canonical_from_siblings(db, vuln)
except Exception as e:
logger.warning("canonical-inherit on new wazuh vuln %s failed: %s", vuln.cve_id, e)
vuln.refresh_scores()
new_count += 1
new_vuln_ids.append(vuln.id)
logger.info(f"Sync: New vulnerability {cve_id} on {asset.hostname} - severity={severity.value}, cvss={score}, wazuh_severity={wazuh_severity}")
except IntegrityError:
# Savepoint rolled back; outer transaction intact.
logger.debug(f"Skipping duplicate CVE {cve_id} for asset {asset.id}")
continue
# 3. Source-aware backfill: remove "wazuh" from sources for any open vuln
# on this asset that Wazuh no longer reports. Only mark as PATCHED when
# ALL sources are gone — Nessus-only findings must not be patched here.
#
# Safety: when Wazuh returns 0 vulns for an agent (transient API outage,
# indexer warm-up after a manager restart, etc.) the loop below would
# mass-patch the entire history for the agent. Skip in that case —
# genuinely "everything fixed" is so rare that the false-positive risk
# is not worth taking.
if not active_cves and not raw_vulns:
logger.warning(
"Wazuh returned 0 vulns for agent %s (%s) — skipping backfill "
"to avoid mass-patching from a transient empty response.",
agent_id, asset.hostname,
)
asset.last_scan = datetime.now()
db.commit()
return new_count, updated_count, new_vuln_ids
stale_wazuh = (
db.query(Vulnerability)
.filter(
Vulnerability.asset_id == asset.id,
Vulnerability.status == VulnerabilityStatus.open,
Vulnerability.sources.contains('"wazuh"'),
)
.all()
)
for v in stale_wazuh:
if v.cve_id in active_cves:
continue # still seen by Wazuh — keep source
v.remove_source("wazuh")
if not v.source_list:
old_st = v.status
v.status = VulnerabilityStatus.patched
v.patched_at = datetime.now()
try:
log_vulnerability_change(
db, None, v.id, old_st, v.status,
reason=f"Wazuh agent {agent_id} no longer reports this CVE on {asset.hostname}",
cve_id=v.cve_id,
source="wazuh_sync",
)
except Exception as e:
logger.warning("audit log for wazuh auto-patch failed (vuln_id=%s): %s", v.id, e)
# Update asset last_scan
asset.last_scan = datetime.now()
db.commit()
logger.info(f"Sync: Completed for agent {agent_id} ({asset.hostname}): {new_count} new, {updated_count} updated, {len(active_cves)} total active CVEs")
# 4. Enrich newly created vulns with EPSS + KEV (best-effort, non-fatal)
if new_count > 0:
try:
new_vulns = db.query(Vulnerability).filter(
Vulnerability.asset_id == asset.id,
Vulnerability.cve_id.in_(active_cves),
Vulnerability.enrichment_updated_at.is_(None)
).all()
if new_vulns:
enrich_vulnerabilities(db, new_vulns)
except Exception as e:
logger.warning(f"Sync: Enrichment skipped for agent {agent_id}: {e}")
# 5. Dispatch ONE digest mail per recipient for this agent's newly
# discovered vulns (replaces per-CVE inline send). When this helper
# is called from a multi-agent loop the caller may prefer to skip
# per-agent dispatch and instead aggregate IDs and send a single
# cross-agent digest at the very end — currently each per-agent call
# produces its own digest.
if new_vuln_ids:
try:
from app.services.email_service import dispatch_new_vuln_notifications
fresh_vulns = db.query(Vulnerability).filter(
Vulnerability.id.in_(new_vuln_ids)
).all()
dispatch_new_vuln_notifications(db, fresh_vulns)
except Exception as e:
logger.warning(f"Sync: notification dispatch failed for agent {agent_id}: {e}")
def map_severity(cvss_score: Optional[float], wazuh_severity: Optional[str] = None) -> VulnerabilitySeverity:
"""Mappt CVSS-Score zu Severity-Enum, mit Wazuh-Severity als Fallback"""
if cvss_score and cvss_score > 0:
if cvss_score >= 9.0:
return VulnerabilitySeverity.critical
elif cvss_score >= 7.0:
return VulnerabilitySeverity.high
elif cvss_score >= 4.0:
return VulnerabilitySeverity.medium
else:
return VulnerabilitySeverity.low
# Fallback: Wazuh's eigene Severity-Einstufung verwenden
if wazuh_severity:
severity_lower = wazuh_severity.strip().lower()
severity_map = {
"critical": VulnerabilitySeverity.critical,
"high": VulnerabilitySeverity.high,
"medium": VulnerabilitySeverity.medium,
"low": VulnerabilitySeverity.low,
}
mapped = severity_map.get(severity_lower)
if mapped:
logger.info(f"No CVSS score available, using Wazuh severity fallback: {wazuh_severity} -> {mapped.value}")
return mapped
return VulnerabilitySeverity.none
# ============================================
# CVSS Override Endpoints (CISA Vulnrichment)
# ============================================
class OverrideDryRunResponse(BaseModel):
"""Response for dry-run mode"""
message: str
would_be_updated: int
checked: int
placeholder_scores_found: int
changes_preview: List[dict]
class OverrideStatsResponse(BaseModel):
"""Response after override execution"""
message: str
updated: int
checked: int
errors: int
changes: List[dict]
@router.get("/override/check")
async def check_incorrect_scores(
db: Session = Depends(get_db),
current_user: User = Depends(get_current_user)
):
"""
Findet alle Vulnerabilities mit wahrscheinlich falschen Scores.
Erkennt:
- Platzhalter-Scores (10.0) von Wazuh
- Unplausible Kombinationen (CVSS 10 + LOW Severity)
Gibt eine Liste mit Empfehlungen zurück.
"""
service = VulnOverrideService(db)
incorrect = service.find_incorrect_scores()
return {
"total_found": len(incorrect),
"placeholder_scores": len([i for i in incorrect if i.get("issue") == "placeholder_score_10"]),
"severity_mismatches": len([i for i in incorrect if i.get("issue") == "severity_mismatch"]),
"details": incorrect
}
@router.post("/override/nessus/{asset_id}")
async def override_from_nessus(
asset_id: int,
scan_id: Optional[int] = None,
dry_run: bool = Query(True, description="Nur simulieren ohne DB-Änderungen"),
db: Session = Depends(get_db),
current_user: User = Depends(RequireEditor)
):
"""
Korrigiert CVSS-Scores und Severity-Werte basierend auf Nessus-Daten.
Verwendet Nessus Plugin Output für verifizierte Scores.
- **asset_id**: Asset-DB-ID
- **scan_id**: Nessus Scan-ID (optional, wird aus Config ermittelt)
- **dry_run**: Wenn True, nur simulieren ohne Änderungen
"""
# Load Nessus config (transparently decrypted)
from app.auth.setting_crypto import read_setting_value
raw_nessus = read_setting_value(db, SETTING_KEY)
if not raw_nessus:
raise HTTPException(
status_code=400,
detail="Nessus configuration not set. Please configure Nessus in Settings."
)
try:
config = json.loads(raw_nessus)
except json.JSONDecodeError:
raise HTTPException(status_code=400, detail="Invalid Nessus configuration format.")
from app.integrations.nessus_client import NessusClient
try:
nessus_client = NessusClient(
base_url=config["base_url"],
access_key=config["access_key"],
secret_key=config["secret_key"],
verify_ssl=bool(config.get("verify_ssl", True)),
)
except ValueError as e:
raise HTTPException(status_code=400, detail=str(e))
service = VulnOverrideService(db)
try:
# Load verified data from Nessus
if scan_id:
verified_data = service.load_nessus_verified_data(asset_id, scan_id, nessus_client)
else:
# Use latest scan for this asset
from app.models.asset import Asset
asset = db.query(Asset).filter(Asset.id == asset_id).first()
if not asset:
raise HTTPException(status_code=404, detail=f"Asset {asset_id} not found")
# Get recent scan IDs
scans = nessus_client.list_scans()
if not scans:
raise HTTPException(status_code=404, detail="No Nessus scans found")
# Find scan for this asset's IP
target_scan_id = None
for scan in scans:
targets = scan.get("targets", "")
if asset.ip_address and asset.ip_address in targets:
target_scan_id = int(scan["id"])
break
if not target_scan_id:
target_scan_id = int(scans[0]["id"]) # Fallback to first scan
verified_data = service.load_nessus_verified_data(asset_id, target_scan_id, nessus_client)
# Apply overrides
if not verified_data:
return {
"message": "No verified CVE data found in Nessus for this asset",
"updated": 0,
"checked": 0
}
stats = service.apply_overrides(asset_id, verified_data, dry_run=dry_run)
# Audit log
if not dry_run and stats["updated"] > 0:
audit_log = AuditLog(
user_id=current_user.id,
event_type=AuditEventType.VULNERABILITY_UPDATED,
event_description=f"CVSS override applied from Nessus for {stats['updated']} vulnerabilities",
resource_type="vulnerability",
resource_id=f"asset:{asset_id}",
timestamp=datetime.now()
)
db.add(audit_log)
db.commit()
return {
"message": "Dry run - no changes applied" if dry_run else "Override applied successfully",
**stats
}
finally:
nessus_client.close()
@router.post("/override/vulnrichment")
async def override_from_vulnrichment(
cve_ids: Optional[List[str]] = Query(None, description="CVE-IDs zum Korrigieren (leer = alle)"),
asset_ids: Optional[List[int]] = Query(None, description="Asset-IDs zum Korrigieren (leer = alle)"),
dry_run: bool = Query(True, description="Nur simulieren ohne DB-Änderungen"),
db: Session = Depends(get_db),
current_user: User = Depends(RequireEditor)
):
"""
Korrigiert CVSS-Scores und Severity-Werte basierend auf CISA Vulnrichment.
Verwendet den offiziellen CISA Vulnrichment JSON Feed für verifizierte
CVSSv3.1 Base Scores und SSVC-Exploitation-Status.
- **cve_ids**: Optionale CVE-Filterliste (leer = alle offenen CVEs)
- **asset_ids**: Optionale Asset-Filterliste (leer = alle Assets)
- **dry_run**: Wenn True, nur simulieren ohne Änderungen
**Beispiel für CVE-2026-8390:**
- Wazuh (falsch): CVSS 10.0, Severity CRITICAL
- Vulnrichment (korrekt): CVSS 7.3, Severity HIGH, Exploitation "none"
"""
result = correct_vulnerability_scores(
db,
cve_ids=cve_ids,
asset_ids=asset_ids,
dry_run=dry_run
)
# Audit log
if not dry_run and result.get("updated", 0) > 0:
audit_log = AuditLog(
user_id=current_user.id,
event_type=AuditEventType.VULNERABILITY_UPDATED,
event_description=f"CVSS override from CISA Vulnrichment: {result['updated']} vulnerabilities corrected",
resource_type="vulnerability",
resource_id="bulk:vulnrichment",
old_value=json.dumps({"method": "cisa_vulnrichment"}),
new_value=json.dumps({"updated": result["updated"], "changes": result.get("changes", [])[:5]}), # Limit to 5 for audit
timestamp=datetime.now()
)
db.add(audit_log)
db.commit()
return result
@router.post("/override/vulnrichment/start")
async def start_vulnrichment_correction_job(
cve_ids: Optional[List[str]] = Query(None, description="CVE-IDs zum Korrigieren (leer = alle)"),
asset_ids: Optional[List[int]] = Query(None, description="Asset-IDs (leer = alle)"),
dry_run: bool = Query(False, description="Nur simulieren ohne DB-Änderungen"),
current_user: User = Depends(RequireEditor),
):
"""
Async-Variante des Vulnrichment-Overrides.
Spawnt einen Background-Thread und liefert sofort eine ``job_id``
zurück. Fortschritt + Endergebnis via
``GET /override/vulnrichment/status/{job_id}``.
Für Bulk-Korrekturen (Frontend "Correct CVSS" Button) verwenden —
der synchrone Endpoint blockiert die Browser-Verbindung und löst
bei großen Datenbanken den 23-Min-Hänger des Frontends aus.
"""
job_id = start_vulnrichment_job(
user_id=current_user.id,
cve_ids=cve_ids,
asset_ids=asset_ids,
dry_run=dry_run,
)
return {
"job_id": job_id,
"status": "queued",
"poll_url": f"/api/v1/vulnerabilities/override/vulnrichment/status/{job_id}",
}
@router.get("/override/vulnrichment/status/{job_id}")
async def get_vulnrichment_correction_status(
job_id: str,
current_user: User = Depends(get_current_user),
):
"""
Snapshot eines mit ``/override/vulnrichment/start`` gestarteten Jobs.
Felder:
- ``status``: queued | running | completed | failed
- ``stage``: human-readable progress label (z.B. "downloading ZIP")
- ``total`` / ``done``: numerische Progress-Hinweise
- ``updated`` / ``checked`` / ``not_found``: End-Stats wenn fertig
- ``error``: gesetzt bei status == "failed"
"""
job = get_override_job(job_id)
if not job:
raise HTTPException(
status_code=404,
detail=f"Job {job_id} not found (may have expired after backend restart)",
)
return job
@router.get("/override/vulnrichment/jobs")
async def list_vulnrichment_correction_jobs(
limit: int = Query(10, le=50),
current_user: User = Depends(get_current_user),
):
"""Most-recent-first list of override jobs (for an admin overview)."""
return list_override_jobs(limit=limit)
@router.post("/eol-check")
async def run_eol_check(
asset_id: Optional[int] = Query(None, description="Scope to one asset; omit for all wazuh-linked assets"),
db: Session = Depends(get_db),
current_user: User = Depends(RequireEditor),
):
"""Trigger endoflife.date EOL detection per installed package.
Closes the gap Wazuh has vs Nessus plugin 64784 (Microsoft SQL
Server Unsupported Version Detection): walks each installed
package from syscollector, resolves the product slug, and creates
one pseudo-CVE (cve_id starts with `EOL-`) per EOL or out-of-
active-support release on the asset.
Idempotent — re-runs upsert existing EOL rows rather than
duplicate them.
"""
from app.integrations.wazuh_client import WazuhClient
from app.models.setting import Setting as _Setting
from app.services import eol_service
config_row = db.query(_Setting).filter(_Setting.key == "wazuh_config").first()
if not config_row or not config_row.value:
raise HTTPException(400, "Wazuh is not configured")
wazuh_config = json.loads(config_row.value)
wazuh = WazuhClient(
api_url=wazuh_config.get("api_url"),
username=wazuh_config.get("username"),
password=wazuh_config.get("password"),
verify_ssl=wazuh_config.get("verify_ssl", True),
)
q = db.query(Asset).filter(Asset.wazuh_agent_id.isnot(None))
if asset_id is not None:
q = q.filter(Asset.id == asset_id)
assets = q.all()
if not assets:
return {"assets_scanned": 0, "eol_findings_total": 0}
stats = {
"assets_scanned": 0,
"packages_checked": 0,
"products_unmapped": 0,
"eol_findings_total": 0,
"eol_findings_new": 0,
"errors": [],
}
for asset in assets:
try:
pkgs = wazuh.get_packages(asset.wazuh_agent_id) or []
except Exception as e:
stats["errors"].append(f"asset {asset.id} ({asset.hostname}): {e}")
continue
stats["assets_scanned"] += 1
seen_keys: set[tuple[str, str]] = set()
for pkg in pkgs:
name = (pkg.get("name") or "").strip()
version = (pkg.get("version") or "").strip()
if not name or not version:
continue
key = (name.lower(), version)
if key in seen_keys:
continue
seen_keys.add(key)
stats["packages_checked"] += 1
slug = eol_service.resolve_product_slug(name)
if not slug:
stats["products_unmapped"] += 1
continue
try:
status = eol_service.check_eol(db, name, version)
except Exception as e:
logger.warning("EOL check failed for %s %s: %s", name, version, e)
continue
if not status:
continue
if not (status.is_eol or status.is_eoas):
continue
try:
_, was_created = eol_service.upsert_eol_vulnerability(
db,
asset_id=asset.id,
product_name=name,
installed_version=version,
status=status,
)
stats["eol_findings_total"] += 1
if was_created:
stats["eol_findings_new"] += 1
except Exception as e:
logger.warning("EOL upsert failed for %s on asset %s: %s", name, asset.id, e)
db.commit()
return stats
@router.post("/recompute-scores")
async def recompute_priority_scores(
db: Session = Depends(get_db),
current_user: User = Depends(RequireEditor),
):
"""One-off backfill of priority_score + cpr_score columns.
Migration 024 adds the columns but leaves them NULL. Forward
sync / enrich / override paths keep them current going forward.
Existing rows need this endpoint once after the upgrade. Cheap —
pure Python, ~5s for 15k rows.
"""
updated = 0
for vuln in db.query(Vulnerability).all():
try:
vuln.refresh_scores()
updated += 1
except Exception as e:
logger.warning("refresh_scores failed for vuln_id=%s: %s", vuln.id, e)
db.commit()
return {"updated": updated}
@router.post("/canonicalize-cve-metadata")
async def canonicalize_cve_metadata(
db: Session = Depends(get_db),
current_user: User = Depends(RequireEditor),
):
"""One-off normalisation of canonical CVE fields across sibling rows.
Tester report: "Same CVE shows different CVSS depending on sort
order." Cause: per-asset rows of the same CVE drifted apart over
time — override service touched some, not others.
For every distinct CVE in the DB, pick the freshest authoritative
source row (exploitation_source set, latest updated_at) and copy
its canonical fields (cvss_score, severity, exploitation_status,
SSVC, cvss_vector) onto all siblings. `fixed_version` is left
alone because Plan I picks per-package fixes that legitimately
differ between streams.
Cheap — single pass over the cve_id → vulns mapping. Forward
sync / enrich / override paths keep the canonical view current
going forward.
"""
from sqlalchemy import desc as _desc, nulls_last
from app.services.vuln_override_service import (
propagate_canonical_to_siblings, _CANONICAL_FIELDS,
)
cve_ids = [
r[0] for r in db.query(Vulnerability.cve_id).distinct().all()
if r[0]
]
cves_changed = 0
rows_changed = 0
for cve_id in cve_ids:
vulns = (
db.query(Vulnerability)
.filter(Vulnerability.cve_id == cve_id)
.all()
)
if len(vulns) < 2:
continue
# Source-of-truth pick: pinned source first, then most recent.
vulns.sort(
key=lambda v: (v.exploitation_source is not None, v.updated_at or datetime.min),
reverse=True,
)
src = vulns[0]
# Skip when the source itself has no authoritative values to share.
if all(getattr(src, f, None) is None for f in _CANONICAL_FIELDS):
continue
touched = propagate_canonical_to_siblings(db, src)
if touched:
cves_changed += 1
rows_changed += touched
db.commit()
return {
"distinct_cves_scanned": len(cve_ids),
"cves_normalised": cves_changed,
"rows_changed": rows_changed,
}
@router.get("/override/stats")
async def get_override_stats(
db: Session = Depends(get_db),
current_user: User = Depends(get_current_user)
):
"""
Gibt Statistiken über potenzielle Score-Fehler zurück.
Zeigt:
- Anzahl Platzhalter-Scores (10.0)
- Anzahl Severity-Fehlanpassungen
- Verteilung nach Severity-Level
"""
from sqlalchemy import func
# Count placeholder scores
placeholder_count = db.query(func.count(Vulnerability.id)).filter(
Vulnerability.cvss_score == 10.0
).scalar()
# Count severity mismatches (CVSS >= 9 but not CRITICAL)
mismatch_count = db.query(func.count(Vulnerability.id)).filter(
Vulnerability.cvss_score >= 9.0,
Vulnerability.severity != VulnerabilitySeverity.critical
).scalar()
# Severity distribution
severity_dist = db.query(
Vulnerability.severity,
func.count(Vulnerability.id)
).filter(
Vulnerability.status == VulnerabilityStatus.open
).group_by(Vulnerability.severity).all()
# CVSS distribution
cvss_ranges = [
("0.0-3.9", 0.0, 3.9),
("4.0-6.9", 4.0, 6.9),
("7.0-8.9", 7.0, 8.9),
("9.0-10.0", 9.0, 10.0),
]
cvss_dist = []
for label, low, high in cvss_ranges:
count = db.query(func.count(Vulnerability.id)).filter(
Vulnerability.cvss_score >= low,
Vulnerability.cvss_score < high if high < 10.0 else Vulnerability.cvss_score <= high
).scalar()
cvss_dist.append({"range": label, "count": count})
return {
"placeholder_scores_10": placeholder_count,
"severity_mismatches": mismatch_count,
"severity_distribution": {str(k): v for k, v in severity_dist},
"cvss_distribution": cvss_dist,
"recommendation": "Run POST /api/v1/vulnerabilities/override/vulnrichment with dry_run=false to correct values"
}
Reference in New Issue View Git Blame Copy Permalink