feat(kev): alert on exploited CVEs that are open in our own environment
Actively-exploited CVEs were shown from CISA KEV only, on a dashboard somebody had to be looking at. ENISA's EUVD exploited catalog was already being fetched for enrichment but never surfaced as a KEV source. - kev_service merges CISA KEV + ENISA EUVD (exploited only — an EU-Critical entry is a priority list, not an exploitation claim) into one entry per CVE carrying both sources. - Inventory impact is counted honestly: active assets only (matching the vulnerabilities list), open findings apart from remediated ones, and an asset with both an open and a closed row counted once, as open. That gap is what made the badge say 64 where the filtered view listed 62. - kev_alert_service mails the CVEs with open findings here — hostnames, IPs, counts, sources — once per CVE, again when more systems are affected. Hourly, plus straight after each threat-intel refresh. - Dashboard pins what is open here above the newest listings; Advisories gains source filters, an alert panel and its config. EUVD entries now carry an explicit `exploited` flag; caches predating it cannot answer the question for an entry that is also EU-Critical, so they are treated as stale and refetched once.
This commit is contained in:
+4
-1
@@ -118,6 +118,8 @@ Pure business logic, no FastAPI imports. Reusable from scheduler + routers.
|
||||
| Service | Role |
|
||||
|---|---|
|
||||
| `enrichment_service.py` | EPSS / KEV / EUVD enrichment per CVE, with 24h cache in `settings` |
|
||||
| `kev_service.py` | Merges the KEV sources (CISA KEV + ENISA EUVD exploited) into one catalog and annotates it with inventory impact (open vs remediated, active assets only) |
|
||||
| `kev_alert_service.py` | Immediate mail when an actively-exploited CVE has open findings here; idempotent per CVE |
|
||||
| `vuln_override_service.py` | 3-stage CVSS / SSVC / fixed_version cascade (Vulnrichment → NVD → cvelistV5) |
|
||||
| `override_jobs.py` | Async job tracker for the long-running "Correct CVSS" button |
|
||||
| `nessus_sync.py` | Per-scan import, asset matching cascade, pseudo-CVE handling |
|
||||
@@ -173,7 +175,8 @@ SSO/LDAP login. `RoleMapper` re-evaluates role on every login from
|
||||
|---|---|---|
|
||||
| `scheduler_sync` | every 60 s | Picks up DB changes to `scan_schedules` and re-registers jobs |
|
||||
| `sla_breach_check` | every 1 h | SLA-overdue scan; honors `sla_breach_enabled` toggle + `PolicyStatus.DISABLED` skip; digest or single mode |
|
||||
| `threat_intel_refresh` | every 24 h | Refreshes EPSS, KEV, EUVD across all open vulns |
|
||||
| `threat_intel_refresh` | every 24 h | Refreshes EPSS, KEV, EUVD across all open vulns; runs `kev_alert_check` straight after, since the catalogs just moved |
|
||||
| `kev_alert_check` | hourly, :25 | Mails actively-exploited CVEs (CISA KEV + ENISA EUVD) that have OPEN findings on active assets; honors `kev_alert_enabled`, idempotent via `kev_alert_state` |
|
||||
| `compliance_sca_nightly` | 02:00 UTC | Wazuh SCA pull for every linked asset |
|
||||
| `vulnrichment_nightly` | 03:00 UTC | 3-stage CVSS/SSVC/fixed_version cascade |
|
||||
| `urs_nightly` | 04:00 UTC | URS recompute + asset_risk_snapshots prune (>90d) |
|
||||
|
||||
+5
-1
@@ -221,7 +221,7 @@ shows the human-friendly source string.
|
||||
| vulnerability_id | INTEGER | FK, INDEX | |
|
||||
| asset_id | INTEGER | FK, INDEX | |
|
||||
| user_id | INTEGER | FK, INDEX | Recipient user |
|
||||
| notification_type | ENUM `notificationtype` | NOT NULL, INDEX | `NEW_VULN` / `SLA_BREACH` / `STATUS_CHANGE` / etc. |
|
||||
| notification_type | ENUM `notificationtype` | NOT NULL, INDEX | `SLA_BREACH` / `ASSIGNMENT` / `NEW_VULNERABILITY` / `KEV_ALERT` / `MANUAL` |
|
||||
| sent_at | TIMESTAMP | NOT NULL, INDEX | |
|
||||
| subject | VARCHAR(500) | NULLABLE | |
|
||||
| recipient_email | VARCHAR(255) | NULLABLE | |
|
||||
@@ -269,6 +269,10 @@ Key-value store. Stable keys:
|
||||
| `email_template_*` | HTML/Jinja-light | Override default mail templates |
|
||||
| `kev_cache_json` | JSON | CISA KEV catalog cache (24h TTL) |
|
||||
| `euvd_cache_json` | JSON | ENISA EUVD catalog cache (24h TTL) |
|
||||
| `kev_alert_enabled` | `true` \| `false` | Master toggle for the hourly KEV alert mail (default on) |
|
||||
| `kev_alert_sources` | `cisa,euvd` | Which KEV sources feed the alert; empty = all |
|
||||
| `kev_alert_recipients` | string | Comma/semicolon list; empty = `notification_default_recipients`, else active admins |
|
||||
| `kev_alert_state` | JSON | `{cve: {at, n}}` — what was already mailed, so a run is idempotent |
|
||||
|
||||
### `compliance_results`
|
||||
|
||||
|
||||
@@ -0,0 +1,33 @@
|
||||
"""Add KEV_ALERT notification type
|
||||
|
||||
Revision ID: 043
|
||||
Revises: 042
|
||||
Create Date: 2026-08-15 10:00:00.000000
|
||||
|
||||
Immediate alerting for actively-exploited CVEs (CISA KEV / ENISA EUVD) that
|
||||
have open findings on our own active assets. Its own notification type so the
|
||||
notification log can be filtered for the alerts that meant "act now" rather
|
||||
than mixing them into the nightly new-vulnerability stream.
|
||||
|
||||
ALTER TYPE ... ADD VALUE cannot run inside a transaction block on older
|
||||
Postgres → autocommit_block. Idempotent (IF NOT EXISTS).
|
||||
"""
|
||||
from alembic import op
|
||||
|
||||
|
||||
revision = "043"
|
||||
down_revision = "042"
|
||||
branch_labels = None
|
||||
depends_on = None
|
||||
|
||||
|
||||
def upgrade() -> None:
|
||||
with op.get_context().autocommit_block():
|
||||
op.execute("ALTER TYPE notificationtype ADD VALUE IF NOT EXISTS 'KEV_ALERT'")
|
||||
|
||||
|
||||
def downgrade() -> None:
|
||||
# Postgres cannot drop an enum label. Rows carrying it would have to be
|
||||
# rewritten first, and losing the record that an alert was sent is worse
|
||||
# than an unused label — so this is deliberately a no-op.
|
||||
pass
|
||||
@@ -42,6 +42,19 @@ def create_initial_data(db: Session):
|
||||
"true",
|
||||
"Enable ENISA EUVD enrichment (EU exploited/critical vulnerabilities)"
|
||||
)
|
||||
_ensure_default_setting(
|
||||
db,
|
||||
"kev_alert_enabled",
|
||||
"true",
|
||||
"Email immediately when an actively exploited CVE (CISA KEV / "
|
||||
"ENISA EUVD) has open findings on an active asset"
|
||||
)
|
||||
_ensure_default_setting(
|
||||
db,
|
||||
"kev_alert_sources",
|
||||
"cisa,euvd",
|
||||
"KEV sources used for immediate alerting (cisa, euvd)"
|
||||
)
|
||||
db.commit()
|
||||
except Exception as e:
|
||||
logger.warning(f"Could not seed enrichment settings: {e}")
|
||||
|
||||
@@ -14,6 +14,9 @@ class NotificationType(str, Enum):
|
||||
SLA_BREACH = "sla_breach"
|
||||
ASSIGNMENT = "assignment"
|
||||
NEW_VULNERABILITY = "new_vulnerability"
|
||||
# Actively exploited (CISA KEV / ENISA EUVD) AND present in our inventory —
|
||||
# its own type so the log can be filtered for the alerts that meant "now".
|
||||
KEV_ALERT = "kev_alert"
|
||||
MANUAL = "manual"
|
||||
|
||||
|
||||
|
||||
@@ -1,4 +1,6 @@
|
||||
"""Security-advisory awareness feeds (CISA KEV + configurable RSS sources)."""
|
||||
"""Security-advisory awareness feeds (CISA KEV + ENISA EUVD + RSS sources)."""
|
||||
from typing import Optional
|
||||
|
||||
from fastapi import APIRouter, Depends, Query
|
||||
from sqlalchemy.orm import Session
|
||||
|
||||
@@ -48,10 +50,56 @@ def refresh_advisory_feeds(
|
||||
@router.get("/kev-recent")
|
||||
def kev_recent(
|
||||
limit: int = Query(20, le=100, description="How many recent KEV entries to return"),
|
||||
sources: Optional[str] = Query(
|
||||
None, description="Comma list of KEV sources (cisa, euvd). Default: all"),
|
||||
in_inventory_only: bool = Query(
|
||||
False, description="Only CVEs we actually carry on an active asset"),
|
||||
db: Session = Depends(get_db),
|
||||
current_user: User = Depends(get_current_user),
|
||||
):
|
||||
"""Most recently added CISA KEV (actively-exploited) CVEs, newest first,
|
||||
annotated with whether we already have that CVE in inventory."""
|
||||
from app.services.advisory_service import get_recent_kev
|
||||
return {"items": get_recent_kev(db, limit=limit)}
|
||||
"""Most recently listed actively-exploited CVEs across every KEV source
|
||||
(CISA KEV + ENISA EUVD), newest first, annotated with inventory impact
|
||||
(open vs already-patched assets, active assets only)."""
|
||||
from app.services.kev_service import KEV_SOURCES, get_recent_kev
|
||||
picked = [s.strip() for s in sources.split(",")] if sources else None
|
||||
return {
|
||||
"items": get_recent_kev(db, limit=limit, sources=picked,
|
||||
in_inventory_only=in_inventory_only),
|
||||
"sources": [{"id": k, "label": v.label, "url": v.url}
|
||||
for k, v in KEV_SOURCES.items()],
|
||||
}
|
||||
|
||||
|
||||
@router.get("/kev-alerts")
|
||||
def kev_alert_preview(
|
||||
db: Session = Depends(get_db),
|
||||
current_user: User = Depends(get_current_user),
|
||||
):
|
||||
"""Every actively-exploited CVE with OPEN findings on active assets —
|
||||
the exact list the alert mail would carry, including the ones already
|
||||
mailed about."""
|
||||
from app.services.kev_alert_service import (SETTING_RECIPIENTS, _get_setting,
|
||||
find_alertable, get_recipients,
|
||||
get_sources, is_enabled)
|
||||
items = find_alertable(db, get_sources(db), include_already_sent=True)
|
||||
return {
|
||||
"enabled": is_enabled(db),
|
||||
# Resolved list (what actually gets mail) AND the raw setting, so the
|
||||
# editor can show "empty = fall back to the notification defaults"
|
||||
# instead of freezing today's fallback into the field.
|
||||
"recipients": [e for _, e, _ in get_recipients(db)],
|
||||
"recipients_setting": _get_setting(db, SETTING_RECIPIENTS) or "",
|
||||
"pending": sum(1 for i in items if not i["previously_notified"]),
|
||||
"items": items,
|
||||
}
|
||||
|
||||
|
||||
@router.post("/kev-alerts/run")
|
||||
# Sync def → worker threadpool; blocking SMTP must stay off the event loop.
|
||||
def run_kev_alerts_now(
|
||||
db: Session = Depends(get_db),
|
||||
current_user: User = Depends(RequireEditor),
|
||||
):
|
||||
"""Send the immediate KEV alert mail now, ignoring the enabled switch."""
|
||||
from app.services.kev_alert_service import run_kev_alerts
|
||||
return run_kev_alerts(db, force=True)
|
||||
|
||||
+34
-1
@@ -456,6 +456,10 @@ def refresh_threat_intel_enrichment():
|
||||
logger.error(f"Threat intel refresh failed: {e}")
|
||||
finally:
|
||||
db.close()
|
||||
# The catalogs just moved — check straight away rather than waiting for the
|
||||
# hourly tick. A CVE that became "actively exploited" a second ago and is
|
||||
# open on our machines is the whole point of the alert.
|
||||
kev_alert_check()
|
||||
|
||||
|
||||
def exploit_intel_nightly():
|
||||
@@ -1008,6 +1012,25 @@ def advisory_feeds_refresh():
|
||||
db.close()
|
||||
|
||||
|
||||
def kev_alert_check():
|
||||
"""Mail out actively-exploited CVEs (CISA KEV / ENISA EUVD) that have open
|
||||
findings on our own active assets. Hourly, not nightly: known exploitation
|
||||
plus confirmed presence is the one case where waiting until tomorrow is the
|
||||
wrong call. Idempotent — a CVE is mailed once, and again only if the number
|
||||
of affected systems grows."""
|
||||
from app.services.kev_alert_service import run_kev_alerts
|
||||
db = SessionLocal()
|
||||
try:
|
||||
stats = run_kev_alerts(db)
|
||||
if stats.get("emails_sent") or stats.get("emails_failed"):
|
||||
logger.info("KEV alert run: %s", stats)
|
||||
except Exception as e:
|
||||
logger.error("KEV alert run failed: %s", e)
|
||||
db.rollback()
|
||||
finally:
|
||||
db.close()
|
||||
|
||||
|
||||
def start_scheduler():
|
||||
"""Starts the background scheduler"""
|
||||
if not HAS_APSCHEDULER or scheduler is None:
|
||||
@@ -1208,9 +1231,19 @@ def start_scheduler():
|
||||
replace_existing=True,
|
||||
)
|
||||
|
||||
# Actively-exploited-and-present alerting, hourly. Offset off the hour so
|
||||
# it doesn't collide with the SLA checker.
|
||||
scheduler.add_job(
|
||||
kev_alert_check,
|
||||
trigger=CronTrigger(minute=25),
|
||||
id="kev_alert_check",
|
||||
name="KEV Alert (actively exploited + in inventory)",
|
||||
replace_existing=True,
|
||||
)
|
||||
|
||||
scheduler.start()
|
||||
logger.info(
|
||||
"Background scheduler started (SLA Breach Checker + Threat Intel Refresh + Vulnrichment Nightly + Compliance SCA Nightly + URS Nightly + Audit-Log Prune)"
|
||||
"Background scheduler started (SLA Breach Checker + Threat Intel Refresh + KEV Alert Hourly + Vulnrichment Nightly + Compliance SCA Nightly + URS Nightly + Audit-Log Prune)"
|
||||
)
|
||||
|
||||
|
||||
|
||||
@@ -1,66 +0,0 @@
|
||||
"""
|
||||
Security-advisory awareness feed.
|
||||
|
||||
Independent of asset findings: a rolling view of what's being actively
|
||||
exploited in the wild (CISA KEV), so operators see 0-days/exploited CVEs even
|
||||
when no scanner has flagged an affected asset yet. Each entry is annotated
|
||||
with whether we already have that CVE in inventory (and on how many assets).
|
||||
|
||||
Reuses the KEV catalog enrichment already fetches + caches (24h).
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import logging
|
||||
from datetime import datetime
|
||||
from typing import List, Optional
|
||||
|
||||
from sqlalchemy import func
|
||||
from sqlalchemy.orm import Session
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
def _parse_date(s) -> Optional[datetime]:
|
||||
try:
|
||||
return datetime.strptime(str(s)[:10], "%Y-%m-%d")
|
||||
except (ValueError, TypeError):
|
||||
return None
|
||||
|
||||
|
||||
def get_recent_kev(db: Session, limit: int = 20) -> List[dict]:
|
||||
"""Most recently added CISA KEV entries, newest first, annotated with our
|
||||
inventory status. Returns [] on fetch failure (awareness is best-effort)."""
|
||||
from app.services.enrichment_service import fetch_kev_catalog
|
||||
try:
|
||||
kev = fetch_kev_catalog(db)
|
||||
except Exception as e:
|
||||
logger.warning("advisory: KEV fetch failed: %s", e)
|
||||
return []
|
||||
|
||||
rows = []
|
||||
for cve, e in kev.items():
|
||||
rows.append({
|
||||
"cve_id": cve,
|
||||
"vendor": e.get("vendor"),
|
||||
"product": e.get("product"),
|
||||
"name": e.get("name"),
|
||||
"date_added": e.get("date_added"),
|
||||
"ransomware": bool(e.get("ransomware_use")),
|
||||
"description": e.get("short_description"),
|
||||
})
|
||||
rows.sort(key=lambda r: (_parse_date(r["date_added"]) or datetime.min), reverse=True)
|
||||
rows = rows[:limit]
|
||||
|
||||
# Annotate with inventory presence in one query.
|
||||
from app.models.vulnerability import Vulnerability
|
||||
cves = [r["cve_id"] for r in rows]
|
||||
counts = {}
|
||||
if cves:
|
||||
q = (db.query(Vulnerability.cve_id, func.count(func.distinct(Vulnerability.asset_id)))
|
||||
.filter(Vulnerability.cve_id.in_(cves))
|
||||
.group_by(Vulnerability.cve_id))
|
||||
counts = {cve: n for cve, n in q.all()}
|
||||
for r in rows:
|
||||
r["asset_count"] = int(counts.get(r["cve_id"], 0))
|
||||
r["in_inventory"] = r["asset_count"] > 0
|
||||
return rows
|
||||
@@ -478,6 +478,12 @@ def get_email_template(db: Session, template_key: str = "email_template_sla_brea
|
||||
elif template_key == "email_template_sla_breach_digest":
|
||||
default_subject = DEFAULT_SLA_DIGEST_SUBJECT
|
||||
default_body = DEFAULT_SLA_DIGEST_TEMPLATE
|
||||
elif template_key == "email_template_kev_alert":
|
||||
# Lives in kev_alert_service (with the rest of that feature) — imported
|
||||
# here rather than duplicated, and locally to avoid an import cycle.
|
||||
from app.services.kev_alert_service import DEFAULT_SUBJECT, DEFAULT_TEMPLATE
|
||||
default_subject = DEFAULT_SUBJECT
|
||||
default_body = DEFAULT_TEMPLATE
|
||||
|
||||
if setting and setting.value:
|
||||
try:
|
||||
|
||||
@@ -261,10 +261,21 @@ def _load_euvd_cache(db: Session) -> Optional[Dict[str, dict]]:
|
||||
return None
|
||||
|
||||
try:
|
||||
return json.loads(cache_setting.value)
|
||||
cached = json.loads(cache_setting.value)
|
||||
except json.JSONDecodeError:
|
||||
return None
|
||||
|
||||
# Caches written before the `exploited` flag existed cannot answer "is this
|
||||
# actively exploited?" for an entry that is ALSO EU-Critical — the two
|
||||
# catalogs merged into one `critical: true` and the origin was lost. Rather
|
||||
# than guess (and under-report the exploited-and-critical CVEs, which are
|
||||
# the worst ones), treat such a cache as stale and refetch once.
|
||||
if isinstance(cached, dict) and any(
|
||||
isinstance(v, dict) and "exploited" not in v for v in cached.values()):
|
||||
logger.info("EUVD: cache predates the exploited flag — refetching")
|
||||
return None
|
||||
return cached
|
||||
|
||||
|
||||
def _store_euvd_cache(db: Session, euvd_map: Dict[str, dict]) -> None:
|
||||
_set_setting(db, EUVD_CACHE_KEY, json.dumps(euvd_map), "ENISA EUVD cache (24h TTL)")
|
||||
@@ -348,6 +359,11 @@ def _parse_euvd_entry(entry: dict, is_critical: bool) -> Iterable[tuple]:
|
||||
yield cve, {
|
||||
"date_added": date_added,
|
||||
"critical": is_critical,
|
||||
# Which ENISA catalog this came from. `critical` alone cannot say
|
||||
# it: the EU-Critical list is a curated priority list, not an
|
||||
# exploitation claim, and a CVE in BOTH ends up critical=True. The
|
||||
# KEV view needs exploitation specifically, so record it.
|
||||
"exploited": not is_critical,
|
||||
"euvd_id": euvd_id,
|
||||
}
|
||||
|
||||
@@ -378,6 +394,7 @@ def _merge_euvd_entries(
|
||||
existing = merged.get(cve)
|
||||
if existing:
|
||||
existing["critical"] = existing.get("critical") or info["critical"]
|
||||
existing["exploited"] = existing.get("exploited") or info["exploited"]
|
||||
if not existing.get("date_added") and info.get("date_added"):
|
||||
existing["date_added"] = info["date_added"]
|
||||
if not existing.get("euvd_id") and info.get("euvd_id"):
|
||||
|
||||
@@ -0,0 +1,349 @@
|
||||
"""
|
||||
Immediate alerting for actively-exploited CVEs that exist in OUR environment.
|
||||
|
||||
The dashboard already showed which KEV entries are "in inventory", but seeing
|
||||
it required someone to be looking at the dashboard. A CVE that is being
|
||||
exploited in the wild AND sits unpatched on our own machines is the one case
|
||||
where waiting for the nightly digest is wrong, so this sends its own mail as
|
||||
soon as the condition is true.
|
||||
|
||||
What triggers a mail:
|
||||
* the CVE is in at least one enabled KEV source (CISA KEV / ENISA EUVD), and
|
||||
* it has at least one OPEN finding on an ACTIVE asset, and
|
||||
* we have not already alerted on it (or the number of affected assets grew).
|
||||
|
||||
Already-patched assets are never the reason for a mail and are not listed —
|
||||
they only appear as a "(N already patched)" note, because the mail is a call
|
||||
to act, not a report.
|
||||
|
||||
Settings:
|
||||
kev_alert_enabled "true"/"false" — master switch (default on)
|
||||
kev_alert_sources comma list of source ids (default: all)
|
||||
kev_alert_recipients comma/semicolon list; falls back to the notification
|
||||
default recipients (i.e. active admins)
|
||||
kev_alert_state JSON {cve: {"at": iso, "n": open_asset_count}} — what
|
||||
we already sent, so a run is idempotent
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import html
|
||||
import json
|
||||
import logging
|
||||
import os
|
||||
import re
|
||||
from datetime import datetime
|
||||
from typing import Dict, List, Optional, Sequence
|
||||
from urllib.parse import quote
|
||||
|
||||
from sqlalchemy.orm import Session
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
SETTING_ENABLED = "kev_alert_enabled"
|
||||
SETTING_SOURCES = "kev_alert_sources"
|
||||
SETTING_RECIPIENTS = "kev_alert_recipients"
|
||||
SETTING_STATE = "kev_alert_state"
|
||||
TEMPLATE_KEY = "email_template_kev_alert"
|
||||
|
||||
# Alerts per mail. Beyond this the mail stops being readable; the rest go out
|
||||
# on the next run (state is only written for what was actually sent).
|
||||
MAX_ALERTS_PER_MAIL = 25
|
||||
|
||||
DEFAULT_SUBJECT = "[TRUEVULN] ACT NOW: {{total}} actively exploited CVE(s) on {{affected_assets_count}} of your systems"
|
||||
|
||||
DEFAULT_TEMPLATE = """<!DOCTYPE html>
|
||||
<html><head><meta charset="utf-8"><style>
|
||||
body{font-family:Arial,sans-serif;color:#1f2937;max-width:820px;margin:24px auto;padding:0 16px}
|
||||
.h{background:#991b1b;color:#fff;padding:14px 18px;border-radius:6px 6px 0 0}
|
||||
.h h1{margin:0;font-size:18px}
|
||||
.h .sub{font-size:12px;opacity:.9;margin-top:4px}
|
||||
.box{background:#fff;border:1px solid #e5e7eb;border-top:none;padding:18px;border-radius:0 0 6px 6px}
|
||||
.lead{background:#fef2f2;border-left:4px solid #991b1b;padding:10px 12px;font-size:13px;margin:0 0 14px 0}
|
||||
table{width:100%;border-collapse:collapse;margin-top:8px;font-size:13px}
|
||||
th{text-align:left;background:#f3f4f6;padding:8px 6px;border-bottom:1px solid #e5e7eb;font-size:11px;text-transform:uppercase;color:#6b7280}
|
||||
td{padding:7px 6px;border-bottom:1px solid #f3f4f6;vertical-align:top}
|
||||
.src{display:inline-block;padding:1px 6px;border-radius:3px;font-size:10px;font-weight:bold;margin-right:3px;background:#fee2e2;color:#991b1b}
|
||||
.hosts{font-family:monospace;font-size:11px;color:#374151}
|
||||
.note{font-size:11px;color:#6b7280}
|
||||
.btn{display:inline-block;margin-top:14px;padding:9px 16px;background:#991b1b;color:#fff;text-decoration:none;border-radius:4px;font-size:13px}
|
||||
.foot{font-size:11px;color:#6b7280;margin-top:14px;padding-top:10px;border-top:1px solid #e5e7eb}
|
||||
</style></head><body>
|
||||
<div class="h">
|
||||
<h1>Actively exploited in the wild — and present in your environment</h1>
|
||||
<div class="sub">{{detected_at}} — recipient: {{recipient_name}}</div>
|
||||
</div>
|
||||
<div class="box">
|
||||
<p class="lead">
|
||||
<strong>{{total}} CVE(s)</strong> listed as actively exploited
|
||||
({{sources}}) currently have <strong>open findings on
|
||||
{{affected_assets_count}} active system(s)</strong>.
|
||||
Known exploitation plus confirmed presence is the top priority class in a
|
||||
risk-based process — remediate or mitigate these immediately.
|
||||
</p>
|
||||
<table>
|
||||
<thead><tr><th>CVE</th><th>Source</th><th>Listed</th><th>Open systems</th><th>Affected systems</th></tr></thead>
|
||||
<tbody>{{rows}}</tbody>
|
||||
</table>
|
||||
<a class="btn" href="{{dashboard_url}}">Open in dashboard</a>
|
||||
<div class="foot">
|
||||
Sent once per CVE, and again if the number of affected systems grows.
|
||||
Already-remediated systems are not listed. Configure recipients and sources
|
||||
under Advisories → KEV Alerting in the TrueVuln UI.
|
||||
</div>
|
||||
</div></body></html>"""
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------
|
||||
# settings
|
||||
# --------------------------------------------------------------------------
|
||||
|
||||
def _get_setting(db: Session, key: str) -> Optional[str]:
|
||||
from app.models.setting import Setting
|
||||
try:
|
||||
row = db.query(Setting).filter(Setting.key == key).first()
|
||||
except Exception:
|
||||
return None
|
||||
return row.value if row and row.value is not None else None
|
||||
|
||||
|
||||
def _set_setting(db: Session, key: str, value: str, description: str = "") -> None:
|
||||
from app.models.setting import Setting
|
||||
row = db.query(Setting).filter(Setting.key == key).first()
|
||||
if row:
|
||||
row.value = value
|
||||
else:
|
||||
db.add(Setting(key=key, value=value, description=description))
|
||||
|
||||
|
||||
def is_enabled(db: Session) -> bool:
|
||||
"""Master switch. Default ON — the feature exists because nobody wants to
|
||||
find out about an exploited CVE by reading a dashboard next week."""
|
||||
raw = (_get_setting(db, SETTING_ENABLED) or "").strip().strip('"').lower()
|
||||
return raw not in ("false", "0", "no", "off")
|
||||
|
||||
|
||||
def get_sources(db: Session) -> Optional[List[str]]:
|
||||
"""Configured KEV source ids, or None for "all"."""
|
||||
raw = _get_setting(db, SETTING_SOURCES)
|
||||
if not raw:
|
||||
return None
|
||||
ids = [s for s in re.split(r"[,;\s]+", raw.strip().strip('"')) if s]
|
||||
return ids or None
|
||||
|
||||
|
||||
def get_recipients(db: Session) -> List[tuple]:
|
||||
"""(user_id_or_None, email, display_name). The dedicated setting wins;
|
||||
otherwise the shared notification default (configured list, else admins)."""
|
||||
from app.models.user import User
|
||||
from app.services.email_service import get_default_recipients
|
||||
|
||||
raw = _get_setting(db, SETTING_RECIPIENTS)
|
||||
if raw and raw.strip():
|
||||
out, seen = [], set()
|
||||
for em in (e.strip() for e in re.split(r"[,;\s]+", raw) if e.strip()):
|
||||
if em in seen:
|
||||
continue
|
||||
seen.add(em)
|
||||
u = db.query(User).filter(User.email == em).first()
|
||||
out.append(((u.id if u else None), em, (u.username if u else em)))
|
||||
if out:
|
||||
return out
|
||||
return get_default_recipients(db)
|
||||
|
||||
|
||||
def _load_state(db: Session) -> Dict[str, dict]:
|
||||
raw = _get_setting(db, SETTING_STATE)
|
||||
if not raw:
|
||||
return {}
|
||||
try:
|
||||
data = json.loads(raw)
|
||||
return data if isinstance(data, dict) else {}
|
||||
except (json.JSONDecodeError, TypeError):
|
||||
logger.warning("kev-alert: state unreadable — treating as empty")
|
||||
return {}
|
||||
|
||||
|
||||
def _save_state(db: Session, state: Dict[str, dict]) -> None:
|
||||
_set_setting(db, SETTING_STATE, json.dumps(state),
|
||||
"KEV alert send state {cve: {at, n}} — prevents duplicate mails")
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------
|
||||
# what needs alerting
|
||||
# --------------------------------------------------------------------------
|
||||
|
||||
def find_alertable(db: Session, sources: Optional[Sequence[str]] = None,
|
||||
include_already_sent: bool = False) -> List[dict]:
|
||||
"""Actively-exploited CVEs with OPEN findings on ACTIVE assets, newest
|
||||
listing first. `include_already_sent` ignores the state (used by the
|
||||
preview endpoint so an operator can see the full picture)."""
|
||||
from app.services.kev_service import get_kev_catalog, inventory_impact
|
||||
|
||||
catalog = get_kev_catalog(db, sources)
|
||||
if not catalog:
|
||||
return []
|
||||
|
||||
impact = inventory_impact(db, catalog.keys(), with_hosts=True)
|
||||
state = {} if include_already_sent else _load_state(db)
|
||||
|
||||
out = []
|
||||
for cve, imp in impact.items():
|
||||
if not imp["open_asset_count"]:
|
||||
continue # every affected system already remediated
|
||||
seen = state.get(cve)
|
||||
# ponytail: re-alert on a GROWING open count, not on set membership.
|
||||
# A newly affected asset raises the count; an asset patched and another
|
||||
# appearing in the same window keeps it level and is missed until the
|
||||
# next change. Store asset ids here if that turns out to matter.
|
||||
if seen and imp["open_asset_count"] <= int(seen.get("n") or 0):
|
||||
continue
|
||||
entry = catalog[cve]
|
||||
out.append({**entry, **imp,
|
||||
"previously_notified": bool(seen),
|
||||
"previous_open_asset_count": int(seen.get("n") or 0) if seen else 0})
|
||||
|
||||
out.sort(key=lambda r: (str(r.get("date_added") or ""), r["open_asset_count"]),
|
||||
reverse=True)
|
||||
return out
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------
|
||||
# rendering + sending
|
||||
# --------------------------------------------------------------------------
|
||||
|
||||
def render_alert_rows(alerts: List[dict], base_url: str = "") -> str:
|
||||
"""One row per CVE: sources, listing date, open-system count, hostnames.
|
||||
Every dynamic value is HTML-escaped — catalog text and scanner-supplied
|
||||
hostnames are not ours."""
|
||||
from app.services.kev_service import KEV_SOURCES
|
||||
|
||||
rows = []
|
||||
for a in alerts:
|
||||
cve = html.escape(str(a.get("cve_id") or ""))
|
||||
link = f"{base_url}?cve_id={quote(str(a.get('cve_id') or ''))}" if base_url else ""
|
||||
cve_cell = f'<a href="{html.escape(link)}">{cve}</a>' if link else cve
|
||||
name = html.escape(str(a.get("name") or a.get("description") or ""))[:160]
|
||||
vendor_product = html.escape(
|
||||
" · ".join(x for x in (a.get("vendor"), a.get("product")) if x))
|
||||
srcs = "".join(
|
||||
f'<span class="src">'
|
||||
f'{html.escape(KEV_SOURCES[s].label if s in KEV_SOURCES else str(s))}'
|
||||
f'</span>'
|
||||
for s in a.get("sources") or [])
|
||||
if a.get("ransomware"):
|
||||
srcs += '<span class="src">RANSOMWARE</span>'
|
||||
|
||||
hosts = a.get("hosts") or []
|
||||
# Hostname plus IP: the recipient has to find the machine, and a
|
||||
# scanner hostname alone is often not enough to do that.
|
||||
host_txt = ", ".join(
|
||||
html.escape(str(h.get("hostname") or ""))
|
||||
+ (f' ({html.escape(str(h["ip_address"]))})' if h.get("ip_address") else "")
|
||||
for h in hosts)
|
||||
extra = int(a.get("hosts_truncated") or 0)
|
||||
if extra:
|
||||
host_txt += f" … +{extra} more"
|
||||
patched = int(a.get("patched_asset_count") or 0)
|
||||
note = f'<div class="note">{patched} system(s) already remediated</div>' if patched else ""
|
||||
|
||||
rows.append(
|
||||
"<tr>"
|
||||
f"<td><strong>{cve_cell}</strong>"
|
||||
f'{f"<div class=note>{name}</div>" if name else ""}'
|
||||
f'{f"<div class=note>{vendor_product}</div>" if vendor_product else ""}</td>'
|
||||
f"<td>{srcs}</td>"
|
||||
f'<td class="note">{html.escape(str(a.get("date_added") or ""))}</td>'
|
||||
f'<td><strong>{int(a.get("open_asset_count") or 0)}</strong>{note}</td>'
|
||||
f'<td class="hosts">{host_txt}</td>'
|
||||
"</tr>"
|
||||
)
|
||||
return "".join(rows)
|
||||
|
||||
|
||||
def run_kev_alerts(db: Session, force: bool = False) -> dict:
|
||||
"""Send the immediate KEV alert mail. Returns a stats dict with the same
|
||||
keys on every path, so a caller can read them without guarding.
|
||||
|
||||
force ignores the enabled switch (manual "send now" from the UI).
|
||||
"""
|
||||
from app.models.notification_log import (NotificationLog, NotificationStatus,
|
||||
NotificationType)
|
||||
from app.services.email_service import (get_email_template, render_template,
|
||||
send_email)
|
||||
from app.services.kev_service import KEV_SOURCES
|
||||
|
||||
stats = {"alerts": 0, "assets": 0, "emails_sent": 0, "emails_failed": 0,
|
||||
"recipients": 0, "skipped": None, "pending_after_run": 0, "cves": []}
|
||||
|
||||
if not force and not is_enabled(db):
|
||||
stats["skipped"] = "disabled"
|
||||
return stats
|
||||
|
||||
sources = get_sources(db)
|
||||
alerts = find_alertable(db, sources)
|
||||
if not alerts:
|
||||
stats["skipped"] = "nothing new"
|
||||
return stats
|
||||
|
||||
sent_alerts = alerts[:MAX_ALERTS_PER_MAIL]
|
||||
stats["alerts"] = len(sent_alerts)
|
||||
stats["pending_after_run"] = len(alerts) - len(sent_alerts)
|
||||
stats["assets"] = sum(a["open_asset_count"] for a in sent_alerts)
|
||||
stats["cves"] = [a["cve_id"] for a in sent_alerts]
|
||||
|
||||
recipients = get_recipients(db)
|
||||
stats["recipients"] = len(recipients)
|
||||
if not recipients:
|
||||
stats["skipped"] = "no recipients"
|
||||
return stats
|
||||
|
||||
dashboard_url = os.getenv("DASHBOARD_URL", "http://localhost:3000").rstrip("/") + "/vulnerabilities"
|
||||
src_ids = sources or list(KEV_SOURCES)
|
||||
source_labels = ", ".join(KEV_SOURCES[s].label for s in KEV_SOURCES if s in src_ids)
|
||||
subject_tpl, body_tpl = get_email_template(db, TEMPLATE_KEY)
|
||||
rows = render_alert_rows(sent_alerts, base_url=dashboard_url)
|
||||
|
||||
for user_id, email, username in recipients:
|
||||
variables = {
|
||||
"total": str(len(sent_alerts)),
|
||||
"affected_assets_count": str(stats["assets"]),
|
||||
"sources": source_labels,
|
||||
"detected_at": datetime.now().strftime("%Y-%m-%d %H:%M"),
|
||||
"recipient_name": username or email,
|
||||
"recipient_email": email,
|
||||
"dashboard_url": dashboard_url,
|
||||
"rows": rows,
|
||||
}
|
||||
subject = render_template(subject_tpl, variables)
|
||||
ok, err = send_email(db, email, subject, render_template(body_tpl, variables))
|
||||
db.add(NotificationLog(
|
||||
user_id=user_id,
|
||||
notification_type=NotificationType.KEV_ALERT,
|
||||
sent_at=datetime.now(),
|
||||
subject=subject[:500],
|
||||
recipient_email=email,
|
||||
status=NotificationStatus.SENT if ok else NotificationStatus.FAILED,
|
||||
message_body=f"{len(sent_alerts)} actively exploited CVE(s) on "
|
||||
f"{stats['assets']} system(s): "
|
||||
+ ", ".join(a["cve_id"] for a in sent_alerts[:20]),
|
||||
error_message=None if ok else err,
|
||||
))
|
||||
if ok:
|
||||
stats["emails_sent"] += 1
|
||||
else:
|
||||
stats["emails_failed"] += 1
|
||||
|
||||
# Only remember what at least one recipient actually received — otherwise a
|
||||
# broken SMTP config would silently swallow the one alert that mattered.
|
||||
if stats["emails_sent"]:
|
||||
state = _load_state(db)
|
||||
now = datetime.now().isoformat()
|
||||
for a in sent_alerts:
|
||||
state[a["cve_id"]] = {"at": now, "n": a["open_asset_count"]}
|
||||
_save_state(db, state)
|
||||
|
||||
db.commit()
|
||||
logger.info("kev-alert: %d CVE(s) on %d asset(s) → %d mail(s), %d failed",
|
||||
stats["alerts"], stats["assets"], stats["emails_sent"],
|
||||
stats["emails_failed"])
|
||||
return stats
|
||||
@@ -0,0 +1,242 @@
|
||||
"""
|
||||
Unified "actively exploited" (KEV) catalog across every source we have.
|
||||
|
||||
CISA KEV was the only source the dashboard and the Advisories page showed,
|
||||
even though the ENISA EUVD exploited catalog was already fetched and cached
|
||||
for enrichment. This module merges them into ONE catalog keyed by CVE, each
|
||||
entry carrying which sources listed it, so a CVE known to both is one row with
|
||||
two badges instead of two rows.
|
||||
|
||||
Sources:
|
||||
cisa — CISA Known Exploited Vulnerabilities catalog
|
||||
euvd — ENISA EUVD, /api/search?exploited=true (the "ENISA KEV")
|
||||
|
||||
Inventory annotation lives here too, and it is deliberately stricter than the
|
||||
raw `count(distinct asset_id)` the Advisories page used to do: assets that are
|
||||
INACTIVE/DECOMMISSIONED are excluded (same rule the Vulnerabilities list
|
||||
applies), and open findings are counted apart from already-patched ones. That
|
||||
is what made the badge say 64 while the filtered vulnerability view listed 62.
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import logging
|
||||
from datetime import datetime
|
||||
from typing import Dict, Iterable, List, NamedTuple, Optional, Sequence
|
||||
|
||||
from sqlalchemy import func
|
||||
from sqlalchemy.orm import Session
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
class KevSource(NamedTuple):
|
||||
label: str
|
||||
url: str
|
||||
|
||||
|
||||
# id → source. Order is display order.
|
||||
KEV_SOURCES: Dict[str, KevSource] = {
|
||||
"cisa": KevSource("CISA KEV",
|
||||
"https://www.cisa.gov/known-exploited-vulnerabilities-catalog"),
|
||||
"euvd": KevSource("ENISA EUVD", "https://euvd.enisa.europa.eu/"),
|
||||
}
|
||||
ALL_SOURCES = tuple(KEV_SOURCES)
|
||||
|
||||
# Hosts listed per CVE in an alert mail. A CVE on 800 machines must not
|
||||
# produce an 800-row table nobody reads (and some MTAs would truncate).
|
||||
MAX_HOSTS_PER_CVE = 50
|
||||
|
||||
|
||||
def euvd_is_exploited(entry: dict) -> bool:
|
||||
"""Is this EUVD entry from the *exploited* catalog?
|
||||
|
||||
Entries without the flag come from a cache written before it existed, and
|
||||
for those the answer is genuinely unknown: a CVE in both ENISA catalogs
|
||||
merged down to `critical=True` and lost where it came from. `_load_euvd_cache`
|
||||
treats such a cache as stale so this is a one-fetch window, but the
|
||||
stale-cache fallback (every ENISA call failed) can still reach here — and
|
||||
then under-reporting an exploited CVE is the worse error, so an unflagged
|
||||
entry counts as exploited.
|
||||
"""
|
||||
if "exploited" in entry:
|
||||
return bool(entry["exploited"])
|
||||
return True
|
||||
|
||||
|
||||
def _parse_date(s) -> Optional[datetime]:
|
||||
try:
|
||||
return datetime.strptime(str(s)[:10], "%Y-%m-%d")
|
||||
except (ValueError, TypeError):
|
||||
return None
|
||||
|
||||
|
||||
def _normalize_sources(sources: Optional[Sequence[str]]) -> tuple:
|
||||
"""Requested source ids → the known ones, in display order. Empty/unknown
|
||||
input means "everything" — a typo must not silently mute the catalog."""
|
||||
if not sources:
|
||||
return ALL_SOURCES
|
||||
wanted = {str(s).strip().lower() for s in sources}
|
||||
picked = tuple(s for s in ALL_SOURCES if s in wanted)
|
||||
return picked or ALL_SOURCES
|
||||
|
||||
|
||||
def get_kev_catalog(db: Session, sources: Optional[Sequence[str]] = None) -> Dict[str, dict]:
|
||||
"""{CVE-ID: entry} of everything known to be actively exploited.
|
||||
|
||||
Per-source fetch failures are logged and skipped: one dead catalog must not
|
||||
blank out the other. Entry shape:
|
||||
{sources: ["cisa", "euvd"], date_added, source_dates, vendor, product,
|
||||
name, description, ransomware, euvd_id}
|
||||
`date_added` is the LATEST date any source listed it, because this feeds a
|
||||
"latest additions" view: a CVE CISA listed in 2023 and ENISA flagged
|
||||
yesterday is news yesterday. Per-source dates stay in `source_dates`.
|
||||
"""
|
||||
picked = _normalize_sources(sources)
|
||||
out: Dict[str, dict] = {}
|
||||
|
||||
def _touch(cve: str, source: str, date_added) -> dict:
|
||||
e = out.get(cve)
|
||||
if e is None:
|
||||
e = out[cve] = {"cve_id": cve, "sources": [], "source_dates": {},
|
||||
"date_added": None,
|
||||
"vendor": None, "product": None, "name": None,
|
||||
"description": None, "ransomware": False, "euvd_id": None}
|
||||
if source not in e["sources"]:
|
||||
e["sources"].append(source)
|
||||
if date_added:
|
||||
day = str(date_added)[:10]
|
||||
e["source_dates"][source] = day
|
||||
old, new = _parse_date(e["date_added"]), _parse_date(day)
|
||||
if new and (old is None or new > old):
|
||||
e["date_added"] = day
|
||||
return e
|
||||
|
||||
if "cisa" in picked:
|
||||
from app.services.enrichment_service import fetch_kev_catalog
|
||||
try:
|
||||
for cve, k in fetch_kev_catalog(db).items():
|
||||
e = _touch(cve.upper(), "cisa", k.get("date_added"))
|
||||
e["vendor"] = e["vendor"] or k.get("vendor")
|
||||
e["product"] = e["product"] or k.get("product")
|
||||
e["name"] = e["name"] or k.get("name")
|
||||
e["description"] = e["description"] or k.get("short_description")
|
||||
e["ransomware"] = e["ransomware"] or bool(k.get("ransomware_use"))
|
||||
except Exception as ex:
|
||||
logger.warning("kev: CISA catalog unavailable: %s", ex)
|
||||
|
||||
if "euvd" in picked:
|
||||
from app.services.enrichment_service import fetch_euvd_catalogs
|
||||
try:
|
||||
for cve, v in fetch_euvd_catalogs(db).items():
|
||||
if not euvd_is_exploited(v):
|
||||
continue # EU-Critical without exploitation is not a KEV entry
|
||||
e = _touch(cve.upper(), "euvd", v.get("date_added"))
|
||||
e["euvd_id"] = e["euvd_id"] or v.get("euvd_id")
|
||||
except Exception as ex:
|
||||
logger.warning("kev: ENISA EUVD catalog unavailable: %s", ex)
|
||||
|
||||
return out
|
||||
|
||||
|
||||
def inventory_impact(db: Session, cve_ids: Iterable[str],
|
||||
with_hosts: bool = False) -> Dict[str, dict]:
|
||||
"""{CVE: impact} for CVEs we actually carry, over ACTIVE assets only.
|
||||
|
||||
open_asset_count — assets where the finding is still open/unverified/
|
||||
patch-failed. This is the number that means "act now".
|
||||
patched_asset_count — assets that already closed it (patched / accepted /
|
||||
false-positive / deferred). Informational.
|
||||
asset_count — distinct assets in either group.
|
||||
|
||||
Findings on INACTIVE or DECOMMISSIONED assets are excluded, matching the
|
||||
Vulnerabilities list; orphan findings (asset_id NULL) carry no asset and so
|
||||
contribute to no count.
|
||||
"""
|
||||
from app.models.asset import Asset, AssetStatus
|
||||
from app.models.vulnerability import Vulnerability, VulnerabilityStatus
|
||||
|
||||
cves = sorted({str(c).upper() for c in cve_ids if c})
|
||||
if not cves:
|
||||
return {}
|
||||
|
||||
OPEN = (VulnerabilityStatus.open, VulnerabilityStatus.pending_verification,
|
||||
VulnerabilityStatus.patch_failed)
|
||||
|
||||
q = (db.query(Vulnerability.cve_id, Vulnerability.status,
|
||||
Asset.id, Asset.hostname, Asset.ip_address)
|
||||
.join(Asset, Vulnerability.asset_id == Asset.id)
|
||||
.filter(Vulnerability.cve_id.in_(cves),
|
||||
Asset.status == AssetStatus.ACTIVE))
|
||||
|
||||
out: Dict[str, dict] = {}
|
||||
for cve, status, asset_id, hostname, ip in q.all():
|
||||
cve = str(cve).upper()
|
||||
e = out.get(cve)
|
||||
if e is None:
|
||||
e = out[cve] = {"open": set(), "patched": set(), "hosts": {}}
|
||||
if status in OPEN:
|
||||
e["open"].add(asset_id)
|
||||
if with_hosts and asset_id not in e["hosts"]:
|
||||
e["hosts"][asset_id] = {"hostname": hostname or f"asset-{asset_id}",
|
||||
"ip_address": ip}
|
||||
else:
|
||||
e["patched"].add(asset_id)
|
||||
|
||||
result: Dict[str, dict] = {}
|
||||
for cve, e in out.items():
|
||||
open_ids, patched_ids = e["open"], e["patched"]
|
||||
# An asset with both an open and a closed row for the same CVE is still
|
||||
# affected — count it as open, never twice.
|
||||
patched_only = patched_ids - open_ids
|
||||
hosts = sorted(e["hosts"].values(), key=lambda h: h["hostname"].lower())
|
||||
result[cve] = {
|
||||
"open_asset_count": len(open_ids),
|
||||
"patched_asset_count": len(patched_only),
|
||||
"asset_count": len(open_ids | patched_ids),
|
||||
"in_inventory": bool(open_ids or patched_ids),
|
||||
"hosts": hosts[:MAX_HOSTS_PER_CVE],
|
||||
"hosts_truncated": max(0, len(hosts) - MAX_HOSTS_PER_CVE),
|
||||
}
|
||||
return result
|
||||
|
||||
|
||||
def get_recent_kev(db: Session, limit: int = 20,
|
||||
sources: Optional[Sequence[str]] = None,
|
||||
in_inventory_only: bool = False) -> List[dict]:
|
||||
"""Most recently added actively-exploited CVEs, newest first, annotated
|
||||
with inventory impact. Returns [] when every catalog is unreachable —
|
||||
awareness is best-effort and must not break the page.
|
||||
|
||||
in_inventory_only keeps the ones with an OPEN finding, not merely a row:
|
||||
a CVE we already remediated everywhere is not something to act on, so it
|
||||
has no business in an "affects me" view.
|
||||
"""
|
||||
catalog = get_kev_catalog(db, sources)
|
||||
if not catalog:
|
||||
return []
|
||||
|
||||
rows = sorted(catalog.values(),
|
||||
key=lambda r: (_parse_date(r["date_added"]) or datetime.min),
|
||||
reverse=True)
|
||||
|
||||
if in_inventory_only:
|
||||
# Impact for the WHOLE catalog: the newest N are usually not the ones
|
||||
# we own, so slicing first would return an empty list while the
|
||||
# environment is on fire over a 2023 CVE.
|
||||
impact = inventory_impact(db, [r["cve_id"] for r in rows])
|
||||
rows = [r for r in rows
|
||||
if impact.get(r["cve_id"], {}).get("open_asset_count")]
|
||||
else:
|
||||
rows = rows[:limit]
|
||||
impact = inventory_impact(db, [r["cve_id"] for r in rows])
|
||||
rows = rows[:limit]
|
||||
|
||||
out = []
|
||||
for r in rows:
|
||||
imp = impact.get(r["cve_id"], {})
|
||||
out.append({**r,
|
||||
"source_labels": [KEV_SOURCES[s][0] for s in r["sources"]],
|
||||
"asset_count": imp.get("asset_count", 0),
|
||||
"open_asset_count": imp.get("open_asset_count", 0),
|
||||
"patched_asset_count": imp.get("patched_asset_count", 0),
|
||||
"in_inventory": imp.get("in_inventory", False)})
|
||||
return out
|
||||
@@ -12,6 +12,13 @@ type FeedCfg = { id: string; name: string; url: string; enabled: boolean };
|
||||
|
||||
export default function AdvisoriesPage() {
|
||||
const [kev, setKev] = useState<any[]>([]);
|
||||
// KEV view controls — '' = every source (CISA KEV + ENISA EUVD merged).
|
||||
const [kevSource, setKevSource] = useState('');
|
||||
const [onlyMine, setOnlyMine] = useState(false);
|
||||
const [alerts, setAlerts] = useState<any>(null);
|
||||
const [sending, setSending] = useState(false);
|
||||
const [alertCfg, setAlertCfg] = useState({ enabled: true, recipients: '' });
|
||||
const [alertCfgMsg, setAlertCfgMsg] = useState('');
|
||||
const [feeds, setFeeds] = useState<Feed[]>([]);
|
||||
const [fetchedAt, setFetchedAt] = useState<string | null>(null);
|
||||
const [loading, setLoading] = useState(true);
|
||||
@@ -27,14 +34,27 @@ export default function AdvisoriesPage() {
|
||||
const [maxItems, setMaxItems] = useState('30');
|
||||
const [limits, setLimits] = useState({ def: 30, min: 1, max: 500 });
|
||||
|
||||
const kevUrl = () => {
|
||||
const p = new URLSearchParams({ limit: '15' });
|
||||
if (kevSource) p.set('sources', kevSource);
|
||||
if (onlyMine) p.set('in_inventory_only', 'true');
|
||||
return `/api/v1/advisories/kev-recent?${p}`;
|
||||
};
|
||||
|
||||
const load = async () => {
|
||||
try {
|
||||
const [k, f, me] = await Promise.all([
|
||||
api.get('/api/v1/advisories/kev-recent?limit=15').catch(() => ({ data: { items: [] } })),
|
||||
const [k, f, me, al] = await Promise.all([
|
||||
api.get(kevUrl()).catch(() => ({ data: { items: [] } })),
|
||||
api.get('/api/v1/advisories/feeds').catch(() => ({ data: { feeds: [], fetched_at: null } })),
|
||||
api.get('/auth/me').catch(() => ({ data: {} })),
|
||||
api.get('/api/v1/advisories/kev-alerts').catch(() => ({ data: null })),
|
||||
]);
|
||||
setKev(k.data?.items || []);
|
||||
setAlerts(al.data || null);
|
||||
if (al.data) setAlertCfg({
|
||||
enabled: al.data.enabled !== false,
|
||||
recipients: al.data.recipients_setting || '',
|
||||
});
|
||||
setFeeds(f.data?.feeds || []);
|
||||
setFetchedAt(f.data?.fetched_at || null);
|
||||
setUserRole(me.data?.role || '');
|
||||
@@ -50,6 +70,43 @@ export default function AdvisoriesPage() {
|
||||
};
|
||||
useEffect(() => { load(); }, []);
|
||||
|
||||
// Source / inventory filter changes refetch only the KEV list — the RSS
|
||||
// feeds and the alert panel are unaffected by them.
|
||||
useEffect(() => {
|
||||
let stale = false;
|
||||
api.get(kevUrl())
|
||||
.then((r) => { if (!stale) setKev(r.data?.items || []); })
|
||||
.catch(() => { });
|
||||
return () => { stale = true; };
|
||||
}, [kevSource, onlyMine]);
|
||||
|
||||
const saveAlertCfg = async () => {
|
||||
try {
|
||||
await api.put('/api/v1/settings/kev_alert_enabled', { value: alertCfg.enabled ? 'true' : 'false' });
|
||||
await api.put('/api/v1/settings/kev_alert_recipients', { value: alertCfg.recipients.trim() });
|
||||
setAlertCfgMsg('Saved.');
|
||||
await load();
|
||||
} catch (e: any) {
|
||||
setAlertCfgMsg(e?.response?.data?.detail || 'Save failed');
|
||||
}
|
||||
};
|
||||
|
||||
const sendAlerts = async () => {
|
||||
if (!confirm('Send the KEV alert mail now to the configured recipients?')) return;
|
||||
setSending(true);
|
||||
try {
|
||||
const r = await api.post('/api/v1/advisories/kev-alerts/run');
|
||||
const d = r.data || {};
|
||||
alert(d.skipped
|
||||
? `Nothing sent: ${d.skipped}`
|
||||
: `Sent ${d.emails_sent} mail(s) covering ${d.alerts} CVE(s) on ${d.assets} system(s).`
|
||||
+ (d.emails_failed ? ` ${d.emails_failed} failed — check the notification log.` : ''));
|
||||
await load();
|
||||
} catch (e: any) {
|
||||
alert(e?.response?.data?.detail || 'Sending failed');
|
||||
} finally { setSending(false); }
|
||||
};
|
||||
|
||||
const refresh = async () => {
|
||||
setRefreshing(true);
|
||||
try { await api.post('/api/v1/advisories/feeds/refresh'); await load(); }
|
||||
@@ -143,21 +200,52 @@ export default function AdvisoriesPage() {
|
||||
</div>
|
||||
)}
|
||||
|
||||
{/* CISA KEV — actively exploited */}
|
||||
{/* Actively exploited — every KEV source we have, merged. A CVE
|
||||
listed by both carries both badges instead of appearing twice. */}
|
||||
<div className="bg-white border border-gray-200 shadow-sm rounded-sm mb-6">
|
||||
<div className="px-4 py-3 border-b border-gray-100 bg-red-50">
|
||||
<h3 className="text-sm font-bold font-mono text-red-800">CISA KEV — Actively Exploited (latest additions)</h3>
|
||||
<div className="px-4 py-3 border-b border-gray-100 bg-red-50 flex items-center justify-between gap-3 flex-wrap">
|
||||
<h3 className="text-sm font-bold font-mono text-red-800">
|
||||
Actively Exploited — KEV (latest additions)
|
||||
<span className="ml-2 font-normal text-red-700/70">CISA KEV · ENISA EUVD</span>
|
||||
</h3>
|
||||
<div className="flex items-center gap-3 text-xs font-mono">
|
||||
<label className="flex items-center gap-1 text-red-800">
|
||||
<input type="checkbox" checked={onlyMine} onChange={(e) => setOnlyMine(e.target.checked)}
|
||||
className="h-3.5 w-3.5 rounded border-red-300 text-red-700" />
|
||||
Only in my inventory
|
||||
</label>
|
||||
<select value={kevSource} onChange={(e) => setKevSource(e.target.value)}
|
||||
className="rounded-md border-gray-300 text-xs font-mono h-7 py-0 pl-2 pr-7">
|
||||
<option value="">All sources</option>
|
||||
<option value="cisa">CISA KEV only</option>
|
||||
<option value="euvd">ENISA EUVD only</option>
|
||||
</select>
|
||||
</div>
|
||||
</div>
|
||||
<ul className="divide-y divide-gray-100">
|
||||
{kev.length === 0 && <li className="px-4 py-3 text-sm text-gray-400 font-mono">No KEV data yet — run Refresh Threat Intel.</li>}
|
||||
{kev.length === 0 && <li className="px-4 py-3 text-sm text-gray-400 font-mono">
|
||||
{onlyMine ? 'Nothing actively exploited is open in your inventory.' : 'No KEV data yet — run Refresh Threat Intel.'}
|
||||
</li>}
|
||||
{kev.map((k: any, i: number) => (
|
||||
<li key={i} className="px-4 py-2 flex items-center justify-between gap-3 text-sm">
|
||||
<div className="min-w-0">
|
||||
<a href={`/vulnerabilities?cve_id=${k.cve_id}`} className="font-mono font-semibold text-truevuln-blue hover:underline">{k.cve_id}</a>
|
||||
<span className="ml-2 text-gray-600">{k.vulnerability_name || k.short_description || ''}</span>
|
||||
{(k.sources || []).map((s: string) => (
|
||||
<span key={s} title={s === 'euvd' ? `ENISA EUVD — exploited${k.euvd_id ? ` (${k.euvd_id})` : ''}` : 'CISA KEV'}
|
||||
className={`ml-1 rounded px-1 py-0.5 text-[9px] font-bold font-mono uppercase ${s === 'euvd' ? 'bg-indigo-100 text-indigo-700' : 'bg-red-100 text-red-700'}`}>
|
||||
{s === 'euvd' ? 'EUVD' : 'CISA'}
|
||||
</span>
|
||||
))}
|
||||
{k.ransomware && <span title="Known ransomware campaign use" className="ml-1">🔒</span>}
|
||||
<span className="ml-2 text-gray-600">{k.name || k.description || [k.vendor, k.product].filter(Boolean).join(' · ')}</span>
|
||||
</div>
|
||||
<div className="flex items-center gap-2 flex-none text-xs font-mono">
|
||||
{k.in_inventory && <span className="rounded px-1.5 py-0.5 bg-red-100 text-red-700 font-bold">IN INVENTORY</span>}
|
||||
{k.in_inventory && (
|
||||
<span className={`rounded px-1.5 py-0.5 font-bold ${k.open_asset_count > 0 ? 'bg-red-100 text-red-700' : 'bg-green-100 text-green-700'}`}
|
||||
title={`${k.open_asset_count} asset(s) still open, ${k.patched_asset_count} already remediated`}>
|
||||
{k.open_asset_count > 0 ? `IN INVENTORY · ${k.open_asset_count}` : 'REMEDIATED'}
|
||||
</span>
|
||||
)}
|
||||
<span className="text-gray-400">{k.date_added || ''}</span>
|
||||
</div>
|
||||
</li>
|
||||
@@ -165,6 +253,74 @@ export default function AdvisoriesPage() {
|
||||
</ul>
|
||||
</div>
|
||||
|
||||
{/* Immediate alerting: exploited in the wild AND open here. */}
|
||||
{alerts && (
|
||||
<div className="bg-white border border-gray-200 shadow-sm rounded-sm mb-6">
|
||||
<div className="px-4 py-3 border-b border-gray-100 bg-gray-50 flex items-center justify-between gap-3 flex-wrap">
|
||||
<div>
|
||||
<h3 className="text-sm font-bold font-mono text-gray-900">
|
||||
KEV Alerting — actively exploited & open in your environment
|
||||
</h3>
|
||||
<p className="text-xs text-gray-500 mt-0.5">
|
||||
{alerts.enabled ? 'Enabled' : 'Disabled'} · checked hourly ·
|
||||
{' '}{alerts.pending} pending · recipients: {(alerts.recipients || []).join(', ') || 'none configured'}
|
||||
</p>
|
||||
</div>
|
||||
{canEdit && (
|
||||
<button onClick={sendAlerts} disabled={sending}
|
||||
className="rounded-md bg-red-700 px-3 py-2 text-xs font-semibold text-white shadow-sm hover:bg-red-800 disabled:opacity-50">
|
||||
{sending ? 'Sending…' : 'Send alert mail now'}
|
||||
</button>
|
||||
)}
|
||||
</div>
|
||||
|
||||
{/* Admin config — the alert mail points recipients here. */}
|
||||
{userRole === 'admin' && (
|
||||
<div className="px-4 py-3 border-b border-gray-100 bg-gray-50/50 flex items-center gap-3 flex-wrap text-xs font-mono">
|
||||
<label className="flex items-center gap-1">
|
||||
<input type="checkbox" checked={alertCfg.enabled}
|
||||
onChange={(e) => setAlertCfg({ ...alertCfg, enabled: e.target.checked })}
|
||||
className="h-3.5 w-3.5 rounded border-gray-300 text-red-700" />
|
||||
Alerting enabled
|
||||
</label>
|
||||
<label className="flex items-center gap-1 flex-1 min-w-[280px]">
|
||||
Recipients
|
||||
<input type="text" value={alertCfg.recipients}
|
||||
onChange={(e) => setAlertCfg({ ...alertCfg, recipients: e.target.value })}
|
||||
placeholder="empty = notification defaults (admins)"
|
||||
className="flex-1 rounded-md border-gray-300 text-xs font-mono h-7 px-2" />
|
||||
</label>
|
||||
<button onClick={saveAlertCfg}
|
||||
className="px-3 py-1.5 border border-gray-300 rounded-md hover:bg-gray-50">Save</button>
|
||||
{alertCfgMsg && <span className="text-gray-500">{alertCfgMsg}</span>}
|
||||
</div>
|
||||
)}
|
||||
<ul className="divide-y divide-gray-100">
|
||||
{(alerts.items || []).length === 0 && (
|
||||
<li className="px-4 py-3 text-sm text-gray-400 font-mono">
|
||||
Nothing actively exploited is open on an active asset. 🎉
|
||||
</li>
|
||||
)}
|
||||
{(alerts.items || []).map((a: any) => (
|
||||
<li key={a.cve_id} className="px-4 py-2 text-sm">
|
||||
<div className="flex items-center justify-between gap-3">
|
||||
<a href={`/vulnerabilities?cve_id=${a.cve_id}`} className="font-mono font-semibold text-truevuln-blue hover:underline">{a.cve_id}</a>
|
||||
<span className="flex-none text-xs font-mono">
|
||||
<span className="rounded px-1.5 py-0.5 bg-red-100 text-red-700 font-bold">{a.open_asset_count} OPEN</span>
|
||||
{a.patched_asset_count > 0 && <span className="ml-1 text-gray-400">{a.patched_asset_count} remediated</span>}
|
||||
{a.previously_notified && <span className="ml-2 text-gray-400">already alerted</span>}
|
||||
</span>
|
||||
</div>
|
||||
<p className="text-xs text-gray-500 font-mono mt-0.5 truncate">
|
||||
{(a.hosts || []).map((h: any) => h.hostname).join(', ')}
|
||||
{a.hosts_truncated > 0 && ` … +${a.hosts_truncated} more`}
|
||||
</p>
|
||||
</li>
|
||||
))}
|
||||
</ul>
|
||||
</div>
|
||||
)}
|
||||
|
||||
{/* RSS feeds */}
|
||||
{feeds.filter(f => f.enabled).map((f) => (
|
||||
<div key={f.id} className="bg-white border border-gray-200 shadow-sm rounded-sm mb-4">
|
||||
|
||||
+40
-12
@@ -248,7 +248,7 @@ export default function Dashboard() {
|
||||
useEffect(() => {
|
||||
const fetchData = async () => {
|
||||
try {
|
||||
const [statsRes, vulnsRes, criticalRes, eolRes, mobileRes, kevRes, schedRes, compRes, ursRes] = await Promise.all([
|
||||
const [statsRes, vulnsRes, criticalRes, eolRes, mobileRes, kevRes, kevMineRes, schedRes, compRes, ursRes] = await Promise.all([
|
||||
api.get('/api/v1/vulnerabilities/reports/dashboard'),
|
||||
// Newly Published: sort by published_date desc. distinct_cve=true
|
||||
// collapses per-asset duplicates server-side so we reliably get 10
|
||||
@@ -287,10 +287,16 @@ export default function Dashboard() {
|
||||
// ≥1y 8 → EOL-soon 5.5 → end-of-active-support 3).
|
||||
api.get('/api/v1/vulnerabilities?limit=15&status=active&finding_type=mobile&sort_by=cvss&sort_order=desc')
|
||||
.catch(() => ({ data: { items: [] } })),
|
||||
// Advisory feed: recently-added CISA KEV (actively exploited in the
|
||||
// wild), independent of whether we have an affected asset yet.
|
||||
// Advisory feed: actively exploited in the wild (CISA KEV + ENISA
|
||||
// EUVD), independent of whether we have an affected asset yet.
|
||||
api.get('/api/v1/advisories/kev-recent?limit=12')
|
||||
.catch(() => ({ data: { items: [] } })),
|
||||
// ...plus the ones open in OUR environment. Separate call because
|
||||
// they are pinned to the top and sorting by listing date would bury
|
||||
// them: a 2023 KEV still open on 265 hosts never makes a newest-12
|
||||
// list, and that is exactly the row somebody has to act on.
|
||||
api.get('/api/v1/advisories/kev-recent?limit=12&in_inventory_only=true')
|
||||
.catch(() => ({ data: { items: [] } })),
|
||||
api.get('/api/v1/scans/schedules').catch(() => ({ data: [] })),
|
||||
api.get('/api/v1/compliance/summary').catch(() => ({ data: null })),
|
||||
api.get('/api/v1/compliance/urs?limit=200').catch(() => ({ data: [] })),
|
||||
@@ -387,7 +393,14 @@ export default function Dashboard() {
|
||||
}
|
||||
setMobileVulns(recentMobile);
|
||||
|
||||
setKevAdvisories((kevRes.data && kevRes.data.items) || []);
|
||||
// Affected-here first, then the newest listings, no CVE twice.
|
||||
const kevMine = (kevMineRes.data && kevMineRes.data.items) || [];
|
||||
const kevSeen = new Set(kevMine.map((k: any) => k.cve_id));
|
||||
setKevAdvisories([
|
||||
...kevMine,
|
||||
...(((kevRes.data && kevRes.data.items) || [])
|
||||
.filter((k: any) => !kevSeen.has(k.cve_id))),
|
||||
].slice(0, 12));
|
||||
|
||||
const schedules = Array.isArray(schedRes.data) ? schedRes.data : [];
|
||||
setActiveScheduleCount(schedules.filter((s: any) => s.enabled).length);
|
||||
@@ -827,15 +840,16 @@ export default function Dashboard() {
|
||||
viewAllHref: '/vulnerabilities?finding_type=mobile&sort_by=cvss&sort_order=desc',
|
||||
})}
|
||||
|
||||
{/* Advisory feed — CISA KEV (actively exploited in the wild), independent
|
||||
of asset findings. "In inventory" badge when we already track it. */}
|
||||
{/* Advisory feed — actively exploited in the wild (CISA KEV + ENISA
|
||||
EUVD), independent of asset findings. "In inventory" counts OPEN
|
||||
findings on active assets, so it matches the vulnerability view. */}
|
||||
<div className="bg-white border border-gray-200 shadow-sm rounded-sm overflow-hidden flex flex-col">
|
||||
<div className="p-4 border-b border-gray-200 flex justify-between items-baseline bg-gray-50/50">
|
||||
<div>
|
||||
<h3 className="text-base font-bold text-gray-900 font-mono">Actively Exploited · CISA KEV</h3>
|
||||
<p className="text-[11px] text-gray-500 font-mono mt-0.5">Newly added known-exploited CVEs · 🔒 = ransomware use</p>
|
||||
<h3 className="text-base font-bold text-gray-900 font-mono">Actively Exploited · KEV</h3>
|
||||
<p className="text-[11px] text-gray-500 font-mono mt-0.5">CISA KEV + ENISA EUVD · 🔒 = ransomware use</p>
|
||||
</div>
|
||||
<a href="https://www.cisa.gov/known-exploited-vulnerabilities-catalog" target="_blank" rel="noreferrer"
|
||||
<a href="/advisories"
|
||||
className="text-truevuln-blue text-[10px] font-bold uppercase tracking-wider font-mono hover:text-blue-700">View All ></a>
|
||||
</div>
|
||||
<div className="overflow-x-auto flex-1">
|
||||
@@ -852,13 +866,27 @@ export default function Dashboard() {
|
||||
{k.ransomware && <span title="Known ransomware campaign use" className="ml-1">🔒</span>}
|
||||
</td>
|
||||
<td className="px-3 py-2 text-xs text-gray-600 truncate max-w-[180px]" title={`${k.vendor || ''} ${k.product || ''}`}>
|
||||
{[k.vendor, k.product].filter(Boolean).join(' · ')}
|
||||
{[k.vendor, k.product].filter(Boolean).join(' · ') || k.name || ''}
|
||||
</td>
|
||||
<td className="px-3 py-2 whitespace-nowrap">
|
||||
{(k.sources || []).map((s: string) => (
|
||||
<span key={s} title={s === 'euvd' ? 'ENISA EUVD — exploited' : 'CISA KEV'}
|
||||
className={`mr-1 inline-flex items-center rounded-sm px-1 py-0.5 text-[9px] font-bold uppercase ${s === 'euvd' ? 'bg-indigo-100 text-indigo-700' : 'bg-red-100 text-red-700'}`}>
|
||||
{s === 'euvd' ? 'EUVD' : 'CISA'}
|
||||
</span>
|
||||
))}
|
||||
</td>
|
||||
<td className="px-3 py-2 whitespace-nowrap text-[11px] text-gray-400 font-mono">{k.date_added}</td>
|
||||
<td className="px-3 py-2 whitespace-nowrap text-right">
|
||||
{k.in_inventory ? (
|
||||
<span className="inline-flex items-center rounded-sm border border-red-200 bg-red-50 px-1.5 py-0.5 text-[10px] font-bold uppercase text-red-700"
|
||||
title={`Present on ${k.asset_count} asset(s)`}>In inventory · {k.asset_count}</span>
|
||||
// Open findings on ACTIVE assets — the number that means "act".
|
||||
// Already-remediated assets are named in the tooltip, not counted.
|
||||
<span className={`inline-flex items-center rounded-sm border px-1.5 py-0.5 text-[10px] font-bold uppercase ${k.open_asset_count > 0 ? 'border-red-200 bg-red-50 text-red-700' : 'border-green-200 bg-green-50 text-green-700'}`}
|
||||
title={`${k.open_asset_count} asset(s) still open, ${k.patched_asset_count} already remediated`}>
|
||||
{k.open_asset_count > 0
|
||||
? `In inventory · ${k.open_asset_count}`
|
||||
: 'Remediated'}
|
||||
</span>
|
||||
) : (
|
||||
<span className="inline-flex items-center rounded-sm border border-gray-200 bg-gray-50 px-1.5 py-0.5 text-[10px] font-mono uppercase text-gray-400">not seen</span>
|
||||
)}
|
||||
|
||||
@@ -0,0 +1,311 @@
|
||||
"""Unified KEV catalog (CISA + ENISA EUVD) and the immediate alert mail.
|
||||
|
||||
Covers the three things that were wrong or missing before:
|
||||
|
||||
1. The Advisories badge counted every vulnerability row for a CVE, patched
|
||||
and inactive-asset ones included, so it said 64 where the filtered
|
||||
vulnerability view listed 62. Impact is now open vs patched, active
|
||||
assets only, and an asset carrying both an open and a closed row for the
|
||||
same CVE counts once — as open.
|
||||
2. ENISA EUVD was fetched but never shown as a KEV source. Its exploited
|
||||
catalog and CISA's now merge into one entry per CVE carrying both
|
||||
sources — and an EU-Critical entry that is NOT exploited stays out.
|
||||
3. Alerting is idempotent: one mail per CVE, another only when the number
|
||||
of affected systems grows.
|
||||
|
||||
Run: python tests/test_kev_alerts.py
|
||||
"""
|
||||
import os
|
||||
import sys
|
||||
|
||||
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
|
||||
|
||||
from app.models.vulnerability import VulnerabilityStatus # noqa: E402
|
||||
from app.services import kev_alert_service as alert # noqa: E402
|
||||
from app.services import kev_service as kev # noqa: E402
|
||||
|
||||
OPEN = VulnerabilityStatus.open
|
||||
PATCHED = VulnerabilityStatus.patched
|
||||
FALSE_POS = VulnerabilityStatus.false_positive
|
||||
PATCH_FAILED = VulnerabilityStatus.patch_failed
|
||||
|
||||
|
||||
class _Row:
|
||||
def __init__(self, value):
|
||||
self.value = value
|
||||
|
||||
|
||||
class _Query:
|
||||
"""Enough of a SQLAlchemy query for the read paths under test."""
|
||||
|
||||
def __init__(self, rows, first=None):
|
||||
self._rows, self._first = rows, first
|
||||
|
||||
def join(self, *a, **kw):
|
||||
return self
|
||||
|
||||
def filter(self, *a, **kw):
|
||||
return self
|
||||
|
||||
def group_by(self, *a, **kw):
|
||||
return self
|
||||
|
||||
def all(self):
|
||||
return self._rows
|
||||
|
||||
def first(self):
|
||||
return self._first
|
||||
|
||||
|
||||
class _DB:
|
||||
def __init__(self, rows=(), setting=None):
|
||||
self.rows, self.setting = list(rows), setting
|
||||
self.written = {}
|
||||
|
||||
def query(self, *a, **kw):
|
||||
return _Query(self.rows, self.setting)
|
||||
|
||||
def add(self, obj):
|
||||
self.written[getattr(obj, "key", "?")] = getattr(obj, "value", None)
|
||||
|
||||
def commit(self):
|
||||
pass
|
||||
|
||||
|
||||
# ---------------------------------------------------------------- EUVD flag
|
||||
|
||||
def test_euvd_exploited_flag():
|
||||
assert kev.euvd_is_exploited({"exploited": True, "critical": True})
|
||||
assert not kev.euvd_is_exploited({"exploited": False, "critical": True})
|
||||
# An entry without the flag is a pre-flag cache, where a CVE in both ENISA
|
||||
# catalogs merged to critical=True and lost its origin. Such a cache is
|
||||
# treated as stale and refetched; only the "every ENISA call failed" fallback
|
||||
# still reaches here, and there under-reporting exploitation is the worse
|
||||
# error — so unflagged counts as exploited, critical or not.
|
||||
assert kev.euvd_is_exploited({"critical": False})
|
||||
assert kev.euvd_is_exploited({"critical": True})
|
||||
|
||||
|
||||
def test_pre_flag_euvd_cache_is_treated_as_stale():
|
||||
"""The flag cannot be reconstructed from an old cache, so the cache is
|
||||
refetched rather than guessed at — a one-fetch window, not 24h of
|
||||
silently-dropped exploited-and-EU-critical CVEs."""
|
||||
import json
|
||||
from datetime import datetime
|
||||
|
||||
from app.services import enrichment_service as enr
|
||||
|
||||
now = datetime.now().isoformat()
|
||||
|
||||
class _TwoKeyDB:
|
||||
"""Returns the timestamp row for one key and the cache row for the other."""
|
||||
|
||||
def __init__(self, payload):
|
||||
self.payload = payload
|
||||
|
||||
def query(self, *a, **kw):
|
||||
return self
|
||||
|
||||
def filter(self, crit):
|
||||
self._is_ts = enr.EUVD_CACHE_TS_KEY in str(crit.right.value)
|
||||
return self
|
||||
|
||||
def first(self):
|
||||
return _Row(now if self._is_ts else json.dumps(self.payload))
|
||||
|
||||
fresh = {"CVE-1": {"critical": True, "exploited": True}}
|
||||
stale = {"CVE-1": {"critical": True}}
|
||||
assert enr._load_euvd_cache(_TwoKeyDB(fresh)) == fresh
|
||||
assert enr._load_euvd_cache(_TwoKeyDB(stale)) is None
|
||||
|
||||
|
||||
def test_source_selection_never_mutes_the_catalog():
|
||||
assert kev._normalize_sources(None) == kev.ALL_SOURCES
|
||||
assert kev._normalize_sources([]) == kev.ALL_SOURCES
|
||||
assert kev._normalize_sources(["typo"]) == kev.ALL_SOURCES # not silence
|
||||
assert kev._normalize_sources(["euvd"]) == ("euvd",)
|
||||
assert kev._normalize_sources(["EUVD ", "cisa"]) == ("cisa", "euvd")
|
||||
|
||||
|
||||
# ------------------------------------------------------------- catalog merge
|
||||
|
||||
def test_catalog_merges_both_sources(monkeypatch):
|
||||
from app.services import enrichment_service as enr
|
||||
monkeypatch.setattr(enr, "fetch_kev_catalog", lambda db, **kw: {
|
||||
"CVE-2026-68820": {"vendor": "Microsoft", "product": "Windows AFD",
|
||||
"name": "AFD EoP", "date_added": "2026-08-11",
|
||||
"ransomware_use": True, "short_description": "boom"},
|
||||
"CVE-2023-1111": {"vendor": "Old", "product": "Thing",
|
||||
"date_added": "2023-01-01"},
|
||||
})
|
||||
monkeypatch.setattr(enr, "fetch_euvd_catalogs", lambda db, **kw: {
|
||||
"CVE-2023-1111": {"date_added": "2026-08-14T09:00:00", "critical": True,
|
||||
"exploited": True, "euvd_id": "EUVD-2023-1"},
|
||||
"CVE-2026-9999": {"date_added": "2026-08-12T00:00:00", "critical": False,
|
||||
"exploited": True, "euvd_id": "EUVD-2026-9"},
|
||||
"CVE-2026-5555": {"date_added": "2026-08-13T00:00:00", "critical": True,
|
||||
"exploited": False, "euvd_id": "EUVD-2026-5"},
|
||||
})
|
||||
|
||||
cat = kev.get_kev_catalog(_DB())
|
||||
# EU-Critical without exploitation is not a KEV entry.
|
||||
assert "CVE-2026-5555" not in cat
|
||||
assert cat["CVE-2026-9999"]["sources"] == ["euvd"]
|
||||
assert cat["CVE-2026-68820"]["sources"] == ["cisa"]
|
||||
assert cat["CVE-2026-68820"]["ransomware"] is True
|
||||
|
||||
both = cat["CVE-2023-1111"]
|
||||
assert both["sources"] == ["cisa", "euvd"]
|
||||
assert both["vendor"] == "Old" # descriptive text from CISA
|
||||
assert both["euvd_id"] == "EUVD-2023-1" # id from ENISA
|
||||
# "Latest additions" means latest: an old CISA entry ENISA flagged
|
||||
# yesterday is news yesterday, not in 2023.
|
||||
assert both["date_added"] == "2026-08-14"
|
||||
assert both["source_dates"] == {"cisa": "2023-01-01", "euvd": "2026-08-14"}
|
||||
|
||||
|
||||
def test_one_dead_catalog_does_not_blank_the_other(monkeypatch):
|
||||
from app.services import enrichment_service as enr
|
||||
|
||||
def _boom(db, **kw):
|
||||
raise RuntimeError("ENISA is having a day")
|
||||
|
||||
monkeypatch.setattr(enr, "fetch_kev_catalog",
|
||||
lambda db, **kw: {"CVE-2026-1": {"date_added": "2026-08-01"}})
|
||||
monkeypatch.setattr(enr, "fetch_euvd_catalogs", _boom)
|
||||
assert list(kev.get_kev_catalog(_DB())) == ["CVE-2026-1"]
|
||||
|
||||
|
||||
# ----------------------------------------------------------- inventory count
|
||||
|
||||
def test_impact_splits_open_from_patched():
|
||||
"""The 62-vs-64 bug: patched findings inflated the badge, and an asset
|
||||
with both an open and a closed row was counted twice."""
|
||||
rows = [
|
||||
("CVE-2026-68820", OPEN, 1, "web01", "10.0.0.1"),
|
||||
("CVE-2026-68820", OPEN, 2, "web02", "10.0.0.2"),
|
||||
("CVE-2026-68820", PATCH_FAILED, 3, "db01", "10.0.0.3"), # still open
|
||||
("CVE-2026-68820", PATCHED, 4, "old01", "10.0.0.4"),
|
||||
("CVE-2026-68820", FALSE_POS, 5, "fp01", "10.0.0.5"),
|
||||
# Same asset, two rows (two packages): open wins, counted once.
|
||||
("CVE-2026-68820", PATCHED, 1, "web01", "10.0.0.1"),
|
||||
]
|
||||
imp = kev.inventory_impact(_DB(rows), ["cve-2026-68820"], with_hosts=True)["CVE-2026-68820"]
|
||||
assert imp["open_asset_count"] == 3
|
||||
assert imp["patched_asset_count"] == 2
|
||||
assert imp["asset_count"] == 5
|
||||
assert imp["in_inventory"] is True
|
||||
assert [h["hostname"] for h in imp["hosts"]] == ["db01", "web01", "web02"]
|
||||
|
||||
|
||||
def test_impact_ignores_unknown_cves_and_empty_input():
|
||||
assert kev.inventory_impact(_DB(), []) == {}
|
||||
assert kev.inventory_impact(_DB(), [None, ""]) == {}
|
||||
assert kev.inventory_impact(_DB(), ["CVE-2026-1"]) == {}
|
||||
|
||||
|
||||
def test_host_list_is_capped():
|
||||
rows = [("CVE-2026-1", OPEN, i, f"host{i:03d}", None) for i in range(120)]
|
||||
imp = kev.inventory_impact(_DB(rows), ["CVE-2026-1"], with_hosts=True)["CVE-2026-1"]
|
||||
assert imp["open_asset_count"] == 120 # the count is never capped
|
||||
assert len(imp["hosts"]) == kev.MAX_HOSTS_PER_CVE
|
||||
assert imp["hosts_truncated"] == 120 - kev.MAX_HOSTS_PER_CVE
|
||||
|
||||
|
||||
# ------------------------------------------------------------------ alerting
|
||||
|
||||
def _patch_alert_inputs(monkeypatch, catalog, impact, state):
|
||||
monkeypatch.setattr(kev, "get_kev_catalog", lambda db, sources=None: catalog)
|
||||
monkeypatch.setattr(kev, "inventory_impact",
|
||||
lambda db, cves, with_hosts=False: impact)
|
||||
monkeypatch.setattr(alert, "_load_state", lambda db: state)
|
||||
|
||||
|
||||
_CATALOG = {
|
||||
"CVE-2026-68820": {"cve_id": "CVE-2026-68820", "sources": ["cisa", "euvd"],
|
||||
"date_added": "2026-08-11", "vendor": "Microsoft",
|
||||
"product": "Windows AFD", "name": "AFD EoP",
|
||||
"ransomware": True, "description": None, "euvd_id": "EUVD-1",
|
||||
"source_dates": {}},
|
||||
"CVE-2026-1234": {"cve_id": "CVE-2026-1234", "sources": ["euvd"],
|
||||
"date_added": "2026-08-09", "vendor": None, "product": None,
|
||||
"name": None, "ransomware": False, "description": None,
|
||||
"euvd_id": "EUVD-2", "source_dates": {}},
|
||||
}
|
||||
|
||||
|
||||
def _impact(open_a=3, patched=2):
|
||||
return {
|
||||
"CVE-2026-68820": {"open_asset_count": open_a, "patched_asset_count": patched,
|
||||
"asset_count": open_a + patched, "in_inventory": True,
|
||||
"hosts": [{"hostname": "web01", "ip_address": "10.0.0.1"}],
|
||||
"hosts_truncated": 0},
|
||||
# Fully remediated — must never trigger a mail.
|
||||
"CVE-2026-1234": {"open_asset_count": 0, "patched_asset_count": 7,
|
||||
"asset_count": 7, "in_inventory": True,
|
||||
"hosts": [], "hosts_truncated": 0},
|
||||
}
|
||||
|
||||
|
||||
def test_only_open_findings_alert(monkeypatch):
|
||||
_patch_alert_inputs(monkeypatch, _CATALOG, _impact(), {})
|
||||
found = alert.find_alertable(_DB())
|
||||
assert [a["cve_id"] for a in found] == ["CVE-2026-68820"]
|
||||
assert found[0]["previously_notified"] is False
|
||||
|
||||
|
||||
def test_alert_is_sent_once(monkeypatch):
|
||||
state = {"CVE-2026-68820": {"at": "2026-08-11T10:00:00", "n": 3}}
|
||||
_patch_alert_inputs(monkeypatch, _CATALOG, _impact(open_a=3), state)
|
||||
assert alert.find_alertable(_DB()) == []
|
||||
|
||||
# ... and again as soon as another system is affected.
|
||||
_patch_alert_inputs(monkeypatch, _CATALOG, _impact(open_a=4), state)
|
||||
again = alert.find_alertable(_DB())
|
||||
assert [a["cve_id"] for a in again] == ["CVE-2026-68820"]
|
||||
assert again[0]["previously_notified"] is True
|
||||
assert again[0]["previous_open_asset_count"] == 3
|
||||
|
||||
|
||||
def test_preview_shows_already_sent(monkeypatch):
|
||||
state = {"CVE-2026-68820": {"at": "x", "n": 3}}
|
||||
_patch_alert_inputs(monkeypatch, _CATALOG, _impact(open_a=3), state)
|
||||
monkeypatch.setattr(alert, "_load_state", lambda db: state)
|
||||
shown = alert.find_alertable(_DB(), include_already_sent=True)
|
||||
assert [a["cve_id"] for a in shown] == ["CVE-2026-68820"]
|
||||
|
||||
|
||||
def test_rows_escape_hostile_input():
|
||||
rows = alert.render_alert_rows([{
|
||||
"cve_id": "CVE-2026-1", "sources": ["cisa"], "date_added": "2026-08-11",
|
||||
"name": "<script>alert(1)</script>", "vendor": None, "product": None,
|
||||
"ransomware": False, "open_asset_count": 2, "patched_asset_count": 1,
|
||||
"hosts": [{"hostname": "<img src=x onerror=1>"}, {"hostname": "web01"}],
|
||||
"hosts_truncated": 3,
|
||||
}], base_url="https://vuln.example/vulnerabilities")
|
||||
assert "<script>" not in rows
|
||||
assert "<script>" in rows
|
||||
assert "<img src=x" not in rows
|
||||
assert "CISA KEV" in rows
|
||||
assert "+3 more" in rows
|
||||
assert "1 system(s) already remediated" in rows
|
||||
assert "cve_id=CVE-2026-1" in rows
|
||||
|
||||
|
||||
def test_enabled_switch_reads_falsey_values():
|
||||
assert alert.is_enabled(_DB(setting=None)) is True # default on
|
||||
assert alert.is_enabled(_DB(setting=_Row("true"))) is True
|
||||
assert alert.is_enabled(_DB(setting=_Row('"false"'))) is False
|
||||
assert alert.is_enabled(_DB(setting=_Row("OFF"))) is False
|
||||
assert alert.is_enabled(_DB(setting=_Row("0"))) is False
|
||||
|
||||
|
||||
def test_configured_sources_parse():
|
||||
assert alert.get_sources(_DB(setting=None)) is None
|
||||
assert alert.get_sources(_DB(setting=_Row("cisa, euvd"))) == ["cisa", "euvd"]
|
||||
assert alert.get_sources(_DB(setting=_Row(" "))) is None
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
import pytest
|
||||
sys.exit(pytest.main([__file__, "-q"]))
|
||||
Reference in New Issue
Block a user