vulncheck
8ad2c592f8
Tester report: EOL check 500'd with duplicate key value violates unique constraint 'ix_settings_key' Key (key)=(eol_cache_mssqlserver) already exists. Cause: a single asset has many SQL-Server-related packages (engine, setup bootstrap, VSS writer, …) all mapping to the same slug 'mssqlserver'. _cache_get returned None for each in quick succession (autoflush quirk on this session), each path added a Setting row, the eventual flush hit the unique constraint. Fix - _PROCESS_MEMO dict in module scope — first fetch per slug per backend-process is the only one that talks to httpx + the DB. All subsequent lookups for the same slug return the in-RAM dict. - _cache_put rewritten as PostgreSQL ON CONFLICT DO UPDATE upsert (sqlalchemy.dialects.postgresql.insert) so even races between concurrent requests can't duplicate-key. - Failed cache write rolls back its own attempt instead of poisoning the outer transaction.
407 lines
14 KiB
Python
407 lines
14 KiB
Python
"""
|
|
endoflife.date integration.
|
|
|
|
Source-of-truth for product end-of-life / end-of-active-support /
|
|
end-of-security-support dates. Closes the EOL detection gap that
|
|
Wazuh syscollector has (Nessus plugin 64784 shows the format we want
|
|
to emulate: pseudo-vuln per EOL finding, severity=high, points the
|
|
operator at the upgrade path).
|
|
|
|
API surface used:
|
|
GET https://endoflife.date/api/v1/products/{slug}
|
|
→ {"result": {"name": ..., "releases": [{
|
|
"name": "13.0", "label": "2016",
|
|
"isMaintained": false,
|
|
"eoasFrom": "YYYY-MM-DD", # active-support ended
|
|
"eolFrom": "YYYY-MM-DD", # security-support ended (EOL)
|
|
"eoesFrom": "YYYY-MM-DD", # extended-security ended
|
|
"latest": {"name": "13.0.6300.2", "date": "..."},
|
|
}, ...]}}
|
|
|
|
Cache: catalog per product cached in `settings` table for 24h to keep
|
|
the EOL feed responsive (changes once per quarter at most).
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import logging
|
|
import re
|
|
from dataclasses import dataclass
|
|
from datetime import datetime, timedelta
|
|
from typing import Dict, List, Optional, Tuple
|
|
|
|
import httpx
|
|
from sqlalchemy.orm import Session
|
|
|
|
from app.models.setting import Setting
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
_API_BASE = "https://endoflife.date/api/v1/products"
|
|
_CACHE_KEY_PREFIX = "eol_cache_"
|
|
_CACHE_TTL = timedelta(hours=24)
|
|
|
|
# Per-process memo so a single sweep over 1000+ syscollector packages
|
|
# doesn't re-query the settings table (and risk a races-into-duplicate-
|
|
# key insert) for the same product slug.
|
|
_PROCESS_MEMO: dict[str, dict] = {}
|
|
|
|
|
|
# ----------------------------------------------------------------------
|
|
# Product slug mapping
|
|
# ----------------------------------------------------------------------
|
|
#
|
|
# Wazuh + Nessus report human product names ("Microsoft SQL Server",
|
|
# "Mozilla Firefox", "Google Chrome"). endoflife.date uses
|
|
# kebab/lowercase slugs. Hand-curated mapping — extend as new products
|
|
# show up. Keys are normalised (lowercase, alphanumeric only).
|
|
#
|
|
# When a name doesn't match exactly, the resolver tries best-effort
|
|
# substring matches against this map's keys before giving up.
|
|
_PRODUCT_SLUGS: dict[str, str] = {
|
|
# Microsoft
|
|
"mssqlserver": "mssqlserver",
|
|
"microsoftsqlserver": "mssqlserver",
|
|
"sqlserver": "mssqlserver",
|
|
"windowsserver": "windows-server",
|
|
"windowsserver2016": "windows-server",
|
|
"windowsserver2019": "windows-server",
|
|
"windowsserver2022": "windows-server",
|
|
"windows10": "windows",
|
|
"windows11": "windows",
|
|
"microsoftexchange": "exchange-server",
|
|
"exchangeserver": "exchange-server",
|
|
"microsoftoffice": "office",
|
|
"msoffice": "office",
|
|
"microsoftedge": "microsoft-edge",
|
|
"powershell": "powershell",
|
|
"dotnet": "dotnet",
|
|
"dotnetframework": "dotnetfx",
|
|
"iisexpress": "iis",
|
|
# Browsers / Mozilla
|
|
"mozillafirefox": "firefox",
|
|
"firefox": "firefox",
|
|
"firefoxesr": "firefox",
|
|
"googlechrome": "chrome",
|
|
"chrome": "chrome",
|
|
# Runtimes
|
|
"java": "java",
|
|
"jdk": "java",
|
|
"openjdk": "java",
|
|
"nodejs": "nodejs",
|
|
"python": "python",
|
|
"go": "go",
|
|
"ruby": "ruby",
|
|
"php": "php",
|
|
# Web / databases
|
|
"apache": "apache",
|
|
"apachehttpserver": "apache",
|
|
"nginx": "nginx",
|
|
"mysql": "mysql",
|
|
"mariadb": "mariadb",
|
|
"postgresql": "postgresql",
|
|
"postgres": "postgresql",
|
|
"mongodb": "mongodb",
|
|
"redis": "redis",
|
|
# Adobe
|
|
"adobeacrobat": "adobe-acrobat",
|
|
"adobeacrobatreader": "adobe-acrobat",
|
|
"adobeacrobatreaderdc": "adobe-acrobat",
|
|
"adobeacrobatdc": "adobe-acrobat",
|
|
# Linux distros
|
|
"ubuntu": "ubuntu",
|
|
"debian": "debian",
|
|
"centos": "centos",
|
|
"rhel": "rhel",
|
|
"redhatenterpriselinux": "rhel",
|
|
"amazonlinux": "amazon-linux",
|
|
"fedora": "fedora",
|
|
}
|
|
|
|
|
|
def _normalise_name(name: str) -> str:
|
|
"""Lowercase + strip non-alphanumeric for slug lookup."""
|
|
return re.sub(r"[^a-z0-9]", "", (name or "").lower())
|
|
|
|
|
|
def resolve_product_slug(product_name: Optional[str]) -> Optional[str]:
|
|
"""Map a Wazuh / Nessus product string to an endoflife.date slug.
|
|
|
|
Returns None when no match — caller should skip the EOL check
|
|
rather than guess (a wrong slug returns 404 from the API).
|
|
"""
|
|
if not product_name:
|
|
return None
|
|
key = _normalise_name(product_name)
|
|
if not key:
|
|
return None
|
|
if key in _PRODUCT_SLUGS:
|
|
return _PRODUCT_SLUGS[key]
|
|
# Substring scan — catches "Microsoft SQL Server 2016 Express"
|
|
# by matching the "microsoftsqlserver" key inside it.
|
|
for k, slug in _PRODUCT_SLUGS.items():
|
|
if k in key or key in k:
|
|
return slug
|
|
return None
|
|
|
|
|
|
# ----------------------------------------------------------------------
|
|
# API client (with settings-table cache)
|
|
# ----------------------------------------------------------------------
|
|
|
|
def _cache_get(db: Session, slug: str) -> Optional[dict]:
|
|
row = db.query(Setting).filter(Setting.key == _CACHE_KEY_PREFIX + slug).first()
|
|
if not row or not row.value:
|
|
return None
|
|
try:
|
|
payload = json.loads(row.value)
|
|
cached_at = datetime.fromisoformat(payload.get("cached_at", ""))
|
|
if datetime.now() - cached_at > _CACHE_TTL:
|
|
return None
|
|
return payload.get("data")
|
|
except (json.JSONDecodeError, ValueError, TypeError):
|
|
return None
|
|
|
|
|
|
def _cache_put(db: Session, slug: str, data: dict) -> None:
|
|
"""Upsert the cache row via PostgreSQL ON CONFLICT so concurrent
|
|
callers (or repeated calls within a non-flushed session) don't
|
|
duplicate-key against ix_settings_key."""
|
|
from sqlalchemy.dialects.postgresql import insert as pg_insert
|
|
|
|
payload = json.dumps({
|
|
"cached_at": datetime.now().isoformat(),
|
|
"data": data,
|
|
})
|
|
key = _CACHE_KEY_PREFIX + slug
|
|
now = datetime.now()
|
|
try:
|
|
stmt = pg_insert(Setting).values(
|
|
key=key,
|
|
value=payload,
|
|
description=f"endoflife.date cache for {slug}",
|
|
created_at=now,
|
|
updated_at=now,
|
|
).on_conflict_do_update(
|
|
index_elements=["key"],
|
|
set_={"value": payload, "updated_at": now},
|
|
)
|
|
db.execute(stmt)
|
|
except Exception as e:
|
|
logger.warning("eol cache upsert failed for %s: %s", slug, e)
|
|
# Roll back this savepoint-less attempt so the outer
|
|
# transaction stays usable for the rest of the loop.
|
|
try:
|
|
db.rollback()
|
|
except Exception:
|
|
pass
|
|
|
|
|
|
def fetch_product(db: Session, slug: str) -> Optional[dict]:
|
|
"""Cached fetch of one product's releases. None on 404 / network error.
|
|
|
|
Three cache layers:
|
|
1. _PROCESS_MEMO (in-RAM) — survives the request, never duplicates.
|
|
2. settings table cache (24h TTL) — survives container restarts.
|
|
3. live HTTP fetch.
|
|
"""
|
|
if slug in _PROCESS_MEMO:
|
|
return _PROCESS_MEMO[slug]
|
|
cached = _cache_get(db, slug)
|
|
if cached is not None:
|
|
_PROCESS_MEMO[slug] = cached
|
|
return cached
|
|
url = f"{_API_BASE}/{slug}"
|
|
try:
|
|
with httpx.Client(timeout=10.0) as client:
|
|
r = client.get(url, headers={"Accept": "application/json"})
|
|
if r.status_code == 404:
|
|
logger.info("endoflife.date: product '%s' not in catalog (404)", slug)
|
|
_PROCESS_MEMO[slug] = {}
|
|
_cache_put(db, slug, {}) # negative cache to skip repeated 404s
|
|
return {}
|
|
r.raise_for_status()
|
|
data = r.json()
|
|
_PROCESS_MEMO[slug] = data
|
|
_cache_put(db, slug, data)
|
|
return data
|
|
except Exception as e:
|
|
logger.warning("endoflife.date fetch failed for %s: %s", slug, e)
|
|
return None
|
|
|
|
|
|
# ----------------------------------------------------------------------
|
|
# EOL evaluation
|
|
# ----------------------------------------------------------------------
|
|
|
|
@dataclass
|
|
class EOLStatus:
|
|
"""Result of one installed-version EOL check."""
|
|
is_eol: bool # security-support ended
|
|
is_eoas: bool # active-support ended (still security-patched)
|
|
is_maintained: bool # endoflife flag — currently maintained
|
|
release_label: Optional[str] = None # "2016" / "150" / "8.0" — human stream name
|
|
release_name: Optional[str] = None # "13.0" / "150" — internal id
|
|
eol_date: Optional[str] = None # ISO YYYY-MM-DD
|
|
eoas_date: Optional[str] = None
|
|
eoes_date: Optional[str] = None
|
|
latest_version: Optional[str] = None
|
|
latest_date: Optional[str] = None
|
|
product_slug: Optional[str] = None
|
|
|
|
|
|
def _version_starts_with(release_name: str, installed: str) -> bool:
|
|
"""True if `installed` starts with `release_name` followed by '.' / end."""
|
|
if not release_name or not installed:
|
|
return False
|
|
if installed == release_name:
|
|
return True
|
|
return installed.startswith(release_name + ".") or installed.startswith(release_name + "-")
|
|
|
|
|
|
def _pick_release(releases: List[dict], installed: str) -> Optional[dict]:
|
|
"""Find the release entry whose `name` is the longest prefix of `installed`.
|
|
|
|
"13.0.4259.0 Express Edition" → matches release.name "13.0" over "13".
|
|
"""
|
|
candidates = [
|
|
r for r in releases
|
|
if isinstance(r, dict) and _version_starts_with(str(r.get("name") or ""), installed)
|
|
]
|
|
if not candidates:
|
|
return None
|
|
candidates.sort(key=lambda r: len(str(r.get("name") or "")), reverse=True)
|
|
return candidates[0]
|
|
|
|
|
|
def _past(date_str: Optional[str]) -> bool:
|
|
if not date_str:
|
|
return False
|
|
try:
|
|
d = datetime.fromisoformat(str(date_str).split("T")[0])
|
|
return d.date() <= datetime.now().date()
|
|
except (ValueError, TypeError):
|
|
return False
|
|
|
|
|
|
def check_eol(db: Session, product_name: str, installed_version: str) -> Optional[EOLStatus]:
|
|
"""Evaluate EOL status for a single (product, version) pair.
|
|
|
|
Returns None when the product can't be mapped to an endoflife slug
|
|
or the API returns no data — caller should treat that as "unknown,
|
|
skip" rather than "supported".
|
|
"""
|
|
slug = resolve_product_slug(product_name)
|
|
if not slug:
|
|
return None
|
|
data = fetch_product(db, slug)
|
|
if not data:
|
|
return None
|
|
result = (data.get("result") if isinstance(data, dict) else None) or {}
|
|
releases = result.get("releases") or []
|
|
rel = _pick_release(releases, installed_version)
|
|
if not rel:
|
|
return None
|
|
eol_date = rel.get("eolFrom")
|
|
eoas_date = rel.get("eoasFrom")
|
|
eoes_date = rel.get("eoesFrom")
|
|
latest = rel.get("latest") or {}
|
|
return EOLStatus(
|
|
is_eol=_past(eol_date) and not _past(eoes_date), # security-support ended (consider ESU extension)
|
|
is_eoas=_past(eoas_date),
|
|
is_maintained=bool(rel.get("isMaintained")),
|
|
release_label=rel.get("label"),
|
|
release_name=rel.get("name"),
|
|
eol_date=eol_date,
|
|
eoas_date=eoas_date,
|
|
eoes_date=eoes_date,
|
|
latest_version=latest.get("name") if isinstance(latest, dict) else None,
|
|
latest_date=latest.get("date") if isinstance(latest, dict) else None,
|
|
product_slug=slug,
|
|
)
|
|
|
|
|
|
# ----------------------------------------------------------------------
|
|
# Pseudo-vuln upsert
|
|
# ----------------------------------------------------------------------
|
|
|
|
def _pseudo_cve_id(slug: str, release_name: str) -> str:
|
|
"""Stable pseudo-CVE id per (product, release) so re-scans converge."""
|
|
safe_rel = re.sub(r"[^A-Za-z0-9._-]", "_", release_name)[:30]
|
|
return f"EOL-{slug.upper()}-{safe_rel}"[:50]
|
|
|
|
|
|
def upsert_eol_vulnerability(
|
|
db: Session,
|
|
*,
|
|
asset_id: int,
|
|
product_name: str,
|
|
installed_version: str,
|
|
status: EOLStatus,
|
|
) -> Tuple[Optional[int], bool]:
|
|
"""Create or refresh an EOL pseudo-vuln on the asset.
|
|
|
|
Returns (vuln_id, was_created).
|
|
"""
|
|
from app.models.vulnerability import Vulnerability, VulnerabilitySeverity, VulnerabilityStatus
|
|
|
|
cve_id = _pseudo_cve_id(status.product_slug or "unknown", status.release_name or "unknown")
|
|
severity = VulnerabilitySeverity.high if status.is_eol else VulnerabilitySeverity.medium
|
|
|
|
existing = (
|
|
db.query(Vulnerability)
|
|
.filter(Vulnerability.cve_id == cve_id, Vulnerability.asset_id == asset_id)
|
|
.first()
|
|
)
|
|
title = f"{product_name} {status.release_label or status.release_name} — {'EOL' if status.is_eol else 'end-of-active-support'}"
|
|
desc_lines = [
|
|
f"endoflife.date reports {product_name} release {status.release_label or status.release_name} "
|
|
f"({status.release_name}) is " + ("EOL (no further security patches)." if status.is_eol else "out of active support."),
|
|
]
|
|
if status.eol_date:
|
|
desc_lines.append(f"Security support ended: {status.eol_date}.")
|
|
if status.eoes_date:
|
|
desc_lines.append(f"Extended security support ends: {status.eoes_date}.")
|
|
if status.latest_version:
|
|
desc_lines.append(f"Latest supported release: {status.latest_version} ({status.latest_date or 'date unknown'}).")
|
|
desc_lines.append(f"Installed on this host: {installed_version}.")
|
|
description = "\n".join(desc_lines)
|
|
|
|
if existing:
|
|
existing.severity = severity
|
|
existing.title = title[:500]
|
|
existing.description = description
|
|
existing.package_version = installed_version[:100]
|
|
existing.fixed_version = (status.latest_version or None)
|
|
if existing.status == VulnerabilityStatus.patched:
|
|
existing.status = VulnerabilityStatus.open
|
|
existing.patched_at = None
|
|
try:
|
|
existing.refresh_scores()
|
|
except Exception:
|
|
pass
|
|
return existing.id, False
|
|
|
|
vuln = Vulnerability(
|
|
cve_id=cve_id,
|
|
asset_id=asset_id,
|
|
cvss_score=9.0 if status.is_eol else 6.0,
|
|
severity=severity,
|
|
status=VulnerabilityStatus.open,
|
|
title=title[:500],
|
|
description=description,
|
|
package_name=product_name[:255],
|
|
package_version=installed_version[:100],
|
|
fixed_version=(status.latest_version or None),
|
|
detected_at=datetime.now(),
|
|
sources='["endoflife.date"]',
|
|
first_detected_by="eol_check",
|
|
)
|
|
db.add(vuln)
|
|
db.flush()
|
|
try:
|
|
vuln.refresh_scores()
|
|
except Exception:
|
|
pass
|
|
return vuln.id, True
|