vulncheckandClaude Opus 5
fdec6a7446
endoflife.date carries no record for Edge — the 'microsoft-edge' slug 404s, and no entry in their catalogue matches 'edge' at all — so every EOL check spent a request to find that out. Nothing is lost: Edge follows the Modern Lifecycle Policy and has no end-of-life date while it stays current, which makes 'is this version too old' a patch question. The CVE scan already answers that one. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
917 lines
36 KiB
Python
917 lines
36 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": "ms-office",
|
|
"msoffice": "ms-office",
|
|
# SharePoint — endoflife.date tracks it as `sharepoint` (2013 EOL
|
|
# 2023-04-11, 2016/2019 EOL 2026-07-14, Subscription Edition still
|
|
# supported). Neither the MS-lifecycle export nor the plain slug lookup
|
|
# caught it, so Foundation/Server installs never got an EOL finding.
|
|
# Covers the Server, Enterprise Server and Foundation flavours generically.
|
|
"sharepoint": "sharepoint",
|
|
"microsoftsharepoint": "sharepoint",
|
|
"microsoftsharepointserver": "sharepoint",
|
|
"microsoftsharepointfoundation": "sharepoint",
|
|
"microsoftsharepointenterpriseserver": "sharepoint",
|
|
"microsoftsharepointdesigner": "sharepoint",
|
|
"microsoftofficeproofing": "ms-office",
|
|
"microsoftofficeosxmui": "ms-office",
|
|
"microsoftofficeosxmuigerman": "ms-office",
|
|
"microsoftofficeformac": "ms-office",
|
|
"office": "ms-office",
|
|
# Microsoft Edge is deliberately NOT mapped. endoflife.date carries no
|
|
# record for it at all — the "microsoft-edge" slug 404s, so every EOL check
|
|
# spent a request finding that out. Edge follows the Modern Lifecycle
|
|
# Policy: it has no end-of-life date as long as it stays current, so "is
|
|
# this version too old" is a patch question, which the CVE scan already
|
|
# answers. Nothing is lost by leaving it out.
|
|
"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",
|
|
# Microsoft Visual C++ Redistributable (all flavours — 2005/2008/2010/2012/2013/2015-2022).
|
|
# endoflife.date exposes the product as `visual-cpp`; map any sane
|
|
# spelling here. Versions are matched by endoflife.date.
|
|
"visualc": "visual-cpp",
|
|
"visualcppredistributable": "visual-cpp",
|
|
"microsoftvisualc": "visual-cpp",
|
|
"microsoftvisualcp": "visual-cpp",
|
|
"microsoftvisualcppr": "visual-cpp",
|
|
"microsoftvisualcpprdistributable": "visual-cpp",
|
|
"microsoftvisualcplusplus": "visual-cpp",
|
|
"vcredist": "visual-cpp",
|
|
"vcruntime": "visual-cpp",
|
|
"msvcr": "visual-cpp",
|
|
"msvcp": "visual-cpp",
|
|
# Adobe
|
|
"adobeacrobat": "adobe-acrobat",
|
|
"adobeacrobatreader": "adobe-acrobat",
|
|
"adobeacrobatreaderdc": "adobe-acrobat",
|
|
"adobeacrobatdc": "adobe-acrobat",
|
|
# Nessus plugin 56213 reports "Adobe Reader" (no "Acrobat"), so the
|
|
# acrobat-prefixed keys above never substring-matched → fell back to
|
|
# EOL-NESSUS-56213. These aliases fix the slug resolution.
|
|
"adobereader": "adobe-acrobat",
|
|
"acrobatreader": "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())
|
|
|
|
|
|
# Third-party tools that merely *mention* a tracked product in their
|
|
# name ("Veeam Explorer for PostgreSQL", "PostgreSQL ODBC Driver",
|
|
# "MySQL Connector/NET"). These wrap/connect-to the product but are NOT
|
|
# the product itself — their version number is the tool's, not the
|
|
# product's, so an EOL match against the mentioned product is a false
|
|
# positive. If any of these tokens appears, skip the EOL check.
|
|
_WRAPPER_TOKENS = (
|
|
"veeam", "explorerfor", "backup", "connector", "odbc", "jdbc",
|
|
"driver", "clientfor", "agentfor", "pluginfor", "extensionfor",
|
|
"providerfor", "managementpack", "monitoringfor",
|
|
# Sub-components of a tracked product that have their own (different)
|
|
# lifecycle — matching the parent would give a false EOL signal.
|
|
"nativeclient", "setupsupportfiles", "setupsql", "setup",
|
|
"premium", "clicktorun", "subscription",
|
|
)
|
|
|
|
|
|
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
|
|
# Exact name always wins (curated full names).
|
|
if key in _PRODUCT_SLUGS:
|
|
return _PRODUCT_SLUGS[key]
|
|
# Guard: third-party wrapper/connector tools name-drop a product
|
|
# ("Veeam Explorer for PostgreSQL") — their version is the tool's,
|
|
# not the product's. Skip rather than emit a false EOL finding.
|
|
if any(tok in key for tok in _WRAPPER_TOKENS):
|
|
return None
|
|
# Anchored substring scan — the product name must *start with* a
|
|
# known key (or vice-versa) so "Microsoft SQL Server 2016 Express"
|
|
# still matches "microsoftsqlserver", but "...for PostgreSQL" (key
|
|
# mentioned mid/suffix) does NOT match "postgresql".
|
|
for k, slug in _PRODUCT_SLUGS.items():
|
|
if key.startswith(k) or k.startswith(key):
|
|
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 (real risk)
|
|
is_eoas: bool # active-support ended (still security-patched)
|
|
is_maintained: bool # endoflife flag — currently maintained
|
|
is_eol_soon: bool = False # security support ends within EOL_SOON_DAYS
|
|
days_to_eol: Optional[int] = None # signed days until eolFrom (negative = past)
|
|
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
|
|
|
|
|
|
# Window for the "EOL SOON" warning — security support ends within
|
|
# this many days from today. Tester: Windows Server 2016 still gets
|
|
# monthly CUs until 2027-01, shouldn't be flagged as already-EOL.
|
|
EOL_SOON_DAYS = 90
|
|
|
|
|
|
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]
|
|
|
|
|
|
# Products whose endoflife.date releases are keyed by *year* (release
|
|
# name/label "2016", "2019", ...) while the installed version reported
|
|
# by Wazuh syscollector is a build number ("16.0.4266.1001"). For these
|
|
# the year lives in the PRODUCT NAME ("Microsoft Office ... 2016"), so we
|
|
# match on the year token, not the numeric version prefix. Mirrors the
|
|
# OS path (resolve_os_to_eol), which also keys on name not version.
|
|
# Release is a YEAR ("2016") while the installed version is a build number
|
|
# ("16.0.5556.1005") — prefix-matching the version against the cycle can never
|
|
# hit, so the year comes from the product name instead. SharePoint is the same
|
|
# shape as Office, and worse: 2016, 2019 AND Subscription Edition all report
|
|
# 16.0.x, so the version alone can't even tell the releases apart.
|
|
_YEAR_KEYED_SLUGS = {"office", "ms-office", "sharepoint"}
|
|
|
|
|
|
def _extract_year(text: Optional[str]) -> Optional[str]:
|
|
"""Pull a 4-digit product year (2000-2099) from a name string."""
|
|
if not text:
|
|
return None
|
|
m = re.search(r"\b(20\d{2})\b", text)
|
|
return m.group(1) if m else None
|
|
|
|
|
|
def _pick_release_by_year(releases: List[dict], year: str) -> Optional[dict]:
|
|
"""Match a release by its year token against release name OR label.
|
|
|
|
endoflife.date office releases expose name="2016" / label="2016".
|
|
"""
|
|
if not year:
|
|
return None
|
|
for r in releases:
|
|
if not isinstance(r, dict):
|
|
continue
|
|
if str(r.get("name") or "") == year or str(r.get("label") or "") == year:
|
|
return r
|
|
return None
|
|
|
|
|
|
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 _build_eol_status(rel: dict, slug: str) -> "EOLStatus":
|
|
"""Construct an EOLStatus from a release dict, computing is_eol /
|
|
is_eoas / is_eol_soon consistently for both package + OS paths.
|
|
|
|
Severity model:
|
|
- is_eol : security support already ended (no patches) AND
|
|
not covered by an active ESU window → real risk.
|
|
- is_eol_soon: security support ends within EOL_SOON_DAYS but
|
|
hasn't yet (e.g. Win Server 2016 → Jan 2027). Still
|
|
patched today, but plan the upgrade.
|
|
- is_eoas : only mainstream/active support ended; security
|
|
patches still flow. Informational, NOT a finding
|
|
on its own (avoids the Server-2016 false-positive).
|
|
"""
|
|
eol_date = rel.get("eolFrom")
|
|
eoas_date = rel.get("eoasFrom")
|
|
eoes_date = rel.get("eoesFrom")
|
|
latest = rel.get("latest") or {}
|
|
days_to_eol = _days_until(eol_date)
|
|
# endoflife may give eolFrom as bool true (= already EOL, no date).
|
|
eol_is_bool_true = isinstance(eol_date, bool) and eol_date is True
|
|
sec_ended = (_past(eol_date) or eol_is_bool_true) and not _past(eoes_date)
|
|
eol_soon = (
|
|
not sec_ended
|
|
and days_to_eol is not None
|
|
and 0 <= days_to_eol <= EOL_SOON_DAYS
|
|
)
|
|
return EOLStatus(
|
|
is_eol=sec_ended,
|
|
is_eoas=_past(eoas_date),
|
|
is_eol_soon=eol_soon,
|
|
days_to_eol=days_to_eol,
|
|
is_maintained=bool(rel.get("isMaintained")),
|
|
release_label=rel.get("label"),
|
|
release_name=rel.get("name"),
|
|
eol_date=str(eol_date) if eol_date is not None else None,
|
|
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,
|
|
)
|
|
|
|
|
|
def _days_until(date_str: Optional[str]) -> Optional[int]:
|
|
"""Signed days from today to date_str. Negative = past, None = unparseable.
|
|
endoflife.date sometimes uses a bool (true/false) for eolFrom instead
|
|
of a date — return None in that case."""
|
|
if not date_str or isinstance(date_str, bool):
|
|
return None
|
|
try:
|
|
d = datetime.fromisoformat(str(date_str).split("T")[0]).date()
|
|
return (d - datetime.now().date()).days
|
|
except (ValueError, TypeError):
|
|
return None
|
|
|
|
|
|
# ----------------------------------------------------------------------
|
|
# OS-level EOL (asset.operating_system + os_version)
|
|
# ----------------------------------------------------------------------
|
|
#
|
|
# Wazuh reports OS strings like:
|
|
# "Microsoft Windows Server 2008 R2" os_version "6.1.7601"
|
|
# "Microsoft Windows Server 2016 Datacenter"
|
|
# "Microsoft Windows 10 Pro" os_version "10.0.19045"
|
|
# "Ubuntu" os_version "22.04.3 LTS"
|
|
# "CentOS Linux" os_version "7"
|
|
# "Debian GNU/Linux" os_version "11"
|
|
#
|
|
# endoflife.date uses release codenames, NOT numeric versions:
|
|
# windows-server: "2008-r2", "2012", "2016", "2019", "2022", "2025"
|
|
# windows: "10", "11" (with sub-cycles 22h2 etc.)
|
|
# ubuntu: "22.04", "20.04", "18.04"
|
|
# debian: "11", "10"
|
|
#
|
|
# So OS matching is name-pattern based, not the numeric _pick_release.
|
|
|
|
def resolve_os_to_eol(os_name: Optional[str], os_version: Optional[str]) -> Optional[tuple]:
|
|
"""Map a Wazuh OS string to (endoflife_slug, release_codename).
|
|
|
|
Returns None when no confident match — caller skips rather than
|
|
guess. Codename is matched against release.name in check_os_eol.
|
|
"""
|
|
if not os_name:
|
|
return None
|
|
n = os_name.lower()
|
|
ver = (os_version or "").lower()
|
|
|
|
# --- Windows Server ---
|
|
if "windows server" in n or ("windows" in n and "server" in n):
|
|
# Pull the year + optional R2 from the name.
|
|
m = re.search(r"server\s+(\d{4})(\s*r2)?", n)
|
|
if m:
|
|
year = m.group(1)
|
|
r2 = "-r2" if m.group(2) else ""
|
|
return ("windows-server", f"{year}{r2}")
|
|
return ("windows-server", None)
|
|
|
|
# --- Windows client (10 / 11) ---
|
|
if "windows" in n:
|
|
m = re.search(r"windows\s+(\d{1,2})", n)
|
|
if m:
|
|
return ("windows", m.group(1))
|
|
# os_version "10.0.x" → Win10/11 distinguished by build, but
|
|
# endoflife slug "windows" release "10"/"11" — best effort: 10.
|
|
if ver.startswith("10.0."):
|
|
# build >= 22000 = Windows 11
|
|
mb = re.search(r"10\.0\.(\d+)", ver)
|
|
if mb and int(mb.group(1)) >= 22000:
|
|
return ("windows", "11")
|
|
return ("windows", "10")
|
|
return ("windows", None)
|
|
|
|
# --- Ubuntu ---
|
|
if "ubuntu" in n:
|
|
m = re.search(r"(\d{2}\.\d{2})", ver) or re.search(r"(\d{2}\.\d{2})", n)
|
|
if m:
|
|
return ("ubuntu", m.group(1))
|
|
return ("ubuntu", None)
|
|
|
|
# --- Debian ---
|
|
if "debian" in n:
|
|
m = re.search(r"(\d{1,2})", ver) or re.search(r"(\d{1,2})", n)
|
|
if m:
|
|
return ("debian", m.group(1))
|
|
return ("debian", None)
|
|
|
|
# --- RHEL / CentOS ---
|
|
if "red hat" in n or "rhel" in n:
|
|
m = re.search(r"(\d{1,2})", ver)
|
|
if m:
|
|
return ("rhel", m.group(1))
|
|
return ("rhel", None)
|
|
if "centos" in n:
|
|
m = re.search(r"(\d{1,2})", ver)
|
|
if m:
|
|
return ("centos", m.group(1))
|
|
return ("centos", None)
|
|
|
|
return None
|
|
|
|
|
|
def check_os_eol(db: Session, os_name: str, os_version: str) -> Optional[EOLStatus]:
|
|
"""EOL evaluation for an operating system. Matches the OS codename
|
|
(e.g. '2008-r2') against endoflife release.name as a prefix."""
|
|
resolved = resolve_os_to_eol(os_name, os_version)
|
|
if not resolved:
|
|
return None
|
|
slug, codename = resolved
|
|
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 = None
|
|
if codename:
|
|
# Prefer exact codename, then prefix.
|
|
for r in releases:
|
|
if str(r.get("name") or "").lower() == codename.lower():
|
|
rel = r
|
|
break
|
|
if rel is None:
|
|
for r in releases:
|
|
rn = str(r.get("name") or "").lower()
|
|
if rn.startswith(codename.lower()) or codename.lower().startswith(rn):
|
|
rel = r
|
|
break
|
|
if rel is None:
|
|
return None
|
|
return _build_eol_status(rel, slug)
|
|
|
|
|
|
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 = None
|
|
# Year-keyed products (Office): release name/label is "2016", but the
|
|
# installed version is a build number ("16.0.4266.1001"). Pull the year
|
|
# from the product name instead of prefix-matching the version.
|
|
if slug in _YEAR_KEYED_SLUGS:
|
|
year = _extract_year(product_name)
|
|
if year:
|
|
rel = _pick_release_by_year(releases, year)
|
|
if rel is None:
|
|
rel = _pick_release(releases, installed_version)
|
|
if not rel:
|
|
return None
|
|
return _build_eol_status(rel, 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 _supersede_old_eol(db: "Session", asset_id: int, slug: Optional[str], keep_cve_id: str) -> None:
|
|
"""A product runs exactly ONE release per asset. When we upsert the EOL
|
|
finding for the current release, any OTHER open EOL finding for the same
|
|
product/asset is stale (the device moved to a new major, e.g. Chrome
|
|
149→150) — resolve it so the old release doesn't linger as a duplicate."""
|
|
from app.models.vulnerability import Vulnerability, VulnerabilityStatus
|
|
if not slug or slug == "unknown":
|
|
return
|
|
prefix = f"EOL-{slug.upper()}-"
|
|
stale = (
|
|
db.query(Vulnerability)
|
|
.filter(Vulnerability.asset_id == asset_id,
|
|
Vulnerability.status == VulnerabilityStatus.open,
|
|
Vulnerability.first_detected_by == "eol_check",
|
|
Vulnerability.cve_id.like(f"{prefix}%"),
|
|
Vulnerability.cve_id != keep_cve_id)
|
|
.all()
|
|
)
|
|
for v in stale:
|
|
old_status = v.status
|
|
v.status = VulnerabilityStatus.patched
|
|
v.patched_at = datetime.now()
|
|
try:
|
|
from app.routers.vulnerabilities import log_vulnerability_change
|
|
log_vulnerability_change(
|
|
db, None, v.id, old_status, v.status,
|
|
reason=f"Superseded — asset moved to a newer {slug} release ({keep_cve_id})",
|
|
cve_id=v.cve_id, source="eol_supersede",
|
|
)
|
|
except Exception as e:
|
|
logger.warning("audit log for EOL supersede failed (vuln_id=%s): %s", v.id, e)
|
|
|
|
|
|
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 tiers:
|
|
# EOL 1000+ days → CRITICAL, cvss 9.8, title "EOL 1000d+"
|
|
# EOL (security ended) → HIGH, cvss 9.0, title "EOL"
|
|
# EOL SOON (≤90d) → MEDIUM, cvss 5.5, title "EOL SOON"
|
|
# EOAS only (still patched)→ LOW, cvss 3.0, title "end-of-active-support"
|
|
# The days-past-EOL escalation mirrors the endoflife.date/Wazuh EOL model:
|
|
# the longer a product has been unpatched, the higher the standing risk.
|
|
if status.is_eol:
|
|
# days_to_eol is signed (negative = past); guard the bool-true case
|
|
# (endoflife eolFrom=true, no date → days unknown).
|
|
days_past = (-status.days_to_eol
|
|
if status.days_to_eol is not None and status.days_to_eol < 0
|
|
else None)
|
|
if days_past is not None and days_past >= 1000:
|
|
severity = VulnerabilitySeverity.critical
|
|
cvss = 9.8
|
|
state_label = f"EOL {days_past}d"
|
|
state_desc = f"EOL for {days_past} days (no security patches — critical exposure)."
|
|
else:
|
|
severity = VulnerabilitySeverity.high
|
|
cvss = 9.0
|
|
state_label = f"EOL {days_past}d" if days_past is not None else "EOL"
|
|
state_desc = (f"EOL for {days_past} days (no further security patches)."
|
|
if days_past is not None
|
|
else "EOL (no further security patches).")
|
|
elif status.is_eol_soon:
|
|
severity = VulnerabilitySeverity.medium
|
|
cvss = 5.5
|
|
d = status.days_to_eol if status.days_to_eol is not None else "?"
|
|
state_label = f"EOL SOON ({d}d)"
|
|
state_desc = f"security support ends in {d} days ({status.eol_date}) — plan the upgrade."
|
|
else:
|
|
# EOAS only — still receiving security patches.
|
|
severity = VulnerabilitySeverity.low
|
|
cvss = 3.0
|
|
state_label = "end-of-active-support"
|
|
state_desc = "out of active/mainstream support (security patches still flow)."
|
|
|
|
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} — {state_label}"
|
|
desc_lines = [
|
|
f"endoflife.date reports {product_name} release {status.release_label or status.release_name} "
|
|
f"({status.release_name}) is {state_desc}",
|
|
]
|
|
if status.eol_date:
|
|
verb = "ended" if status.is_eol else "ends"
|
|
desc_lines.append(f"Security support {verb}: {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}.")
|
|
if status.is_eol:
|
|
# Running EOL software is an explicit control failure in the major
|
|
# frameworks — surface the mapping so audits/reports can cite it.
|
|
desc_lines.append(
|
|
"Compliance: running end-of-life software violates PCI-DSS 6.3.3, "
|
|
"NIST 800-53 CM-8, and HIPAA 164.312(a)(1)."
|
|
)
|
|
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)
|
|
from app.services.audit_events import reopen_if_patched
|
|
reopen_if_patched(db, existing, reason="endoflife.date check reports this product as EOL again", source="eol_check")
|
|
# Resync bumps detected_at so the Newly EOL/EOS widget ranks the
|
|
# freshest finding first.
|
|
existing.detected_at = datetime.now()
|
|
try:
|
|
existing.refresh_scores()
|
|
except Exception:
|
|
pass
|
|
_supersede_old_eol(db, asset_id, status.product_slug, cve_id)
|
|
return existing.id, False
|
|
|
|
vuln = Vulnerability(
|
|
cve_id=cve_id,
|
|
asset_id=asset_id,
|
|
cvss_score=cvss,
|
|
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
|
|
# Revisionssicher: initial detected-event for the new EOL finding.
|
|
try:
|
|
from app.services.audit_events import audit_new_vulnerabilities
|
|
audit_new_vulnerabilities(db, [vuln.id], source="eol_check")
|
|
except Exception:
|
|
pass
|
|
_supersede_old_eol(db, asset_id, status.product_slug, cve_id)
|
|
return vuln.id, True
|
|
|
|
|
|
def run_eol_for_packages(db: "Session", asset, packages: list) -> int:
|
|
"""Source-agnostic per-package EOL detection for one asset.
|
|
|
|
`packages` = list of {name, version}. endoflife.date first, then the
|
|
MS-lifecycle export / hardcoded-exotics fallback when endoflife has
|
|
nothing actionable (same precedence as the eol-check endpoint). Used by
|
|
both the Wazuh eol-check and the Intune detectedApps inventory. Returns
|
|
the number of EOL findings upserted. Caller commits.
|
|
"""
|
|
count = 0
|
|
seen: set = set()
|
|
kept_msl: set = set() # MS-lifecycle EOL cve_ids still valid this run
|
|
for pkg in packages or []:
|
|
name = (pkg.get("name") or "").strip()
|
|
version = (pkg.get("version") or "").strip()
|
|
if not name or not version:
|
|
continue
|
|
key = (name.lower(), version)
|
|
if key in seen:
|
|
continue
|
|
seen.add(key)
|
|
|
|
status = None
|
|
if resolve_product_slug(name):
|
|
try:
|
|
status = check_eol(db, name, version)
|
|
except Exception:
|
|
status = None
|
|
actionable = status and (status.is_eol or status.is_eol_soon or status.is_eoas)
|
|
if not actionable:
|
|
try:
|
|
from app.services import ms_lifecycle_service
|
|
ms = ms_lifecycle_service.resolve_ms_lifecycle_eol(db, name, version)
|
|
if ms and (ms.is_eol or ms.is_eol_soon):
|
|
status = ms
|
|
actionable = True
|
|
except Exception:
|
|
pass
|
|
if not actionable:
|
|
continue
|
|
if status and status.product_slug == "ms-lifecycle":
|
|
kept_msl.add(_pseudo_cve_id("ms-lifecycle", status.release_name or "unknown"))
|
|
try:
|
|
upsert_eol_vulnerability(
|
|
db, asset_id=asset.id, product_name=name,
|
|
installed_version=version, status=status,
|
|
)
|
|
count += 1
|
|
except Exception as e:
|
|
logger.warning("EOL-for-packages upsert failed (%s on asset %s): %s", name, asset.id, e)
|
|
|
|
# Guard on a non-empty inventory: an empty list is a transient/failed read,
|
|
# not proof the products are gone (same safety as the other reconciles).
|
|
if packages:
|
|
_resolve_stale_ms_lifecycle(db, asset.id, kept_msl)
|
|
return count
|
|
|
|
|
|
def _resolve_stale_ms_lifecycle(db: "Session", asset_id: int, kept: set) -> None:
|
|
"""Resolve open MS-lifecycle EOL findings the current run no longer produced.
|
|
|
|
Without this, fixing a bad name→product match (tester: 'Microsoft Edge'
|
|
browser mis-mapped to the 'Azure Stack Edge' listing) left the wrong finding
|
|
open forever, since _supersede_old_eol only fires when a REPLACEMENT is
|
|
created. Scoped to the EOL-MS-LIFECYCLE- prefix, which only the package path
|
|
produces — OS-level endoflife.date findings use other slugs and are
|
|
untouched. Only reconciles when at least one package was inventoried (empty
|
|
package list = nothing to conclude)."""
|
|
from app.models.vulnerability import Vulnerability, VulnerabilityStatus
|
|
stale = (
|
|
db.query(Vulnerability)
|
|
.filter(Vulnerability.asset_id == asset_id,
|
|
Vulnerability.status == VulnerabilityStatus.open,
|
|
Vulnerability.cve_id.like("EOL-MS-LIFECYCLE-%"))
|
|
.all()
|
|
)
|
|
for v in stale:
|
|
if v.cve_id in kept:
|
|
continue
|
|
old_status = v.status
|
|
v.status = VulnerabilityStatus.patched
|
|
v.patched_at = datetime.now()
|
|
try:
|
|
from app.routers.vulnerabilities import log_vulnerability_change
|
|
log_vulnerability_change(
|
|
db, None, v.id, old_status, v.status,
|
|
reason="MS lifecycle no longer matches this installed product "
|
|
"(re-evaluated — not end-of-life)",
|
|
cve_id=v.cve_id, source="eol_reconcile",
|
|
)
|
|
except Exception as e:
|
|
logger.warning("audit log for MS-lifecycle reconcile failed (vuln_id=%s): %s", v.id, e)
|
|
|
|
|
|
def revalidate_ms_lifecycle_findings(db: "Session") -> int:
|
|
"""Re-check every OPEN MS-lifecycle EOL finding and close the ones that no
|
|
longer match. Returns how many were closed.
|
|
|
|
Path-independent on purpose. _resolve_stale_ms_lifecycle only runs inside
|
|
run_eol_for_packages, but the EOL-check endpoint (the button) has its own
|
|
loop and never called it — so a finding produced by a since-fixed name match
|
|
stayed open forever (tester: 'Microsoft Edge' the browser matched the
|
|
'Azure Stack Edge' listing; the match was fixed, the finding was not).
|
|
Re-asking the resolver per finding is cheap: the lifecycle rows are memoised
|
|
in-process, so this costs one fetch at most.
|
|
"""
|
|
from app.models.vulnerability import Vulnerability, VulnerabilityStatus
|
|
from app.services import ms_lifecycle_service
|
|
|
|
rows = (db.query(Vulnerability)
|
|
.filter(Vulnerability.status == VulnerabilityStatus.open,
|
|
Vulnerability.cve_id.like("EOL-MS-LIFECYCLE-%"))
|
|
.all())
|
|
closed = 0
|
|
for v in rows:
|
|
name = (v.package_name or "").strip()
|
|
if not name:
|
|
continue
|
|
try:
|
|
st = ms_lifecycle_service.resolve_ms_lifecycle_eol(
|
|
db, name, v.package_version or "")
|
|
except Exception as e:
|
|
logger.debug("MS-lifecycle revalidate failed for %s: %s", name, e)
|
|
continue # unreachable source → leave the finding alone
|
|
if st and (st.is_eol or st.is_eol_soon):
|
|
continue # still EOL → keep
|
|
old_status = v.status
|
|
v.status = VulnerabilityStatus.patched
|
|
v.patched_at = datetime.now()
|
|
closed += 1
|
|
try:
|
|
from app.routers.vulnerabilities import log_vulnerability_change
|
|
log_vulnerability_change(
|
|
db, None, v.id, old_status, v.status,
|
|
reason=f"MS lifecycle no longer reports '{name}' as end-of-life "
|
|
f"(re-evaluated — earlier match was wrong)",
|
|
cve_id=v.cve_id, source="eol_revalidate",
|
|
hostname=(v.asset.hostname if v.asset else None),
|
|
)
|
|
except Exception as e:
|
|
logger.warning("audit log for MS-lifecycle revalidate failed (%s): %s", v.id, e)
|
|
if closed:
|
|
db.commit()
|
|
logger.info("MS-lifecycle revalidate: closed %d stale finding(s)", closed)
|
|
return closed
|