vulncheck
fdd1019fbd
The product mapping was hand-written and never checked against the actual catalogue, so a quarter of it was wrong. Every EOL run asked for ms-office, exchange-server, java, visual-cpp, adobe-acrobat, apache and iis, and got 404 back each time — visible in the log as "not in catalog", invisible everywhere else. The scan simply found no lifecycle data and moved on, which is why Microsoft Exchange Server 2016 came out looking supported. Checked against the live list (462 products) rather than guessed: exchange-server -> msexchange (the reported Exchange 2016 case) ms-office -> office apache -> apache-http-server java -> oracle-jdk The other three do not exist there at all and never will: endoflife.date has no record for Adobe Acrobat, the Visual C++ redistributables, or IIS. Their 18 mapping entries are removed rather than left to 404 on every run — IIS ships with Windows Server and is covered by that entry, the other two depend on the Microsoft export or a vendor page. Same reasoning as the existing Edge note. tools/check_eol_slugs.py verifies the whole mapping against the live catalogue and exits non-zero when a slug disappears, so this cannot rot again unnoticed.
906 lines
36 KiB
Python
906 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": "msexchange",
|
|
"exchangeserver": "msexchange",
|
|
"microsoftoffice": "office",
|
|
"msoffice": "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": "office",
|
|
"microsoftofficeosxmui": "office",
|
|
"microsoftofficeosxmuigerman": "office",
|
|
"microsoftofficeformac": "office",
|
|
"office": "office",
|
|
# NOT mapped, on purpose — verified against the live catalogue (462
|
|
# products) rather than assumed: endoflife.date carries no record for
|
|
# Adobe Acrobat, the Visual C++ redistributables or IIS, so every lookup
|
|
# returned 404 and filled the scan log with "not in catalog" for products
|
|
# that were never going to be there. IIS ships with Windows Server and is
|
|
# covered by that entry; the other two have no lifecycle feed at all and
|
|
# depend on the Microsoft export or a vendor page.
|
|
# 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",
|
|
# Browsers / Mozilla
|
|
"mozillafirefox": "firefox",
|
|
"firefox": "firefox",
|
|
"firefoxesr": "firefox",
|
|
"googlechrome": "chrome",
|
|
"chrome": "chrome",
|
|
# Runtimes
|
|
"java": "oracle-jdk",
|
|
"jdk": "oracle-jdk",
|
|
"openjdk": "oracle-jdk",
|
|
"nodejs": "nodejs",
|
|
"python": "python",
|
|
"go": "go",
|
|
"ruby": "ruby",
|
|
"php": "php",
|
|
# Web / databases
|
|
"apache": "apache-http-server",
|
|
"apachehttpserver": "apache-http-server",
|
|
"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.
|
|
# Adobe
|
|
# 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.
|
|
# 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", "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
|