fix(feeds): stop repeating downloads the app already made
Follow-up to the per-device vulnrichment fix: an audit of the backend found
more feeds pulled again by every caller.
cvelistV5 (~557 MB) had three downloaders: the index build, the CVSS
cascade's stage 3 and the date backfill. Callers arriving together each
pulled the whole ZIP, and after a failed download every later caller
retried at once. _ensure_zip is now the only one: a lock with a freshness
re-check after acquiring it, a zipfile check before the swap, and a 15 min
backoff after a failure (the on-disk ZIP still serves meanwhile).
Per-CVE lookups that keep missing are remembered for 6h: the M365 title
fetch on a cvelistV5 404 (MSRC publishes days before CVE.org), and NVD /
GHSA for CVEs they answered without a score. Rate limits and network
errors are not answers and are retried.
A failed or partial index build stores nothing, which keeps the cache
stale, so every per-asset scan and every per-sync enrichment rebuilt it
during an outage. The MFSA, repo-advisory and TeamViewer indexes and the
KEV / EUVD / VulnCheck refreshes now serve the stale copy for 1h after a
failure; the nightly and forced paths still rebuild. The app scan loads
the repo-advisory and TeamViewer indexes once per run, like the IGEL ISNs.
The repo-advisory index no longer falls back to {} when a build raises:
the reconcile would read every untouched finding as patched.
MSRC CVRF documents were downloaded by the 01:30 index build and again by
the 03:50 / Sunday ingest. Both now go through fetch_cvrf_doc, a disk
cache keyed by the revision date /updates lists (max 4h old), so a
revised document is still fetched again.
This commit is contained in:
+7
-1
@@ -906,12 +906,18 @@ bounded even for full-DB corrections.
|
||||
┌─ Stage 3 — cvelistV5 (MITRE/CVE.org) ──────────────────────────┐
|
||||
│ github.com/CVEProject/cvelistV5 │
|
||||
│ • 557 MB ZIP, disk-cached 12h at │
|
||||
│ /tmp/truevuln-cvelistv5-cache.zip │
|
||||
│ /tmp/truevuln-cvelistv5-cache.zip — one downloader for the │
|
||||
│ whole app (cvelistv5_scan_service._ensure_zip: lock, │
|
||||
│ ZIP check, 15 min backoff after a failed download) │
|
||||
│ • exhaustive — every published CVE (~250k+) │
|
||||
│ • same CVE-5 JSON shape as Vulnrichment, parser reused │
|
||||
└─────────────────────────────────────────────────────────────────┘
|
||||
```
|
||||
|
||||
NVD and GHSA (stage 4) remember for 6h a CVE they answered without a
|
||||
score (404 / empty / no CVSS), so per-device checks don't re-ask them for
|
||||
fresh CVEs nobody has scored yet; rate limits and errors are retried.
|
||||
|
||||
Verified records carry `source ∈ {vulnrichment, nvd, cvelistv5}` so
|
||||
the operator can tell where each correction came from. All three are
|
||||
treated as authoritative — Nessus sync respects an
|
||||
|
||||
@@ -2320,6 +2320,20 @@ def run_app_cve_scan(db: Session, asset_id: Optional[int] = None,
|
||||
except Exception as e:
|
||||
logger.warning("app-cve: IGEL ISN index load failed: %s", e)
|
||||
|
||||
# Repo advisories + TeamViewer bulletins: loaded once per run, not per
|
||||
# asset. None (load failed) lets scan_asset try again on its own.
|
||||
repo_adv_index = tv_index = None
|
||||
try:
|
||||
from app.services import github_repo_advisory_service
|
||||
repo_adv_index = github_repo_advisory_service.get_index(db)
|
||||
except Exception as e:
|
||||
logger.warning("app-cve: repo-advisory index load failed: %s", e)
|
||||
try:
|
||||
from app.services import teamviewer_bulletin_service
|
||||
tv_index = teamviewer_bulletin_service.get_index(db)
|
||||
except Exception as e:
|
||||
logger.warning("app-cve: teamviewer-bulletin index load failed: %s", e)
|
||||
|
||||
# MSRC fixed-build index (Windows Server, SharePoint, Edge). Rebuilt at
|
||||
# 01:30 by vuln_index_refresh_nightly and loaded here ONCE — it used to be
|
||||
# parsed from its JSON blob again for every asset.
|
||||
@@ -2656,7 +2670,8 @@ def run_app_cve_scan(db: Session, asset_id: Optional[int] = None,
|
||||
from app.services import github_repo_advisory_service
|
||||
stats["repo_advisory_findings"] = stats.get("repo_advisory_findings", 0) + \
|
||||
github_repo_advisory_service.scan_asset(
|
||||
db, asset, packages, new_ids, touched=touched_cves)
|
||||
db, asset, packages, new_ids, touched=touched_cves,
|
||||
index=repo_adv_index)
|
||||
except Exception as e:
|
||||
stats["errors"].append(f"asset {asset.id} repo-advisories: {e}")
|
||||
# TeamViewer's own bulletins. Same reason as the repo advisories:
|
||||
@@ -2667,7 +2682,8 @@ def run_app_cve_scan(db: Session, asset_id: Optional[int] = None,
|
||||
from app.services import teamviewer_bulletin_service
|
||||
stats["teamviewer_findings"] = stats.get("teamviewer_findings", 0) + \
|
||||
teamviewer_bulletin_service.scan_asset(
|
||||
db, asset, packages, new_ids, touched=touched_cves)
|
||||
db, asset, packages, new_ids, touched=touched_cves,
|
||||
index=tv_index)
|
||||
except Exception as e:
|
||||
stats["errors"].append(f"asset {asset.id} teamviewer-bulletins: {e}")
|
||||
touched = True
|
||||
|
||||
@@ -39,10 +39,11 @@ from app.services import app_cve_scanner_service as cpe
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
# Reuse the same ZIP the override service already downloads/caches (12h).
|
||||
# Shared ZIP cache (12h) — every reader in the app goes through _ensure_zip.
|
||||
_ZIP_PATH = "/tmp/truevuln-cvelistv5-cache.zip"
|
||||
_ZIP_URL = "https://github.com/CVEProject/cvelistV5/archive/refs/heads/main.zip"
|
||||
_ZIP_TTL = 12 * 3600
|
||||
_ZIP_MIN_BYTES = 100_000_000 # sanity floor; the real ZIP is ~557 MB
|
||||
_INDEX_SETTING = "cvelistv5_product_index_v48" # v48: Zoom bulletins
|
||||
_INDEX_TTL = timedelta(hours=26) # rebuilt nightly; a missed night still serves
|
||||
|
||||
@@ -1261,6 +1262,14 @@ def _esr_patched(installed: str, esr_fixes: List[str]) -> bool:
|
||||
|
||||
|
||||
# ---------- index build + cache ----------
|
||||
# One downloader for the whole app: the nightly index build, the GUI app scan,
|
||||
# the CVSS cascade and the date backfill all come through here. The lock makes
|
||||
# a second caller wait for the first download instead of starting its own.
|
||||
_ZIP_LOCK = threading.Lock()
|
||||
_ZIP_RETRY_BACKOFF = 15 * 60
|
||||
_zip_retry_after = 0.0 # time.monotonic() before which a failed download is not retried
|
||||
|
||||
|
||||
def _ensure_zip(force: bool = False) -> bool:
|
||||
"""Make sure the shared cvelistV5 ZIP is on disk.
|
||||
|
||||
@@ -1272,42 +1281,57 @@ def _ensure_zip(force: bool = False) -> bool:
|
||||
published (Chrome 151.0.7922.169 surfaced >24h late that way). On-demand
|
||||
callers keep the cache; the one GUI path that forces a download is the
|
||||
Data-jobs app scan, whose dialog says so (refresh_indexes).
|
||||
|
||||
After a failed download, callers within _ZIP_RETRY_BACKOFF use what is on
|
||||
disk (or get False) instead of each pulling 557 MB from GitHub again.
|
||||
"""
|
||||
on_disk = (os.path.exists(_ZIP_PATH)
|
||||
and os.path.getsize(_ZIP_PATH) > 100_000_000)
|
||||
fresh = (on_disk
|
||||
and (time.time() - os.path.getmtime(_ZIP_PATH)) < _ZIP_TTL)
|
||||
if fresh and not force:
|
||||
return True
|
||||
import httpx
|
||||
# Per-writer temp name: the nightly refresh, the GUI app scan and the
|
||||
# threat-intel cascade (vuln_override_service) all download this file and
|
||||
# can overlap; a shared .part interleaved their bytes into a "fresh" ZIP.
|
||||
tmp = f"{_ZIP_PATH}.{os.getpid()}.{threading.get_ident()}.part"
|
||||
try:
|
||||
logger.info("cvelistv5-scan: downloading ZIP snapshot (~557 MB)…")
|
||||
with httpx.Client(timeout=httpx.Timeout(600.0, connect=15.0), follow_redirects=True) as c:
|
||||
with c.stream("GET", _ZIP_URL) as r:
|
||||
r.raise_for_status()
|
||||
with open(tmp, "wb") as f:
|
||||
for chunk in r.iter_bytes(chunk_size=1024 * 512):
|
||||
f.write(chunk)
|
||||
os.replace(tmp, _ZIP_PATH)
|
||||
return True
|
||||
except Exception as e:
|
||||
logger.warning("cvelistv5-scan: ZIP download failed: %s", e)
|
||||
if os.path.exists(tmp):
|
||||
try:
|
||||
os.remove(tmp)
|
||||
except Exception:
|
||||
pass
|
||||
# A forced refresh that fails still beats no walk at all: walking a
|
||||
# stale-but-valid ZIP keeps the index moving on a transient blip.
|
||||
if on_disk:
|
||||
logger.warning("cvelistv5-scan: falling back to the on-disk ZIP (%.1fh old)",
|
||||
(time.time() - os.path.getmtime(_ZIP_PATH)) / 3600)
|
||||
global _zip_retry_after
|
||||
asked_at = time.time()
|
||||
with _ZIP_LOCK:
|
||||
on_disk = (os.path.exists(_ZIP_PATH)
|
||||
and os.path.getsize(_ZIP_PATH) > _ZIP_MIN_BYTES)
|
||||
mtime = os.path.getmtime(_ZIP_PATH) if on_disk else 0.0
|
||||
fresh = on_disk and (time.time() - mtime) < _ZIP_TTL
|
||||
# A forced caller is also served by a download that finished while it
|
||||
# waited for the lock.
|
||||
if fresh and (not force or mtime >= asked_at):
|
||||
return True
|
||||
return False
|
||||
if time.monotonic() < _zip_retry_after:
|
||||
logger.warning("cvelistv5-scan: ZIP download failed recently — %s",
|
||||
"using the on-disk ZIP" if on_disk else "no ZIP available")
|
||||
return on_disk
|
||||
import httpx
|
||||
# Per-writer temp name: never let a reader see a half-written file.
|
||||
tmp = f"{_ZIP_PATH}.{os.getpid()}.{threading.get_ident()}.part"
|
||||
try:
|
||||
logger.info("cvelistv5-scan: downloading ZIP snapshot (~557 MB)…")
|
||||
with httpx.Client(timeout=httpx.Timeout(600.0, connect=15.0), follow_redirects=True) as c:
|
||||
with c.stream("GET", _ZIP_URL) as r:
|
||||
r.raise_for_status()
|
||||
with open(tmp, "wb") as f:
|
||||
for chunk in r.iter_bytes(chunk_size=1024 * 512):
|
||||
f.write(chunk)
|
||||
if not zipfile.is_zipfile(tmp):
|
||||
raise ValueError("download is not a ZIP")
|
||||
os.replace(tmp, _ZIP_PATH)
|
||||
_zip_retry_after = 0.0
|
||||
return True
|
||||
except Exception as e:
|
||||
_zip_retry_after = time.monotonic() + _ZIP_RETRY_BACKOFF
|
||||
logger.warning("cvelistv5-scan: ZIP download failed: %s", e)
|
||||
# A forced refresh that fails still beats no walk at all: walking a
|
||||
# stale-but-valid ZIP keeps the index moving on a transient blip.
|
||||
if on_disk:
|
||||
logger.warning("cvelistv5-scan: falling back to the on-disk ZIP (%.1fh old)",
|
||||
(time.time() - mtime) / 3600)
|
||||
return True
|
||||
return False
|
||||
finally:
|
||||
if os.path.exists(tmp):
|
||||
try:
|
||||
os.remove(tmp)
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
|
||||
def build_product_index(db: Session, force_fresh: bool = False) -> dict:
|
||||
|
||||
@@ -174,6 +174,26 @@ def fetch_epss_scores(cve_ids: Iterable[str]) -> Dict[str, Dict[str, float]]:
|
||||
return results
|
||||
|
||||
|
||||
# A failed refresh stamps nothing, so the cache stays stale and every
|
||||
# enrichment run (one per sync) would pull the whole catalog again during an
|
||||
# outage. After a failure, non-forced callers get the stale cache instead.
|
||||
_FETCH_RETRY_SECONDS = 3600
|
||||
_fetch_failed_at: Dict[str, float] = {} # source → time.monotonic()
|
||||
|
||||
|
||||
def _stale_while_backing_off(db: Session, source: str, cache_key: str) -> Optional[Dict[str, dict]]:
|
||||
"""The stale cache when `source` failed within _FETCH_RETRY_SECONDS (raises
|
||||
when there is none), else None: go fetch."""
|
||||
t = _fetch_failed_at.get(source)
|
||||
if t is None or time.monotonic() - t >= _FETCH_RETRY_SECONDS:
|
||||
return None
|
||||
stale = _stale_ttl_cache(db, cache_key)
|
||||
if stale is None:
|
||||
raise EnrichmentError(f"{source} fetch failed recently and no cache available")
|
||||
logger.debug(f"{source}: last refresh failed recently, serving the stale cache")
|
||||
return stale
|
||||
|
||||
|
||||
# ============================================================
|
||||
# CISA KEV
|
||||
# ============================================================
|
||||
@@ -216,6 +236,9 @@ def fetch_kev_catalog(db: Session, force_refresh: bool = False) -> Dict[str, dic
|
||||
if cached is not None:
|
||||
logger.debug(f"KEV: using cached catalog ({len(cached)} entries)")
|
||||
return cached
|
||||
stale = _stale_while_backing_off(db, "KEV", KEV_CACHE_KEY)
|
||||
if stale is not None:
|
||||
return stale
|
||||
|
||||
logger.info("KEV: fetching fresh catalog from CISA")
|
||||
try:
|
||||
@@ -225,6 +248,7 @@ def fetch_kev_catalog(db: Session, force_refresh: bool = False) -> Dict[str, dic
|
||||
payload = resp.json()
|
||||
except httpx.HTTPError as e:
|
||||
logger.error(f"KEV fetch failed: {e}")
|
||||
_fetch_failed_at["KEV"] = time.monotonic()
|
||||
# Fall back to stale cache if present
|
||||
cache_setting = db.query(Setting).filter(Setting.key == KEV_CACHE_KEY).first()
|
||||
if cache_setting and cache_setting.value:
|
||||
@@ -251,6 +275,7 @@ def fetch_kev_catalog(db: Session, force_refresh: bool = False) -> Dict[str, dic
|
||||
}
|
||||
|
||||
_store_kev_cache(db, kev_map)
|
||||
_fetch_failed_at.pop("KEV", None)
|
||||
logger.info(f"KEV: cached {len(kev_map)} entries from CISA")
|
||||
return kev_map
|
||||
|
||||
@@ -516,6 +541,9 @@ def fetch_euvd_catalogs(db: Session, force_refresh: bool = False) -> Dict[str, d
|
||||
if cached is not None:
|
||||
logger.debug(f"EUVD: using cached catalog ({len(cached)} entries)")
|
||||
return cached
|
||||
stale = _stale_while_backing_off(db, "EUVD", EUVD_CACHE_KEY)
|
||||
if stale is not None:
|
||||
return stale
|
||||
|
||||
logger.info("EUVD: fetching fresh catalogs from ENISA")
|
||||
merged: Dict[str, dict] = {}
|
||||
@@ -542,6 +570,7 @@ def fetch_euvd_catalogs(db: Session, force_refresh: bool = False) -> Dict[str, d
|
||||
logger.error(f"EUVD client error: {e}")
|
||||
|
||||
if not merged:
|
||||
_fetch_failed_at["EUVD"] = time.monotonic()
|
||||
# All fetches failed: try stale cache
|
||||
cache_setting = db.query(Setting).filter(Setting.key == EUVD_CACHE_KEY).first()
|
||||
if cache_setting and cache_setting.value:
|
||||
@@ -553,6 +582,7 @@ def fetch_euvd_catalogs(db: Session, force_refresh: bool = False) -> Dict[str, d
|
||||
raise EnrichmentError("EUVD fetch failed and no cache available")
|
||||
|
||||
_store_euvd_cache(db, merged)
|
||||
_fetch_failed_at.pop("EUVD", None)
|
||||
logger.info(
|
||||
f"EUVD: cached {len(merged)} unique CVEs from ENISA "
|
||||
f"(exploited pages processed={exploited_count}, critical entries={critical_count})"
|
||||
@@ -726,6 +756,9 @@ def fetch_vulncheck_kev(db: Session, force_refresh: bool = False) -> Dict[str, d
|
||||
if cached is not None:
|
||||
logger.debug(f"VulnCheck KEV: using cached catalog ({len(cached)} entries)")
|
||||
return cached
|
||||
stale = _stale_while_backing_off(db, "VulnCheck KEV", VULNCHECK_CACHE_KEY)
|
||||
if stale is not None:
|
||||
return stale
|
||||
|
||||
logger.info("VulnCheck KEV: fetching fresh catalog")
|
||||
catalog: Dict[str, dict] = {}
|
||||
@@ -741,6 +774,7 @@ def fetch_vulncheck_kev(db: Session, force_refresh: bool = False) -> Dict[str, d
|
||||
logger.error(f"VulnCheck KEV fetch failed: {e}")
|
||||
|
||||
if not catalog:
|
||||
_fetch_failed_at["VulnCheck KEV"] = time.monotonic()
|
||||
stale = _stale_ttl_cache(db, VULNCHECK_CACHE_KEY)
|
||||
if stale is not None:
|
||||
logger.warning("VulnCheck KEV: returning stale cache after fetch failure")
|
||||
@@ -751,6 +785,7 @@ def fetch_vulncheck_kev(db: Session, force_refresh: bool = False) -> Dict[str, d
|
||||
"VulnCheck KEV cache (24h TTL)")
|
||||
_set_setting(db, VULNCHECK_CACHE_TS_KEY, datetime.now().isoformat(),
|
||||
"Timestamp of last VulnCheck KEV refresh")
|
||||
_fetch_failed_at.pop("VulnCheck KEV", None)
|
||||
logger.info(f"VulnCheck KEV: cached {len(catalog)} CVEs")
|
||||
return catalog
|
||||
|
||||
|
||||
@@ -22,6 +22,7 @@ no separate source/reconcile pair to maintain.
|
||||
import json
|
||||
import logging
|
||||
import re
|
||||
import time
|
||||
from datetime import datetime, timedelta
|
||||
from typing import Dict, List, Optional, Tuple
|
||||
|
||||
@@ -481,26 +482,44 @@ def load_index(db: Session) -> Optional[Dict[str, list]]:
|
||||
return _stored(db)
|
||||
|
||||
|
||||
# A failed or partial build stores nothing, so the index stays stale and the
|
||||
# next caller would rebuild again — per asset / per sync during an outage.
|
||||
# After a failed build, callers get the stale index for this long instead.
|
||||
_BUILD_RETRY_SECONDS = 3600
|
||||
_build_failed_at: Optional[float] = None # time.monotonic()
|
||||
|
||||
|
||||
def get_index(db: Session) -> Dict[str, list]:
|
||||
global _build_failed_at
|
||||
idx = load_index(db)
|
||||
if idx is not None:
|
||||
return idx
|
||||
if _build_failed_at is not None and time.monotonic() - _build_failed_at < _BUILD_RETRY_SECONDS:
|
||||
return _stored(db) or {}
|
||||
try:
|
||||
return build_index(db)
|
||||
built = build_index(db)
|
||||
except Exception as e:
|
||||
# Fall back to the STALE index, never to an empty one. An empty index
|
||||
# leaves every finding of this source untouched this run, and the
|
||||
# app-scan reconcile reads untouched as "no longer detected" → a live
|
||||
# flaw is marked patched because the vendor was briefly unreachable.
|
||||
logger.warning("repo-advisory index build failed: %s", e)
|
||||
return {}
|
||||
built = _stored(db) or {}
|
||||
# build_index stores (and stamps) only a complete index.
|
||||
_build_failed_at = None if load_index(db) is not None else time.monotonic()
|
||||
return built
|
||||
|
||||
|
||||
def scan_asset(db: Session, asset, packages: list, new_ids: Optional[list] = None,
|
||||
touched: Optional[set] = None) -> int:
|
||||
touched: Optional[set] = None, index: Optional[dict] = None) -> int:
|
||||
"""Match installed software against repo-published advisories.
|
||||
Findings go through the app-scan upsert (source 'app-scan'), so the
|
||||
existing app-scan reconcile closes them when the host updates.
|
||||
Returns findings upserted. Caller commits."""
|
||||
from app.services import app_cve_scanner_service as cpe
|
||||
|
||||
index = get_index(db)
|
||||
if index is None:
|
||||
index = get_index(db)
|
||||
if not index:
|
||||
return 0
|
||||
if new_ids is None:
|
||||
|
||||
@@ -25,6 +25,7 @@ syscollector name, so we approximate it from the product name —
|
||||
import json
|
||||
import logging
|
||||
import re
|
||||
import time
|
||||
from datetime import datetime, timedelta
|
||||
from typing import Dict, List, Optional, Tuple
|
||||
|
||||
@@ -670,11 +671,21 @@ _M365_SOURCE_NOTE = (
|
||||
)
|
||||
|
||||
|
||||
# CVEs cvelistV5 does not carry yet (404). MSRC publishes M365 CVEs before
|
||||
# CVE.org does, so these 404 for days; without the cache every device asked
|
||||
# again on every sync. Network errors are not recorded.
|
||||
_TEXT_MISS_TTL = 6 * 3600
|
||||
_text_misses: Dict[str, float] = {} # CVE → time.monotonic() of the 404
|
||||
|
||||
|
||||
def _fetch_cve_title_desc(cve_id: str) -> Tuple[Optional[str], Optional[str]]:
|
||||
"""Real CVE title + English description from CVE.org cvelistV5 raw."""
|
||||
m = re.fullmatch(r"CVE-(\d{4})-(\d+)", cve_id.upper())
|
||||
if not m:
|
||||
return None, None
|
||||
missed = _text_misses.get(cve_id.upper())
|
||||
if missed is not None and time.monotonic() - missed < _TEXT_MISS_TTL:
|
||||
return None, None
|
||||
year, num = m.group(1), m.group(2)
|
||||
url = (f"https://raw.githubusercontent.com/CVEProject/cvelistV5/main/"
|
||||
f"cves/{year}/{int(num) // 1000}xxx/{cve_id.upper()}.json")
|
||||
@@ -682,6 +693,8 @@ def _fetch_cve_title_desc(cve_id: str) -> Tuple[Optional[str], Optional[str]]:
|
||||
with httpx.Client(timeout=15.0, follow_redirects=True,
|
||||
headers={"User-Agent": "TrueVuln/1.0"}) as c:
|
||||
r = c.get(url)
|
||||
if r.status_code == 404:
|
||||
_text_misses[cve_id.upper()] = time.monotonic()
|
||||
if r.status_code != 200:
|
||||
return None, None
|
||||
cna = (r.json().get("containers") or {}).get("cna") or {}
|
||||
|
||||
@@ -26,6 +26,7 @@ regular — top-level `fixed_in:` list, then an `advisories:` map of
|
||||
import json
|
||||
import logging
|
||||
import re
|
||||
import time
|
||||
from datetime import datetime, timedelta
|
||||
from typing import Dict, List, Optional, Tuple
|
||||
|
||||
@@ -268,21 +269,33 @@ def load_index(db: Session, allow_stale: bool = False) -> Optional[Dict[str, dic
|
||||
return None
|
||||
|
||||
|
||||
# A failed or partial build stores nothing, so the index stays stale and the
|
||||
# next caller would rebuild again — per asset / per sync during an outage.
|
||||
# After a failed build, callers get the stale index for this long instead.
|
||||
_BUILD_RETRY_SECONDS = 3600
|
||||
_build_failed_at: Optional[float] = None # time.monotonic()
|
||||
|
||||
|
||||
def get_index(db: Session) -> Dict[str, dict]:
|
||||
"""Cached index (TTL 24h), lazily (re)built when absent/stale.
|
||||
|
||||
A failed rebuild falls back to the STALE cache rather than to {}: yesterday's
|
||||
Firefox advisories are a far better answer than none, and {} silently turns
|
||||
the Firefox scan off."""
|
||||
global _build_failed_at
|
||||
idx = load_index(db)
|
||||
if idx is not None:
|
||||
return idx
|
||||
if _build_failed_at is not None and time.monotonic() - _build_failed_at < _BUILD_RETRY_SECONDS:
|
||||
return load_index(db, allow_stale=True) or {}
|
||||
try:
|
||||
built = build_index(db)
|
||||
if built:
|
||||
_build_failed_at = None
|
||||
return built
|
||||
except Exception as e:
|
||||
logger.warning("MFSA index build failed: %s", e)
|
||||
_build_failed_at = time.monotonic()
|
||||
return load_index(db, allow_stale=True) or {}
|
||||
|
||||
|
||||
|
||||
@@ -200,8 +200,8 @@ def build_product_index(db: Session, months_back: Optional[int] = None) -> dict:
|
||||
"""Walk the recent monthly CVRF docs → {product_key: [{cve, build, kb}]}."""
|
||||
import httpx
|
||||
from app.services.msrc_service import (
|
||||
MSRC_BASE, HTTP_TIMEOUT, DEFAULT_MONTHS_BACK, SETTING_MONTHS_BACK,
|
||||
_setting_int, _list_recent_docs,
|
||||
HTTP_TIMEOUT, DEFAULT_MONTHS_BACK, SETTING_MONTHS_BACK,
|
||||
_setting_int, _list_recent_docs, fetch_cvrf_doc,
|
||||
)
|
||||
|
||||
months = months_back or _setting_int(db, SETTING_MONTHS_BACK, DEFAULT_MONTHS_BACK)
|
||||
@@ -217,9 +217,7 @@ def build_product_index(db: Session, months_back: Optional[int] = None) -> dict:
|
||||
|
||||
for doc_id in doc_ids:
|
||||
try:
|
||||
resp = client.get(f"{MSRC_BASE}/cvrf/{doc_id}")
|
||||
resp.raise_for_status()
|
||||
doc = resp.json()
|
||||
doc = fetch_cvrf_doc(client, doc_id)
|
||||
except Exception as e:
|
||||
logger.warning("msrc-scan: doc %s failed: %s", doc_id, e)
|
||||
failed.append(doc_id)
|
||||
|
||||
@@ -21,8 +21,12 @@ CVRF shapes we use, per Vulnerability:
|
||||
Title "Workarounds" → kind=workaround (HTML → text)
|
||||
Title "Mitigations" → kind=mitigation (HTML → text)
|
||||
"""
|
||||
import json
|
||||
import logging
|
||||
import os
|
||||
import re
|
||||
import threading
|
||||
import time
|
||||
from datetime import datetime, timedelta
|
||||
from typing import Dict, List, Optional
|
||||
|
||||
@@ -156,10 +160,11 @@ def _list_recent_docs(client: "httpx.Client", months_back: int) -> List[str]:
|
||||
"""Return the last `months_back` monthly CVRF document IDs (YYYY-Mon)."""
|
||||
r = client.get(f"{MSRC_BASE}/updates", headers={"Accept": "application/json"})
|
||||
r.raise_for_status()
|
||||
docs = [
|
||||
u["ID"] for u in (r.json().get("value") or [])
|
||||
if u.get("ID") and re.match(r"^\d{4}-[A-Za-z]{3}$", u["ID"])
|
||||
]
|
||||
docs = []
|
||||
for u in r.json().get("value") or []:
|
||||
if u.get("ID") and re.match(r"^\d{4}-[A-Za-z]{3}$", u["ID"]):
|
||||
docs.append(u["ID"])
|
||||
_doc_revision[u["ID"]] = u.get("CurrentReleaseDate") or ""
|
||||
# Sort by the month the document IS, taken from its ID. NOT by
|
||||
# CurrentReleaseDate: that is the last REVISION date, and Microsoft
|
||||
# re-touches ancient documents in bulk (on 19.08.2026 one sweep restamped
|
||||
@@ -172,6 +177,45 @@ def _list_recent_docs(client: "httpx.Client", months_back: int) -> List[str]:
|
||||
return docs[-months_back:]
|
||||
|
||||
|
||||
# Both CVRF walks — the 01:30 fixed-build index (msrc_scan_service) and the
|
||||
# 03:50 / Sunday 04:40 ingest below — read the same documents, up to 18 of
|
||||
# several MB each. A document is cached on disk with the revision /updates
|
||||
# listed, so the later walk reuses the earlier download unless Microsoft
|
||||
# revised it since; the age cap bounds a revision bump Microsoft forgot.
|
||||
_CVRF_CACHE_DIR = "/tmp/truevuln-msrc-cvrf"
|
||||
_CVRF_CACHE_MAX_AGE = 4 * 3600 # 01:30 → 04:40 fits
|
||||
_doc_revision: Dict[str, str] = {} # doc ID → CurrentReleaseDate, set by _list_recent_docs
|
||||
|
||||
|
||||
def fetch_cvrf_doc(client: "httpx.Client", doc_id: str) -> dict:
|
||||
"""One monthly CVRF document, from the disk cache when the cached copy is
|
||||
the revision _list_recent_docs last saw. Raises like client.get +
|
||||
raise_for_status + json()."""
|
||||
rev = _doc_revision.get(doc_id)
|
||||
path = os.path.join(_CVRF_CACHE_DIR, f"{doc_id}.json")
|
||||
if rev:
|
||||
try:
|
||||
if time.time() - os.path.getmtime(path) < _CVRF_CACHE_MAX_AGE:
|
||||
with open(path, encoding="utf-8") as f:
|
||||
if f.readline().rstrip("\n") == rev:
|
||||
return json.load(f)
|
||||
except (OSError, ValueError):
|
||||
pass
|
||||
r = client.get(f"{MSRC_BASE}/cvrf/{doc_id}", headers={"Accept": "application/json"})
|
||||
r.raise_for_status()
|
||||
doc = r.json()
|
||||
if rev:
|
||||
tmp = f"{path}.{os.getpid()}.{threading.get_ident()}.part"
|
||||
try:
|
||||
os.makedirs(_CVRF_CACHE_DIR, exist_ok=True)
|
||||
with open(tmp, "w", encoding="utf-8") as f:
|
||||
f.write(rev + "\n" + r.text)
|
||||
os.replace(tmp, path)
|
||||
except OSError as e:
|
||||
logger.debug("MSRC: caching %s failed: %s", doc_id, e)
|
||||
return doc
|
||||
|
||||
|
||||
def _cve_description(v: dict) -> Optional[str]:
|
||||
"""The flaw text, from the CVRF Note titled "Description" (HTML)."""
|
||||
for n in v.get("Notes", []) or []:
|
||||
@@ -279,12 +323,7 @@ def refresh_msrc(db: Session, months_back: Optional[int] = None,
|
||||
|
||||
for doc_id in doc_ids:
|
||||
try:
|
||||
r = client.get(f"{MSRC_BASE}/cvrf/{doc_id}",
|
||||
headers={"Accept": "application/json"})
|
||||
if r.status_code != 200:
|
||||
stats["errors"].append(f"{doc_id}: HTTP {r.status_code}")
|
||||
continue
|
||||
doc = r.json()
|
||||
doc = fetch_cvrf_doc(client, doc_id)
|
||||
except (httpx.HTTPError, ValueError) as e:
|
||||
stats["errors"].append(f"{doc_id}: {e}")
|
||||
continue
|
||||
|
||||
@@ -35,6 +35,7 @@ source/reconcile pair to maintain.
|
||||
import json
|
||||
import logging
|
||||
import re
|
||||
import time
|
||||
from datetime import datetime, timedelta
|
||||
from typing import Dict, List, Optional, Tuple
|
||||
|
||||
@@ -357,28 +358,42 @@ def load_index(db: Session) -> Optional[Dict[str, dict]]:
|
||||
return _stored(db)
|
||||
|
||||
|
||||
# A failed or partial build stores nothing, so the index stays stale and the
|
||||
# next caller would rebuild again — per asset / per sync during an outage.
|
||||
# After a failed build, callers get the stale index for this long instead.
|
||||
_BUILD_RETRY_SECONDS = 3600
|
||||
_build_failed_at: Optional[float] = None # time.monotonic()
|
||||
|
||||
|
||||
def get_index(db: Session) -> Dict[str, dict]:
|
||||
global _build_failed_at
|
||||
idx = load_index(db)
|
||||
if idx is not None:
|
||||
return idx
|
||||
if _build_failed_at is not None and time.monotonic() - _build_failed_at < _BUILD_RETRY_SECONDS:
|
||||
return _stored(db) or {}
|
||||
try:
|
||||
return build_index(db)
|
||||
built = build_index(db)
|
||||
except Exception as e:
|
||||
# Fall back to the STALE index, never to an empty one. An empty index
|
||||
# leaves every TeamViewer finding untouched this run, and the app-scan
|
||||
# reconcile reads untouched as "no longer detected" → a live flaw is
|
||||
# marked patched because teamviewer.com was briefly unreachable.
|
||||
# leaves every finding of this source untouched this run, and the
|
||||
# app-scan reconcile reads untouched as "no longer detected" → a live
|
||||
# flaw is marked patched because the vendor was briefly unreachable.
|
||||
logger.warning("teamviewer-bulletin index build failed: %s", e)
|
||||
return _stored(db) or {}
|
||||
built = _stored(db) or {}
|
||||
# build_index stores (and stamps) only a complete index.
|
||||
_build_failed_at = None if load_index(db) is not None else time.monotonic()
|
||||
return built
|
||||
|
||||
|
||||
def scan_asset(db: Session, asset, packages: list, new_ids: Optional[list] = None,
|
||||
touched: Optional[set] = None) -> int:
|
||||
touched: Optional[set] = None, index: Optional[dict] = None) -> int:
|
||||
"""Match installed TeamViewer clients against the vendor's bulletins.
|
||||
Returns findings upserted. Caller commits."""
|
||||
from app.services import app_cve_scanner_service as cpe
|
||||
|
||||
index = get_index(db)
|
||||
if index is None:
|
||||
index = get_index(db)
|
||||
if not index:
|
||||
return 0
|
||||
if new_ids is None:
|
||||
|
||||
@@ -330,6 +330,34 @@ class SSVCExploitationStatus(str):
|
||||
WIDESPREAD = "widespread" # Weit verbreitete Ausnutzung
|
||||
|
||||
|
||||
# Per-source negative cache: CVEs a source answered without a score (404,
|
||||
# empty result, no CVSS). Fresh CVEs sit there for days; without this every
|
||||
# per-device check asked NVD/GHSA again for each of them. Rate limits and
|
||||
# network errors are not answers and are never recorded.
|
||||
_NO_SCORE_TTL = 6 * 3600
|
||||
_no_score: Dict[Tuple[str, str], float] = {}
|
||||
|
||||
|
||||
def _unscored_recently(source: str, cve_id: str) -> bool:
|
||||
import time
|
||||
t = _no_score.get((source, cve_id))
|
||||
return t is not None and time.monotonic() - t < _NO_SCORE_TTL
|
||||
|
||||
|
||||
def _mark_unscored(source: str, cve_id: str) -> None:
|
||||
import time
|
||||
_no_score[(source, cve_id)] = time.monotonic()
|
||||
|
||||
|
||||
def _cvelistv5_zip() -> str:
|
||||
"""Path of the shared cvelistV5 ZIP, downloading it if stale. Raises when
|
||||
there is no usable ZIP so the cascade logs the stage as failed."""
|
||||
from app.services import cvelistv5_scan_service as c5
|
||||
if not c5._ensure_zip():
|
||||
raise RuntimeError("cvelistV5 ZIP unavailable")
|
||||
return c5._ZIP_PATH
|
||||
|
||||
|
||||
def _merge_cvss_into(
|
||||
bucket: Dict[str, "VerifiedCVEData"],
|
||||
cve_id: str,
|
||||
@@ -601,13 +629,8 @@ class VulnOverrideService:
|
||||
_ZIP_CACHE_PATH = "/tmp/truevuln-vulnrichment-cache.zip"
|
||||
_ZIP_CACHE_TTL_SECONDS = 6 * 3600
|
||||
|
||||
# cvelistV5 — official MITRE/CVE.org JSON 5 cache, ~557 MB.
|
||||
# Used as a 3rd-stage fallback after Vulnrichment + NVD because it
|
||||
# covers far more CVEs (every published CVE, not just CISA-curated
|
||||
# ones). Disk-cached for 12h to avoid hammering GitHub.
|
||||
_CVELIST_ZIP_URL = "https://github.com/CVEProject/cvelistV5/archive/refs/heads/main.zip"
|
||||
_CVELIST_CACHE_PATH = "/tmp/truevuln-cvelistv5-cache.zip"
|
||||
_CVELIST_CACHE_TTL_SECONDS = 12 * 3600
|
||||
# cvelistV5 (~557 MB) is read from the app-wide cache that
|
||||
# cvelistv5_scan_service._ensure_zip downloads (12h TTL, one writer).
|
||||
|
||||
def load_cisa_vulnrichment_data(self, cve_ids: List[str]) -> Dict[str, VerifiedCVEData]:
|
||||
"""
|
||||
@@ -652,7 +675,8 @@ class VulnOverrideService:
|
||||
# cvelistV5 (stage 3) mops up the rest.
|
||||
import os as _os
|
||||
nvd_cap = 1500 if _os.getenv("NVD_API_KEY", "").strip() else 100
|
||||
missing = [c for c in cve_ids_upper if _needs_cvss(c)]
|
||||
missing = [c for c in cve_ids_upper
|
||||
if _needs_cvss(c) and not _unscored_recently("nvd", c)]
|
||||
if missing and len(missing) <= nvd_cap:
|
||||
try:
|
||||
nvd_data = self._load_via_nvd(missing)
|
||||
@@ -692,7 +716,8 @@ class VulnOverrideService:
|
||||
# a score after NVD + cvelistV5, i.e. very fresh CVEs GHSA has but the
|
||||
# others don't yet (the gap seen in the field). Self-throttles on the
|
||||
# GitHub rate limit; a github_pat setting lifts it to 5000 req/h.
|
||||
missing = [c for c in cve_ids_upper if _needs_cvss(c)]
|
||||
missing = [c for c in cve_ids_upper
|
||||
if _needs_cvss(c) and not _unscored_recently("ghsa", c)]
|
||||
if missing:
|
||||
try:
|
||||
ghsa_data = self._load_via_ghsa(missing)
|
||||
@@ -788,6 +813,7 @@ class VulnOverrideService:
|
||||
continue
|
||||
arr = r.json() or []
|
||||
if not arr:
|
||||
_mark_unscored("ghsa", cve_id)
|
||||
continue
|
||||
adv = arr[0]
|
||||
cvss = adv.get("cvss") or {}
|
||||
@@ -800,6 +826,7 @@ class VulnOverrideService:
|
||||
score, cvss = s, sev_block[k]
|
||||
break
|
||||
if not score:
|
||||
_mark_unscored("ghsa", cve_id)
|
||||
continue
|
||||
score = float(score)
|
||||
sev = (adv.get("severity") or "").lower() or \
|
||||
@@ -849,11 +876,14 @@ class VulnOverrideService:
|
||||
"https://services.nvd.nist.gov/rest/json/cves/2.0",
|
||||
params={"cveId": cve_id},
|
||||
)
|
||||
if r.status_code == 404:
|
||||
_mark_unscored("nvd", cve_id)
|
||||
if r.status_code != 200:
|
||||
continue
|
||||
data = r.json()
|
||||
items = data.get("vulnerabilities") or []
|
||||
if not items:
|
||||
_mark_unscored("nvd", cve_id)
|
||||
continue
|
||||
cve = items[0].get("cve", {})
|
||||
metrics = cve.get("metrics", {})
|
||||
@@ -915,6 +945,7 @@ class VulnOverrideService:
|
||||
fixed_version = str(c.less_than)[:100]
|
||||
break
|
||||
if cvss_score is None and not fixed_version:
|
||||
_mark_unscored("nvd", cve_id)
|
||||
continue
|
||||
verified[cve_id] = VerifiedCVEData(
|
||||
cve_id=cve_id,
|
||||
@@ -933,7 +964,7 @@ class VulnOverrideService:
|
||||
Stage 3 fallback — official MITRE/CVE.org cvelistV5 cache.
|
||||
|
||||
Pulls https://github.com/CVEProject/cvelistV5/archive/refs/heads/main.zip
|
||||
once per 12h (disk cache at /tmp/truevuln-cvelistv5-cache.zip),
|
||||
once per 12h (shared cache, cvelistv5_scan_service._ensure_zip),
|
||||
then walks the wanted CVE files in-place. Same CVE-5 JSON shape
|
||||
as Vulnrichment so _parse_vulnrichment_record handles both.
|
||||
|
||||
@@ -945,11 +976,6 @@ class VulnOverrideService:
|
||||
cvelistV5 is exhaustive (~250k+ CVEs). Downside is size
|
||||
(~557 MB). Disk cache amortises the download cost across runs.
|
||||
"""
|
||||
import httpx
|
||||
import os
|
||||
import re
|
||||
import threading
|
||||
import time
|
||||
import zipfile
|
||||
|
||||
verified: Dict[str, VerifiedCVEData] = {}
|
||||
@@ -964,48 +990,7 @@ class VulnOverrideService:
|
||||
bucket = f"{int(num) // 1000}xxx"
|
||||
return f"cvelistV5-main/cves/{year}/{bucket}/{cve_id}.json"
|
||||
|
||||
cache_path = self._CVELIST_CACHE_PATH
|
||||
ttl = self._CVELIST_CACHE_TTL_SECONDS
|
||||
cache_fresh = (
|
||||
os.path.exists(cache_path)
|
||||
and (time.time() - os.path.getmtime(cache_path)) < ttl
|
||||
and os.path.getsize(cache_path) > 100_000_000 # > 100 MB sanity
|
||||
)
|
||||
if not cache_fresh:
|
||||
logger.info(
|
||||
"cvelistV5: downloading fresh ZIP (cache miss / stale) to %s",
|
||||
cache_path,
|
||||
)
|
||||
# Per-writer name — see cvelistv5_scan_service._ensure_zip.
|
||||
tmp_path = f"{cache_path}.{os.getpid()}.{threading.get_ident()}.part"
|
||||
try:
|
||||
with httpx.Client(timeout=httpx.Timeout(120.0, connect=15.0),
|
||||
follow_redirects=True) as client:
|
||||
with client.stream("GET", self._CVELIST_ZIP_URL) as resp:
|
||||
resp.raise_for_status()
|
||||
with open(tmp_path, "wb") as f:
|
||||
for chunk in resp.iter_bytes(chunk_size=1024 * 512):
|
||||
f.write(chunk)
|
||||
os.replace(tmp_path, cache_path)
|
||||
logger.info(
|
||||
"cvelistV5: download complete (%.1f MB)",
|
||||
os.path.getsize(cache_path) / 1_048_576,
|
||||
)
|
||||
except Exception:
|
||||
# leave any prior cache file alone, drop the partial
|
||||
if os.path.exists(tmp_path):
|
||||
try:
|
||||
os.remove(tmp_path)
|
||||
except Exception:
|
||||
pass
|
||||
raise
|
||||
else:
|
||||
logger.info(
|
||||
"cvelistV5: using cached ZIP (%.1f MB, %.1fh old)",
|
||||
os.path.getsize(cache_path) / 1_048_576,
|
||||
(time.time() - os.path.getmtime(cache_path)) / 3600,
|
||||
)
|
||||
|
||||
cache_path = _cvelistv5_zip()
|
||||
with zipfile.ZipFile(cache_path) as zf:
|
||||
names = set(zf.namelist())
|
||||
hits = 0
|
||||
@@ -1042,9 +1027,6 @@ class VulnOverrideService:
|
||||
thousands of per-CVE HTTP round-trips. Dates come from the
|
||||
authoritative cveMetadata.datePublished / .dateUpdated.
|
||||
"""
|
||||
import os
|
||||
import re
|
||||
import time
|
||||
import zipfile
|
||||
|
||||
out: Dict[str, dict] = {}
|
||||
@@ -1058,15 +1040,7 @@ class VulnOverrideService:
|
||||
year, num = m.group(1), m.group(2)
|
||||
return f"cvelistV5-main/cves/{year}/{int(num) // 1000}xxx/{cve_id}.json"
|
||||
|
||||
cache_path = self._CVELIST_CACHE_PATH
|
||||
cache_fresh = (
|
||||
os.path.exists(cache_path)
|
||||
and (time.time() - os.path.getmtime(cache_path)) < self._CVELIST_CACHE_TTL_SECONDS
|
||||
and os.path.getsize(cache_path) > 100_000_000
|
||||
)
|
||||
if not cache_fresh:
|
||||
# Reuse the cascade's downloader (handles streaming + .part swap).
|
||||
self._download_cvelistv5_zip()
|
||||
cache_path = _cvelistv5_zip()
|
||||
|
||||
with zipfile.ZipFile(cache_path) as zf:
|
||||
names = set(zf.namelist())
|
||||
@@ -1086,36 +1060,6 @@ class VulnOverrideService:
|
||||
logger.info("cvelistV5 dates: %d/%d CVEs found in ZIP", len(out), len(wanted))
|
||||
return out
|
||||
|
||||
def _download_cvelistv5_zip(self) -> None:
|
||||
"""Stream the cvelistV5 main.zip to the shared disk cache (12h TTL)."""
|
||||
import httpx
|
||||
import os
|
||||
import threading
|
||||
|
||||
cache_path = self._CVELIST_CACHE_PATH
|
||||
# Per-writer name — a shared .part interleaves concurrent downloads
|
||||
# (see cvelistv5_scan_service._ensure_zip).
|
||||
tmp_path = f"{cache_path}.{os.getpid()}.{threading.get_ident()}.part"
|
||||
logger.info("cvelistV5: downloading fresh ZIP to %s", cache_path)
|
||||
try:
|
||||
with httpx.Client(timeout=httpx.Timeout(120.0, connect=15.0),
|
||||
follow_redirects=True) as client:
|
||||
with client.stream("GET", self._CVELIST_ZIP_URL) as resp:
|
||||
resp.raise_for_status()
|
||||
with open(tmp_path, "wb") as f:
|
||||
for chunk in resp.iter_bytes(chunk_size=1024 * 512):
|
||||
f.write(chunk)
|
||||
os.replace(tmp_path, cache_path)
|
||||
logger.info("cvelistV5: download complete (%.1f MB)",
|
||||
os.path.getsize(cache_path) / 1_048_576)
|
||||
except Exception:
|
||||
if os.path.exists(tmp_path):
|
||||
try:
|
||||
os.remove(tmp_path)
|
||||
except Exception:
|
||||
pass
|
||||
raise
|
||||
|
||||
def _load_via_zip_snapshot(self, cve_ids: List[str]) -> Dict[str, VerifiedCVEData]:
|
||||
"""One ZIP download (disk-cached, see _ZIP_CACHE_PATH) → walk locally
|
||||
for the requested CVE files. Two orders of magnitude faster than
|
||||
|
||||
@@ -83,6 +83,7 @@ def _with_fresh_zip(tmp_path, force):
|
||||
return ok, calls
|
||||
finally:
|
||||
c5._ZIP_PATH = old_path
|
||||
c5._zip_retry_after = 0.0 # the fake body is not a ZIP → backoff
|
||||
if old_httpx is not None:
|
||||
sys.modules["httpx"] = old_httpx
|
||||
else:
|
||||
|
||||
@@ -0,0 +1,123 @@
|
||||
"""One cvelistV5 ZIP downloader — run: python tests/test_cvelistv5_single_download.py
|
||||
|
||||
The ~557 MB cvelistV5 main.zip had three downloaders (the index build, the
|
||||
CVSS cascade's stage 3 and the date backfill), none of which saw the others:
|
||||
callers arriving together each pulled the whole ZIP, and after a failed
|
||||
download every later caller retried at once. All of them now go through
|
||||
cvelistv5_scan_service._ensure_zip, which serialises downloads, re-checks
|
||||
freshness after the lock, rejects non-ZIP bodies and backs off after a failure.
|
||||
"""
|
||||
import io
|
||||
import os
|
||||
import sys
|
||||
import tempfile
|
||||
import threading
|
||||
import time
|
||||
import types
|
||||
import zipfile
|
||||
|
||||
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
|
||||
os.environ.setdefault("JWT_SECRET_KEY", "test-key-test-key-test-key-test-key")
|
||||
|
||||
from app.services import cvelistv5_scan_service as c5 # noqa: E402
|
||||
from app.services.vuln_override_service import VulnOverrideService # noqa: E402
|
||||
|
||||
_buf = io.BytesIO()
|
||||
with zipfile.ZipFile(_buf, "w") as _zf:
|
||||
_zf.writestr("cvelistV5-main/cves/2026/1xxx/CVE-2026-1234.json",
|
||||
'{"cveMetadata": {"datePublished": "2026-10-01T00:00:00"}}')
|
||||
ZIP_BYTES = _buf.getvalue()
|
||||
|
||||
|
||||
def _fake_httpx(calls, status=200, body=ZIP_BYTES):
|
||||
mod = types.ModuleType("httpx")
|
||||
|
||||
class _Resp:
|
||||
def raise_for_status(self):
|
||||
if status >= 400:
|
||||
raise RuntimeError(f"HTTP {status}")
|
||||
|
||||
def iter_bytes(self, chunk_size=0):
|
||||
time.sleep(0.2) # long enough for a second caller to queue up
|
||||
yield body
|
||||
|
||||
def __enter__(self):
|
||||
return self
|
||||
|
||||
def __exit__(self, *a):
|
||||
return False
|
||||
|
||||
class Client:
|
||||
def __init__(self, *a, **k):
|
||||
pass
|
||||
|
||||
def __enter__(self):
|
||||
return self
|
||||
|
||||
def __exit__(self, *a):
|
||||
return False
|
||||
|
||||
def stream(self, method, url):
|
||||
calls.append(url)
|
||||
return _Resp()
|
||||
|
||||
mod.Client, mod.Timeout = Client, lambda *a, **k: None
|
||||
return mod
|
||||
|
||||
|
||||
def _run(fn, **fake):
|
||||
calls = []
|
||||
saved = (c5._ZIP_PATH, c5._ZIP_MIN_BYTES, c5._zip_retry_after, sys.modules.get("httpx"))
|
||||
tmp = tempfile.mkdtemp()
|
||||
c5._ZIP_PATH = os.path.join(tmp, "cvelist.zip")
|
||||
c5._ZIP_MIN_BYTES = 0
|
||||
c5._zip_retry_after = 0.0
|
||||
sys.modules["httpx"] = _fake_httpx(calls, **fake)
|
||||
try:
|
||||
return fn(), calls, os.listdir(tmp)
|
||||
finally:
|
||||
c5._ZIP_PATH, c5._ZIP_MIN_BYTES, c5._zip_retry_after, httpx = saved
|
||||
if httpx is not None:
|
||||
sys.modules["httpx"] = httpx
|
||||
|
||||
|
||||
def test_concurrent_callers_share_one_download():
|
||||
def both():
|
||||
results = []
|
||||
t = threading.Thread(target=lambda: results.append(c5._ensure_zip()))
|
||||
t.start()
|
||||
# The cascade and the date backfill both read the same shared file.
|
||||
dates = VulnOverrideService(None).load_cve_dates_via_zip(["CVE-2026-1234"])
|
||||
t.join()
|
||||
return results, dates
|
||||
(results, dates), calls, files = _run(both)
|
||||
assert results == [True], results
|
||||
assert dates["CVE-2026-1234"]["published"] == "2026-10-01T00:00:00", dates
|
||||
assert len(calls) == 1, f"downloaded {len(calls)}x"
|
||||
assert files == ["cvelist.zip"], files
|
||||
|
||||
|
||||
def test_failed_download_backs_off():
|
||||
def twice():
|
||||
return c5._ensure_zip(), c5._ensure_zip()
|
||||
results, calls, files = _run(twice, status=503)
|
||||
assert results == (False, False), results
|
||||
assert len(calls) == 1, f"retried immediately: {len(calls)} downloads"
|
||||
assert files == [], files
|
||||
|
||||
|
||||
def test_error_page_never_becomes_the_cache():
|
||||
results, calls, files = _run(c5._ensure_zip, body=b"<html>rate limited</html>")
|
||||
assert results is False
|
||||
assert files == [], files
|
||||
|
||||
|
||||
def demo():
|
||||
test_concurrent_callers_share_one_download()
|
||||
test_failed_download_backs_off()
|
||||
test_error_page_never_becomes_the_cache()
|
||||
print("ok one cvelistV5 download, shared, validated, backed off after failure")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
demo()
|
||||
@@ -0,0 +1,106 @@
|
||||
"""A failed index/catalog build is not retried by every caller — run:
|
||||
python tests/test_failed_index_build_backoff.py
|
||||
|
||||
A failed or partial build stores (and stamps) nothing, which keeps the stale
|
||||
index safe — but it also left the cache "stale", so during a vendor/GitHub
|
||||
outage every per-asset scan and every per-sync enrichment rebuilt the whole
|
||||
index again (MFSA ~180 GitHub requests, EUVD dozens of pages). Now the first
|
||||
failure is remembered and later callers get the stale index for an hour.
|
||||
"""
|
||||
import json
|
||||
import os
|
||||
import sys
|
||||
from datetime import datetime, timedelta
|
||||
|
||||
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
|
||||
os.environ.setdefault("JWT_SECRET_KEY", "test-key-test-key-test-key-test-key")
|
||||
|
||||
import httpx # noqa: E402
|
||||
from sqlalchemy import create_engine # noqa: E402
|
||||
from sqlalchemy.orm import sessionmaker # noqa: E402
|
||||
|
||||
import app.models # noqa: F401,E402
|
||||
from app.models.base import Base # noqa: E402
|
||||
from app.models.setting import Setting # noqa: E402
|
||||
from app.services import enrichment_service as enr # noqa: E402
|
||||
from app.services import github_repo_advisory_service as ghra # noqa: E402
|
||||
from app.services import mozilla_advisory_service as mfsa # noqa: E402
|
||||
from app.services import teamviewer_bulletin_service as tv # noqa: E402
|
||||
|
||||
OLD = (datetime.now() - timedelta(days=3)).isoformat()
|
||||
|
||||
|
||||
def _db(rows):
|
||||
engine = create_engine("sqlite://")
|
||||
Base.metadata.create_all(engine)
|
||||
db = sessionmaker(bind=engine)()
|
||||
for k, v in rows.items():
|
||||
db.add(Setting(key=k, value=v))
|
||||
db.commit()
|
||||
return db
|
||||
|
||||
|
||||
def _index_backoff(mod, stale):
|
||||
db = _db({mod.INDEX_SETTING: json.dumps(stale), mod.INDEX_TS_SETTING: OLD})
|
||||
builds = []
|
||||
saved = (mod.build_index, mod._build_failed_at)
|
||||
# Outage: stores nothing; like the real builders, MFSA answers {} and the
|
||||
# others the last stored index.
|
||||
stored = getattr(mod, "_stored", lambda db: {})
|
||||
mod.build_index = lambda db, *a, **k: builds.append(1) or stored(db) or {}
|
||||
mod._build_failed_at = None
|
||||
try:
|
||||
got = [mod.get_index(db) for _ in range(3)]
|
||||
finally:
|
||||
mod.build_index, mod._build_failed_at = saved
|
||||
assert got == [stale] * 3, got
|
||||
assert len(builds) == 1, f"{mod.__name__}: rebuilt {len(builds)}x"
|
||||
|
||||
|
||||
def test_vendor_indexes_back_off():
|
||||
_index_backoff(mfsa, {"CVE-2026-1": {"sev": "high"}})
|
||||
_index_backoff(tv, {"TV-2026-1": {"cves": []}})
|
||||
_index_backoff(ghra, {"notepad-plus-plus/notepad-plus-plus": []})
|
||||
|
||||
|
||||
def test_kev_refresh_backs_off():
|
||||
stale = {"CVE-2026-2": {"date_added": "2026-01-01"}}
|
||||
db = _db({enr.KEV_CACHE_KEY: json.dumps(stale), enr.KEV_CACHE_TS_KEY: OLD})
|
||||
calls = []
|
||||
|
||||
class _Down:
|
||||
def __init__(self, *a, **k):
|
||||
pass
|
||||
|
||||
def __enter__(self):
|
||||
return self
|
||||
|
||||
def __exit__(self, *a):
|
||||
return False
|
||||
|
||||
def get(self, url, **k):
|
||||
calls.append(url)
|
||||
raise httpx.ConnectError("down")
|
||||
|
||||
saved = httpx.Client
|
||||
httpx.Client = _Down
|
||||
enr._fetch_failed_at.clear()
|
||||
try:
|
||||
got = [enr.fetch_kev_catalog(db) for _ in range(3)]
|
||||
forced = enr.fetch_kev_catalog(db, force_refresh=True)
|
||||
finally:
|
||||
httpx.Client = saved
|
||||
enr._fetch_failed_at.clear()
|
||||
assert got == [stale] * 3 and forced == stale
|
||||
# 1 for the first failure, 1 for the forced refresh (the nightly/GUI button).
|
||||
assert len(calls) == 2, calls
|
||||
|
||||
|
||||
def demo():
|
||||
test_vendor_indexes_back_off()
|
||||
test_kev_refresh_backs_off()
|
||||
print("ok a failed index build is retried after the backoff, not per caller")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
demo()
|
||||
@@ -28,6 +28,7 @@ from sqlalchemy.orm import sessionmaker # noqa: E402
|
||||
import app.models # noqa: F401,E402 (registers every mapper)
|
||||
from app.models.asset import Asset, AssetSource # noqa: E402
|
||||
from app.models.vulnerability import Vulnerability # noqa: E402
|
||||
from app.services import cvelistv5_scan_service as c5 # noqa: E402
|
||||
from app.services import m365_service # noqa: E402
|
||||
from app.services.vuln_override_service import VulnOverrideService # noqa: E402
|
||||
|
||||
@@ -132,12 +133,12 @@ def test_devices_share_one_download():
|
||||
tmp = tempfile.mkdtemp()
|
||||
saved = (httpx.Client, m365_service.fetch_security_data,
|
||||
m365_service._os_owned_cve_ids,
|
||||
VulnOverrideService._CVELIST_CACHE_PATH,
|
||||
c5._ZIP_PATH, c5._zip_retry_after,
|
||||
getattr(VulnOverrideService, "_ZIP_CACHE_PATH", None))
|
||||
httpx.Client = _FakeClient
|
||||
m365_service.fetch_security_data = lambda db, force_refresh=False: RELEASES
|
||||
m365_service._os_owned_cve_ids = lambda db: set()
|
||||
VulnOverrideService._CVELIST_CACHE_PATH = os.path.join(tmp, "cvelist.zip")
|
||||
c5._ZIP_PATH = os.path.join(tmp, "cvelist.zip")
|
||||
VulnOverrideService._ZIP_CACHE_PATH = os.path.join(tmp, "vulnrichment.zip")
|
||||
import time as _time
|
||||
_sleep, _time.sleep = _time.sleep, lambda s: None
|
||||
@@ -155,7 +156,7 @@ def test_devices_share_one_download():
|
||||
_time.sleep = _sleep
|
||||
(httpx.Client, m365_service.fetch_security_data,
|
||||
m365_service._os_owned_cve_ids,
|
||||
VulnOverrideService._CVELIST_CACHE_PATH, zcache) = saved
|
||||
c5._ZIP_PATH, c5._zip_retry_after, zcache) = saved
|
||||
VulnOverrideService._ZIP_CACHE_PATH = zcache
|
||||
|
||||
zip_pulls = CALLS.count(VulnOverrideService._ZIP_URL)
|
||||
|
||||
@@ -0,0 +1,90 @@
|
||||
"""MSRC CVRF documents are downloaded once per revision — run:
|
||||
python tests/test_msrc_cvrf_shared_cache.py
|
||||
|
||||
The 01:30 fixed-build index (msrc_scan_service.build_product_index) and the
|
||||
03:50 / Sunday 04:40 ingest (msrc_service.refresh_msrc) each pulled the same
|
||||
monthly CVRF documents (up to 18, several MB each) from MSRC. They now share a
|
||||
disk cache keyed by the document's revision from /updates: the second walk
|
||||
reuses the download, and a revised document is fetched again.
|
||||
"""
|
||||
import json
|
||||
import os
|
||||
import sys
|
||||
import tempfile
|
||||
|
||||
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
|
||||
os.environ.setdefault("JWT_SECRET_KEY", "test-key-test-key-test-key-test-key")
|
||||
|
||||
import httpx # noqa: E402
|
||||
from sqlalchemy import create_engine # noqa: E402
|
||||
from sqlalchemy.orm import sessionmaker # noqa: E402
|
||||
|
||||
import app.models # noqa: F401,E402
|
||||
from app.models.base import Base # noqa: E402
|
||||
from app.services import msrc_scan_service, msrc_service # noqa: E402
|
||||
|
||||
DOC = {"ProductTree": {"FullProductName": []}, "Vulnerability": []}
|
||||
REVISION = {"2026-Sep": "2026-09-09T07:00:00Z", "2026-Oct": "2026-10-13T07:00:00Z"}
|
||||
CALLS = []
|
||||
|
||||
|
||||
class _Resp:
|
||||
def __init__(self, payload):
|
||||
self._p = payload
|
||||
self.status_code = 200
|
||||
self.text = json.dumps(payload)
|
||||
|
||||
def raise_for_status(self):
|
||||
pass
|
||||
|
||||
def json(self):
|
||||
return self._p
|
||||
|
||||
|
||||
class _Client:
|
||||
def __init__(self, *a, **k):
|
||||
pass
|
||||
|
||||
def __enter__(self):
|
||||
return self
|
||||
|
||||
def __exit__(self, *a):
|
||||
return False
|
||||
|
||||
def get(self, url, headers=None, **k):
|
||||
CALLS.append(url)
|
||||
if url.endswith("/updates"):
|
||||
return _Resp({"value": [{"ID": d, "CurrentReleaseDate": r}
|
||||
for d, r in REVISION.items()]})
|
||||
return _Resp(DOC)
|
||||
|
||||
|
||||
def test_second_walk_reuses_the_documents():
|
||||
db = sessionmaker(bind=create_engine("sqlite://"))()
|
||||
Base.metadata.create_all(db.get_bind())
|
||||
saved = (httpx.Client, msrc_service._CVRF_CACHE_DIR)
|
||||
httpx.Client = _Client
|
||||
msrc_service._CVRF_CACHE_DIR = tempfile.mkdtemp()
|
||||
CALLS.clear()
|
||||
try:
|
||||
msrc_scan_service.build_product_index(db, months_back=2) # 01:30
|
||||
msrc_service.refresh_msrc(db, months_back=2) # 03:50
|
||||
docs = [u for u in CALLS if "/cvrf/2026-" in u]
|
||||
assert len(docs) == 2, f"each document once, got {docs}"
|
||||
|
||||
REVISION["2026-Oct"] = "2026-10-15T07:00:00Z" # Microsoft revised it
|
||||
CALLS.clear()
|
||||
msrc_service.refresh_msrc(db, months_back=2)
|
||||
docs = [u for u in CALLS if "/cvrf/2026-" in u]
|
||||
assert docs == [f"{msrc_service.MSRC_BASE}/cvrf/2026-Oct"], docs
|
||||
finally:
|
||||
httpx.Client, msrc_service._CVRF_CACHE_DIR = saved
|
||||
|
||||
|
||||
def demo():
|
||||
test_second_walk_reuses_the_documents()
|
||||
print("ok CVRF documents shared between the index build and the ingest")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
demo()
|
||||
@@ -0,0 +1,106 @@
|
||||
"""Per-CVE lookups that keep missing are not repeated — run:
|
||||
python tests/test_negative_cve_cache.py
|
||||
|
||||
MSRC publishes M365 CVEs days before CVE.org, NVD and GHSA carry them. Every
|
||||
Intune device sync asked cvelistV5 for the title (404) and NVD + GHSA for a
|
||||
score (nothing) again for each such CVE. A real "no" from the source is now
|
||||
remembered for a few hours; an outage or rate limit is not an answer and is
|
||||
retried.
|
||||
"""
|
||||
import json
|
||||
import os
|
||||
import sys
|
||||
|
||||
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
|
||||
os.environ.setdefault("JWT_SECRET_KEY", "test-key-test-key-test-key-test-key")
|
||||
|
||||
import httpx # noqa: E402
|
||||
|
||||
from app.services import cvelistv5_scan_service as c5 # noqa: E402
|
||||
from app.services import m365_service # noqa: E402
|
||||
from app.services import vuln_override_service as vos # noqa: E402
|
||||
|
||||
CVE = "CVE-2026-99123"
|
||||
CALLS = []
|
||||
|
||||
|
||||
def _client(status_for):
|
||||
class _Resp:
|
||||
def __init__(self, url, status):
|
||||
self.url, self.status_code, self.headers = url, status, {}
|
||||
|
||||
def json(self):
|
||||
return [] if "api.github.com" in self.url else {"vulnerabilities": []}
|
||||
|
||||
class Client:
|
||||
def __init__(self, *a, **k):
|
||||
pass
|
||||
|
||||
def __enter__(self):
|
||||
return self
|
||||
|
||||
def __exit__(self, *a):
|
||||
return False
|
||||
|
||||
def get(self, url, params=None, **k):
|
||||
full = url + ("?" + json.dumps(params) if params else "")
|
||||
CALLS.append(full)
|
||||
return _Resp(full, status_for(full))
|
||||
return Client
|
||||
|
||||
|
||||
def _patched(status_for, fn):
|
||||
saved = (httpx.Client, c5._ensure_zip)
|
||||
httpx.Client = _client(status_for)
|
||||
c5._ensure_zip = lambda force=False: False # stage 3: no ZIP, not under test
|
||||
m365_service._text_misses.clear()
|
||||
vos._no_score.clear()
|
||||
CALLS.clear()
|
||||
try:
|
||||
fn()
|
||||
fn()
|
||||
return list(CALLS)
|
||||
finally:
|
||||
httpx.Client, c5._ensure_zip = saved
|
||||
|
||||
|
||||
def _m365_text():
|
||||
assert m365_service._fetch_cve_title_desc(CVE) == (None, None)
|
||||
|
||||
|
||||
def _cascade():
|
||||
assert vos.VulnOverrideService(None).load_cisa_vulnrichment_data([CVE]) == {}
|
||||
|
||||
|
||||
def test_m365_text_404_is_remembered():
|
||||
calls = _patched(lambda url: 404, _m365_text)
|
||||
assert len(calls) == 1, calls
|
||||
|
||||
|
||||
def test_m365_text_outage_is_retried():
|
||||
calls = _patched(lambda url: 503, _m365_text)
|
||||
assert len(calls) == 2, calls
|
||||
|
||||
|
||||
def test_cascade_does_not_reask_nvd_and_ghsa():
|
||||
calls = _patched(lambda url: 200 if "api.github.com" in url else 404, _cascade)
|
||||
nvd = [u for u in calls if "nvd.nist.gov" in u]
|
||||
ghsa = [u for u in calls if "api.github.com" in u]
|
||||
assert len(nvd) == 1 and len(ghsa) == 1, calls
|
||||
|
||||
|
||||
def test_cascade_retries_after_rate_limit():
|
||||
calls = _patched(lambda url: 429 if "nvd.nist.gov" in url else 404, _cascade)
|
||||
assert len([u for u in calls if "nvd.nist.gov" in u]) == 2, calls
|
||||
|
||||
|
||||
def demo():
|
||||
test_m365_text_404_is_remembered()
|
||||
test_m365_text_outage_is_retried()
|
||||
test_cascade_does_not_reask_nvd_and_ghsa()
|
||||
test_cascade_retries_after_rate_limit()
|
||||
print("ok per-CVE misses are cached; outages are retried")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
demo()
|
||||
Reference in New Issue
Block a user