Compare commits
2
Commits
6c8cf7dd21
...
ece21b9a00
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
ece21b9a00 | ||
|
|
2921278f5b |
+2
-1
@@ -186,8 +186,9 @@ SSO/LDAP login. `RoleMapper` re-evaluates role on every login from
|
||||
| `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; runs `kev_alert_check` straight after, since the catalogs just moved |
|
||||
| `kev_alert_check` | hourly, :25 | Mails actively-exploited CVEs (sources per `kev_alert_sources`, default CISA KEV + ENISA EUVD) that have OPEN findings on active assets; honors `kev_alert_enabled`, idempotent via `kev_alert_state` |
|
||||
| `vuln_index_refresh_nightly` | 01:30 UTC | Rebuilds the cvelistV5 (+MFSA), GitHub-advisory, TeamViewer and IGEL-ISN indexes before the Intune/Defender, vCenter and IGEL syncs read them |
|
||||
| `compliance_sca_nightly` | 02:00 UTC | Wazuh SCA pull for every linked asset |
|
||||
| `intune_sync_nightly` | 02:10 UTC | Intune managed devices → assets + OS-EOL |
|
||||
| `intune_sync_nightly` | 02:10 UTC | Intune managed devices → assets + OS-EOL + per-device app-CVE scan, Defender TVM CVEs when enabled |
|
||||
| `vcenter_sync_nightly` | 02:20 UTC | vCenter + ESXi hosts → assets, EOL, vSphere CVEs (refreshes the build catalog first) |
|
||||
| `igel_sync_nightly` | 02:30 UTC | IGEL UMS server + endpoint devices → assets, IGEL OS CVEs |
|
||||
| `vulnrichment_nightly` | 03:00 UTC | 3-stage CVSS/SSVC/fixed_version cascade |
|
||||
|
||||
@@ -125,6 +125,7 @@ stay uncorrected until the following night.
|
||||
|
||||
| Time | Job | Stage |
|
||||
|------|-----|-------|
|
||||
| 01:30 | Vulnerability index refresh (cvelistV5 + MFSA, GitHub advisories, TeamViewer, IGEL ISN) — before every sync that decides from them | catalog |
|
||||
| 02:00 | Wazuh SCA compliance refresh | inventory |
|
||||
| 02:10 | Intune inventory sync | inventory |
|
||||
| 02:30 | Network exposure + risk dimensions | inventory |
|
||||
|
||||
+82
-28
@@ -711,6 +711,71 @@ def m365_check_nightly():
|
||||
db.close()
|
||||
|
||||
|
||||
def vuln_index_refresh_nightly():
|
||||
"""Rebuild every vulnerability index a nightly sync decides from — BEFORE
|
||||
the first sync of the night, not inside the last scan.
|
||||
|
||||
01:30 UTC. The Intune sync at 02:10 runs the app-CVE scan for each of its
|
||||
devices (and the Defender TVM pull), the IGEL sync at 02:30 scans its
|
||||
endpoints and the UMS server, the vCenter sync at 02:20 its hosts — and
|
||||
every one of them reads the cvelistV5 index and the vendor indexes as
|
||||
stored. Those used to be rebuilt by the app-CVE job at 03:20, i.e. AFTER
|
||||
all three syncs, so an Intune-only device was matched against yesterday's
|
||||
catalogue every night (the 03:20 pass over all assets caught up, but only
|
||||
once it got there — hours later on a large estate). Now:
|
||||
|
||||
01:30 this job cvelistV5 (+MFSA), GitHub advisories,
|
||||
TeamViewer bulletins, IGEL ISNs
|
||||
02:10 Intune/Defender sync reads them fresh
|
||||
02:20 vCenter sync
|
||||
02:30 IGEL sync
|
||||
03:20 app-CVE scan reads the same index; builds only if missing
|
||||
|
||||
Each rebuild is independent and a failed one keeps the cached index (see
|
||||
the build functions), so an offline vendor site costs that source a night,
|
||||
never the others and never the syncs.
|
||||
|
||||
ponytail: the 40-minute gap is the guarantee. The elapsed time is logged;
|
||||
if this job ever runs past 02:10, chain the Intune sync onto it instead of
|
||||
widening the gap.
|
||||
"""
|
||||
import time as _time
|
||||
from app.services import (cvelistv5_scan_service, github_repo_advisory_service,
|
||||
igel_isn_service, teamviewer_bulletin_service)
|
||||
db = SessionLocal()
|
||||
t0 = _time.monotonic()
|
||||
try:
|
||||
# force_fresh: the /tmp ZIP is shared with the threat-intel refresh,
|
||||
# which runs on an interval anchored to app startup — under the plain
|
||||
# 12h TTL this build kept reusing an afternoon snapshot and missed the
|
||||
# evening's CVEs for a full extra night (Chrome 151.0.7922.169).
|
||||
try:
|
||||
cvelistv5_scan_service.build_product_index(db, force_fresh=True)
|
||||
except Exception as e:
|
||||
logger.warning("cvelistV5 index build failed (non-fatal, cache kept): %s", e)
|
||||
# Vendor indexes: all cache 24h and used to be refreshed by whoever
|
||||
# asked first — the scan itself, one build-duration short of the TTL
|
||||
# every night, so they only ever cleared it every OTHER night (the
|
||||
# MFSA bug, 978d4f3). Notepad++, Wazuh and the IGEL ISNs without a
|
||||
# CVE reach no other source at all; TeamViewer publishes days before
|
||||
# NVD — a night late is a night blind.
|
||||
for _mod, _label in ((github_repo_advisory_service, "repo-advisory"),
|
||||
(teamviewer_bulletin_service, "teamviewer-bulletin"),
|
||||
(igel_isn_service, "igel-isn")):
|
||||
try:
|
||||
_mod.build_index(db)
|
||||
except Exception as e:
|
||||
logger.warning("%s index build failed (non-fatal, cache kept): %s",
|
||||
_label, e)
|
||||
logger.info("Vulnerability index refresh done in %.0fs — the 02:10 "
|
||||
"Intune/Defender sync reads these", _time.monotonic() - t0)
|
||||
except Exception as e:
|
||||
logger.error("Vulnerability index refresh failed: %s", e)
|
||||
db.rollback()
|
||||
finally:
|
||||
db.close()
|
||||
|
||||
|
||||
def app_cve_scan_nightly():
|
||||
"""Built-in app→CVE scanner for every asset with software inventory.
|
||||
|
||||
@@ -719,38 +784,15 @@ def app_cve_scan_nightly():
|
||||
closes the coverage gap for Intune-only / mobile devices that have no
|
||||
real scanner. Source 'app-scan'; cross-confirms with the other scanners.
|
||||
|
||||
Slotted at 03:25 UTC — after M365 (03:20), before the audit prune (03:30).
|
||||
Slotted at 03:20 UTC. The cvelistV5 and vendor indexes it decides from
|
||||
are rebuilt at 01:30 by vuln_index_refresh_nightly — before the Intune,
|
||||
vCenter and IGEL syncs that read them too; run_app_cve_scan builds the
|
||||
cvelistV5 index itself only when none is stored yet.
|
||||
Cache (TTL 7d) keeps OSV/NVD load bounded; NVD_API_KEY recommended.
|
||||
"""
|
||||
from app.services import (app_cve_scanner_service, cvelistv5_scan_service,
|
||||
github_repo_advisory_service,
|
||||
teamviewer_bulletin_service)
|
||||
from app.services import app_cve_scanner_service
|
||||
db = SessionLocal()
|
||||
try:
|
||||
# Rebuild the cvelistV5 reverse index first (one ~557 MB ZIP walk) so
|
||||
# the scan below has fresh product→CVE ranges for curated software.
|
||||
# force_fresh: the /tmp ZIP is shared with the threat-intel refresh,
|
||||
# which runs on an interval anchored to app startup — under the plain
|
||||
# 12h TTL this build kept reusing an afternoon snapshot and missed the
|
||||
# evening's CVEs for a full extra night (Chrome 151.0.7922.169).
|
||||
try:
|
||||
cvelistv5_scan_service.build_product_index(db, force_fresh=True)
|
||||
except Exception as e:
|
||||
logger.warning("cvelistV5 index build failed (non-fatal): %s", e)
|
||||
# Same for the two vendor indexes the scan below decides from. Both
|
||||
# cache 24h and used to be refreshed by whoever asked first — which is
|
||||
# the scan itself, one build-duration short of the TTL every night, so
|
||||
# they only ever cleared it every OTHER night (the MFSA bug, 978d4f3).
|
||||
# Notepad++ and Wazuh advisories reach no other source at all, and
|
||||
# TeamViewer publishes days before NVD, so a night late is a night
|
||||
# blind. A failed rebuild keeps the cached index (see build_index).
|
||||
for _mod, _label in ((github_repo_advisory_service, "repo-advisory"),
|
||||
(teamviewer_bulletin_service, "teamviewer-bulletin")):
|
||||
try:
|
||||
_mod.build_index(db)
|
||||
except Exception as e:
|
||||
logger.warning("%s index build failed (non-fatal, cache kept): %s",
|
||||
_label, e)
|
||||
# FP-suppression is part of run_app_cve_scan itself now, so every way of
|
||||
# starting a scan — nightly, GUI, single asset — produces the same
|
||||
# result. It used to hang off this job alone.
|
||||
@@ -1280,6 +1322,18 @@ def start_scheduler():
|
||||
)
|
||||
|
||||
# Nightly Microsoft Intune device/inventory sync (02:10 UTC).
|
||||
# Nightly vulnerability index refresh (01:30 UTC) — cvelistV5 (+MFSA),
|
||||
# GitHub advisories, TeamViewer bulletins, IGEL ISNs. First job of the
|
||||
# scan night on purpose: the Intune/Defender sync at 02:10 decides from
|
||||
# these, and it can run for hours on a large estate.
|
||||
scheduler.add_job(
|
||||
vuln_index_refresh_nightly,
|
||||
trigger=CronTrigger(hour=1, minute=30),
|
||||
id="vuln_index_refresh_nightly",
|
||||
name="Nightly Vulnerability Index Refresh",
|
||||
replace_existing=True,
|
||||
)
|
||||
|
||||
scheduler.add_job(
|
||||
intune_sync_nightly,
|
||||
trigger=CronTrigger(hour=2, minute=10),
|
||||
|
||||
@@ -1449,6 +1449,13 @@ def run_app_cve_scan(db: Session, asset_id: Optional[int] = None) -> dict:
|
||||
except Exception as e:
|
||||
logger.warning("app-cve: cvelistV5 index build/load failed: %s", e)
|
||||
|
||||
isn_index = {}
|
||||
try:
|
||||
from app.services import igel_isn_service
|
||||
isn_index = igel_isn_service.get_index(db)
|
||||
except Exception as e:
|
||||
logger.warning("app-cve: IGEL ISN index load failed: %s", e)
|
||||
|
||||
q = db.query(Asset)
|
||||
if asset_id is not None:
|
||||
q = q.filter(Asset.id == asset_id)
|
||||
@@ -1510,6 +1517,19 @@ def run_app_cve_scan(db: Session, asset_id: Optional[int] = None) -> dict:
|
||||
except Exception as e:
|
||||
stats["errors"].append(f"asset {asset.id} vsphere: {e}")
|
||||
|
||||
# IGEL's own security notices (ISN) — endpoints AND the UMS
|
||||
# server. Before scan_asset_igel so its hits land in
|
||||
# `touched_cves` for that pass's reconcile.
|
||||
try:
|
||||
from app.services import igel_isn_service
|
||||
n = igel_isn_service.scan_asset(db, asset, new_ids,
|
||||
touched=touched_cves, index=isn_index)
|
||||
if n:
|
||||
stats["findings"] += n
|
||||
touched = True
|
||||
except Exception as e:
|
||||
stats["errors"].append(f"asset {asset.id} igel-isn: {e}")
|
||||
|
||||
# IGEL OS thin clients. Same shape as vSphere — the firmware
|
||||
# version already sits on the asset, put there by the IGEL UMS
|
||||
# connector — and the same reason for existing: an endpoint that
|
||||
|
||||
@@ -953,7 +953,7 @@ def _ensure_zip(force: bool = False) -> bool:
|
||||
`force` bypasses the 12h TTL and always re-downloads. The nightly index
|
||||
build needs it: the same /tmp file is refreshed by the threat-intel job,
|
||||
whose IntervalTrigger(24h) is anchored to app startup — so on a container
|
||||
started in the late afternoon the ZIP is only ~10h old at 03:20 and the
|
||||
started in the late afternoon the ZIP is only ~8h old at 01:30 and the
|
||||
TTL kept the build on a snapshot taken hours before the day's CVEs were
|
||||
published (Chrome 151.0.7922.169 surfaced >24h late that way). On-demand
|
||||
callers keep the cache so the GUI never triggers a 557 MB download.
|
||||
|
||||
@@ -0,0 +1,338 @@
|
||||
"""
|
||||
IGEL Security Notices (ISN) → findings for IGEL OS endpoints and the UMS server.
|
||||
|
||||
The vendor source for IGEL, for the same reason TeamViewer has its bulletins
|
||||
(teamviewer_bulletin_service) and Firefox has MFSA: the CVE feeds do not see
|
||||
everything the vendor states.
|
||||
|
||||
1. NOT EVERY ISN HAS A CVE. ISN 2026-09 (UMS information disclosure, 8.6),
|
||||
ISN 2025-31 (stored XSS in UMS, 8.0) and ISN 2025-24 (command execution
|
||||
in IGEL OS, 8.8) carry no CVE id at all — a CVE-keyed source cannot see
|
||||
them however well it parses. The ISN id stands in as the identifier,
|
||||
exactly as TV-, GHSA- and NESSUS-PLUGIN- ids already do in this schema.
|
||||
2. THE FIX IS STATED PER RELEASE LINE, in the "Update Instructions" section:
|
||||
"OS 12: Update to IGEL OS 12.7.1. OS 11: Update to IGEL OS 11.10.410".
|
||||
That is exactly the per-line bound scan_asset_igel needs, and for the
|
||||
ISNs without a CVE it is the only place it is written down.
|
||||
|
||||
The page is prose, not a table. What is read:
|
||||
|
||||
* the header — "First published … CVSS:3.1: 8.6 (High) CVSS:3.1/AV:N/…" —
|
||||
for the score and severity (both optional: 66 of 198 ISNs state one of
|
||||
them loosely or not at all);
|
||||
* every CVE id anywhere on the page;
|
||||
* the "Update Instructions" section, split at its release-line anchors
|
||||
("OS 12:", "OS 11:", "UMS 12:", "UMS:") — the first version in each chunk
|
||||
is that line's fix.
|
||||
|
||||
A chunk whose version is not on the line's own major is NOT a firmware bound
|
||||
and is dropped: "OS 12: Update the Chromium app to 148.0.7778.96" is an app
|
||||
version, and the base-system version on the asset says nothing about which
|
||||
Chromium app is installed. The one exception is a UMS line told to upgrade to
|
||||
a newer major ("UMS 6: Upgrade to UMS 12.02.130"), which affects the whole old
|
||||
line. Same rule as TeamViewer's "affected, no bound": a chunk without a
|
||||
version ("IGEL is preparing a fixed OS 12 base system") yields nothing, since
|
||||
"affected, no fix yet" cannot become a per-device verdict without flagging
|
||||
every device forever once the vendor forgets to update the page.
|
||||
|
||||
Two products are in scope and both come from the asset's OS string, which
|
||||
the UMS connector writes: "IGEL OS" for endpoints (release from the version,
|
||||
never from the name — see cvelistv5_scan_service.IGEL_OS_RE) and "IGEL
|
||||
Universal Management Suite" for the server. ISNs for ICG, OS Creator, W10 and
|
||||
the Windows Embedded line have no anchor and are skipped.
|
||||
|
||||
Findings are written through the app-scan upsert with the app-scan source.
|
||||
The OS path relies on scan_asset_igel's reconcile (it runs after this one and
|
||||
folds the shared `touched` set in); the UMS server had no reconcile at all, so
|
||||
this pass closes its own stale findings.
|
||||
"""
|
||||
import json
|
||||
import logging
|
||||
import re
|
||||
from datetime import datetime, timedelta
|
||||
from typing import Dict, List, Optional
|
||||
|
||||
from sqlalchemy.orm import Session
|
||||
|
||||
from app.models.setting import Setting
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
INDEX_SETTING = "igel_isn_cache_v1"
|
||||
INDEX_TS_SETTING = "igel_isn_cache_v1_ts"
|
||||
TTL_HOURS = 24
|
||||
BASE = "https://kb.igel.com/en/security-safety/current"
|
||||
_SLUG_RE = re.compile(r'href="/en/security-safety/current/(isn-\d{4}-\d+[^"#?]*)"', re.I)
|
||||
|
||||
# "OS 12:", "IGEL OS 11:", "UMS 12:", "UMS:", "UMS 6.x:". The colon is required:
|
||||
# without it "IGEL OS 11.10.410" inside a chunk reads as a new anchor "OS 11"
|
||||
# and the fix degrades to "10.410".
|
||||
_ANCHOR_RE = re.compile(
|
||||
r"(?:IGEL\s+)?\b(OS|UMS|Universal Management Suite)\b\s*(\d+)?(?:\.x)?\s*:", re.I)
|
||||
_VER_RE = re.compile(r"(?<![\d.])(\d+\.\d+(?:\.\d+)*)")
|
||||
_CVE_RE = re.compile(r"CVE-\d{4}-\d{4,}", re.I)
|
||||
_SEV_RE = re.compile(r"\b(critical|high|medium|moderate|low)\b", re.I)
|
||||
_SEV = {"critical": "critical", "high": "high", "medium": "medium",
|
||||
"moderate": "medium", "low": "low"}
|
||||
|
||||
UMS_LABEL = "IGEL Universal Management Suite"
|
||||
|
||||
|
||||
def _plain_text(fragment: str) -> str:
|
||||
import html as _html
|
||||
return re.sub(r"\s+", " ",
|
||||
_html.unescape(re.sub(r"(?is)<[^>]+>", " ", fragment))).strip()
|
||||
|
||||
|
||||
def _sections(body: str) -> Dict[str, str]:
|
||||
"""heading → text, split at h2-h4."""
|
||||
parts = re.split(r"(?is)<h[2-4][^>]*>(.*?)</h[2-4]>", body)
|
||||
out: Dict[str, str] = {}
|
||||
for i in range(1, len(parts), 2):
|
||||
# The Confluence export appends its own <script> after the article.
|
||||
out[_plain_text(parts[i])] = re.sub(r"// Mark block images.*", "",
|
||||
_plain_text(parts[i + 1]))
|
||||
return out
|
||||
|
||||
|
||||
def _major(version: str) -> int:
|
||||
return int(version.split(".", 1)[0])
|
||||
|
||||
|
||||
def _parse_bounds(instructions: str) -> List[dict]:
|
||||
"""Update Instructions text → [{prod, release, fix, all}]."""
|
||||
bounds: List[dict] = []
|
||||
anchors = list(_ANCHOR_RE.finditer(instructions))
|
||||
if not anchors:
|
||||
# "Update to UMS 12.11.100 or newer when available." — one line, no
|
||||
# prefix. The product is the word, the release is the fix's major.
|
||||
if re.search(r"\b(UMS|IGEL OS)\b", instructions):
|
||||
m = _VER_RE.search(instructions)
|
||||
if m:
|
||||
prod = "ums" if re.search(r"\bUMS\b", instructions) else "os"
|
||||
bounds.append({"prod": prod, "release": _major(m.group(1)),
|
||||
"fix": m.group(1), "all": False})
|
||||
return bounds
|
||||
for i, a in enumerate(anchors):
|
||||
end = anchors[i + 1].start() if i + 1 < len(anchors) else len(instructions)
|
||||
chunk = instructions[a.end():end]
|
||||
m = _VER_RE.search(chunk)
|
||||
if not m:
|
||||
continue
|
||||
fix = m.group(1)
|
||||
prod = "os" if a.group(1).upper() == "OS" else "ums"
|
||||
release = int(a.group(2)) if a.group(2) else _major(fix)
|
||||
if _major(fix) == release:
|
||||
bounds.append({"prod": prod, "release": release, "fix": fix, "all": False})
|
||||
elif prod == "ums" and _major(fix) > release:
|
||||
bounds.append({"prod": prod, "release": release, "fix": fix, "all": True})
|
||||
# else: an app version on an OS line — not a firmware bound.
|
||||
return bounds
|
||||
|
||||
|
||||
def _parse_isn(slug: str, html_text: str) -> Optional[dict]:
|
||||
m = re.search(r"(?is)<(main|article)[^>]*>(.*?)</\1>", html_text)
|
||||
body = m.group(2) if m else html_text
|
||||
secs = _sections(body)
|
||||
text = _plain_text(body)
|
||||
|
||||
t = re.search(r"(?is)<h1[^>]*>(.*?)</h1>", body)
|
||||
title = _plain_text(t.group(1)) if t else slug
|
||||
# From the slug, not the heading: ISN 2023-18's h1 is written without
|
||||
# the "ISN 2023-18:" prefix every other notice carries.
|
||||
m = re.match(r"isn-(\d{4})-(\d+)", slug, re.I)
|
||||
if not m:
|
||||
return None
|
||||
isn_id = f"ISN-{m.group(1)}-{m.group(2)}"
|
||||
|
||||
# Header = after the LAST "First published" (an "Updated … (fix version
|
||||
# to 12.8.1)" line may precede it) and before "Summary".
|
||||
head = text.split("Summary", 1)[0]
|
||||
if "First published" in head:
|
||||
head = head.rsplit("First published", 1)[1]
|
||||
head = re.sub(r"\b3\.[01]\b", "", head) # the "CVSS:3.1" version token
|
||||
cvss = None
|
||||
for sm in re.finditer(r"(?<![\d.])(\d{1,2}\.\d)(?![\d.])", head):
|
||||
if float(sm.group(1)) <= 10:
|
||||
cvss = float(sm.group(1))
|
||||
break
|
||||
sm = _SEV_RE.search(head)
|
||||
sev = _SEV.get(sm.group(1).lower()) if sm else None
|
||||
|
||||
instr = next((v for k, v in secs.items()
|
||||
if re.search(r"update|mitigation|resolution", k, re.I)), "")
|
||||
affected = _parse_bounds(instr)
|
||||
if not affected:
|
||||
return None
|
||||
|
||||
cves = list(dict.fromkeys(c.upper() for c in _CVE_RE.findall(text)))
|
||||
desc = " ".join(x for x in (secs.get("Summary"), secs.get("Details")) if x)
|
||||
return {"id": isn_id, "url": f"{BASE}/{slug}", "title": title,
|
||||
"cves": cves, "cvss": cvss, "sev": sev,
|
||||
"desc": desc[:2000] or None, "affected": affected}
|
||||
|
||||
|
||||
def build_index(db: Session) -> Dict[str, dict]:
|
||||
"""Fetch the ISN overview, then every notice it links (~200 pages)."""
|
||||
import httpx
|
||||
|
||||
index: Dict[str, dict] = {}
|
||||
with httpx.Client(timeout=httpx.Timeout(30.0, connect=10.0),
|
||||
follow_redirects=True,
|
||||
headers={"User-Agent": "TrueVuln/1.0 (+security scanner)"}) as client:
|
||||
try:
|
||||
r = client.get(BASE)
|
||||
r.raise_for_status()
|
||||
except Exception as e:
|
||||
logger.warning("igel-isn: overview fetch failed: %s", e)
|
||||
return _stored(db) or {}
|
||||
slugs = sorted({s.lower() for s in _SLUG_RE.findall(r.text)})
|
||||
if not slugs:
|
||||
logger.warning("igel-isn: overview linked no notices — keeping the last index")
|
||||
return _stored(db) or {}
|
||||
failed = 0
|
||||
for slug in slugs:
|
||||
try:
|
||||
p = client.get(f"{BASE}/{slug}")
|
||||
if p.status_code != 200:
|
||||
failed += 1
|
||||
continue
|
||||
entry = _parse_isn(slug, p.text)
|
||||
except Exception as e:
|
||||
failed += 1
|
||||
logger.debug("igel-isn: %s failed: %s", slug, e)
|
||||
continue
|
||||
if entry:
|
||||
index[slug] = entry
|
||||
if failed and not index:
|
||||
return _stored(db) or {}
|
||||
no_cve = sum(1 for e in index.values() if not e["cves"])
|
||||
logger.info("igel-isn: %d notices, %d with a firmware/UMS bound "
|
||||
"(%d without a CVE id), %d unreachable",
|
||||
len(slugs), len(index), no_cve, failed)
|
||||
if not index:
|
||||
# Every page answered and none parsed — a markup change. Storing {}
|
||||
# would leave every ISN finding untouched for a scan, which the
|
||||
# reconcile reads as "no longer detected" → patched.
|
||||
logger.warning("igel-isn: %d pages parsed, no bounds — keeping the last index",
|
||||
len(slugs))
|
||||
return _stored(db) or {}
|
||||
_store(db, index)
|
||||
return index
|
||||
|
||||
|
||||
def _store(db: Session, index: Dict[str, dict]) -> None:
|
||||
for key, val in ((INDEX_SETTING, json.dumps(index)),
|
||||
(INDEX_TS_SETTING, datetime.now().isoformat())):
|
||||
row = db.query(Setting).filter(Setting.key == key).first()
|
||||
if row:
|
||||
row.value = val
|
||||
else:
|
||||
db.add(Setting(key=key, value=val))
|
||||
db.commit()
|
||||
|
||||
|
||||
def _stored(db: Session) -> Optional[Dict[str, dict]]:
|
||||
row = db.query(Setting).filter(Setting.key == INDEX_SETTING).first()
|
||||
if not row or not row.value:
|
||||
return None
|
||||
try:
|
||||
return json.loads(row.value)
|
||||
except json.JSONDecodeError:
|
||||
return None
|
||||
|
||||
|
||||
def load_index(db: Session) -> Optional[Dict[str, dict]]:
|
||||
ts = db.query(Setting).filter(Setting.key == INDEX_TS_SETTING).first()
|
||||
if not ts or not ts.value:
|
||||
return None
|
||||
try:
|
||||
if datetime.now() - datetime.fromisoformat(ts.value) > timedelta(hours=TTL_HOURS):
|
||||
return None
|
||||
except ValueError:
|
||||
return None
|
||||
return _stored(db)
|
||||
|
||||
|
||||
def get_index(db: Session) -> Dict[str, dict]:
|
||||
idx = load_index(db)
|
||||
if idx is not None:
|
||||
return idx
|
||||
try:
|
||||
return build_index(db)
|
||||
except Exception as e:
|
||||
logger.warning("igel-isn index build failed: %s", e)
|
||||
return _stored(db) or {}
|
||||
|
||||
|
||||
def _product(asset) -> Optional[str]:
|
||||
from app.services import cvelistv5_scan_service as c5
|
||||
name = (asset.operating_system or "").strip()
|
||||
if c5.IGEL_OS_RE.match(name):
|
||||
return "os"
|
||||
if name.lower().startswith(UMS_LABEL.lower()):
|
||||
return "ums"
|
||||
return None
|
||||
|
||||
|
||||
def _affected(installed: str, bound: dict) -> bool:
|
||||
from app.services import app_cve_scanner_service as cpe
|
||||
if bound["all"]:
|
||||
return True
|
||||
return (cpe._vcmp(installed, bound["fix"]) or 0) < 0
|
||||
|
||||
|
||||
def scan_asset(db: Session, asset, new_ids: Optional[list] = None,
|
||||
touched: Optional[set] = None, index: Optional[dict] = None) -> int:
|
||||
"""Match an IGEL OS endpoint or the UMS server against the ISNs.
|
||||
Returns findings upserted. Caller commits."""
|
||||
from app.services import app_cve_scanner_service as cpe
|
||||
from app.services import cvelistv5_scan_service as c5
|
||||
|
||||
prod = _product(asset)
|
||||
if not prod:
|
||||
return 0
|
||||
if index is None:
|
||||
index = get_index(db)
|
||||
if not index:
|
||||
return 0
|
||||
cver = cpe._clean_version(asset.os_version or "")
|
||||
if not cver:
|
||||
return 0
|
||||
release = c5.igel_release(cver)
|
||||
if release is None:
|
||||
return 0
|
||||
if new_ids is None:
|
||||
new_ids = []
|
||||
label = c5.IGEL_LABEL if prod == "os" else UMS_LABEL
|
||||
|
||||
count = 0
|
||||
still_affected: set = set()
|
||||
for entry in index.values():
|
||||
bound = next((b for b in entry["affected"]
|
||||
if b["prod"] == prod and b["release"] == release
|
||||
and _affected(cver, b)), None)
|
||||
if not bound:
|
||||
continue
|
||||
for ident in (entry["cves"] or [entry["id"]]):
|
||||
still_affected.add(ident.upper())
|
||||
c = {"cve": ident, "cvss": entry.get("cvss"),
|
||||
"severity": entry.get("sev"), "fixed": bound["fix"],
|
||||
"desc": entry.get("desc"),
|
||||
"refs": json.dumps([entry["url"]])}
|
||||
try:
|
||||
before = len(new_ids)
|
||||
cpe._upsert(db, asset, label, asset.os_version or cver, c,
|
||||
new_ids, touched=touched, vendor="IGEL")
|
||||
count += 1 if len(new_ids) > before else 0
|
||||
except Exception as e:
|
||||
logger.debug("igel-isn upsert failed (%s on %s): %s", ident, asset.id, e)
|
||||
|
||||
if prod == "ums":
|
||||
# The endpoint path reconciles in scan_asset_igel (which runs after
|
||||
# this and folds `touched` in). The UMS server has no other reconcile:
|
||||
# the CPE OS scan before this one never closed anything.
|
||||
c5._resolve_stale_appliance(db, asset, label,
|
||||
still_affected | set(touched or ()), cver,
|
||||
tag="igel-isn")
|
||||
return count
|
||||
@@ -499,7 +499,9 @@ def _run_cve_scan(db: Session, asset_ids: set) -> int:
|
||||
return 0
|
||||
from app.services import app_cve_scanner_service as cpe
|
||||
from app.services import cvelistv5_scan_service as c5
|
||||
from app.services import igel_isn_service as isn
|
||||
index = c5.load_index(db) or {}
|
||||
isn_index = isn.get_index(db)
|
||||
if not index:
|
||||
logger.info("IGEL sync: no cvelistV5 index yet — CVE pass deferred "
|
||||
"to the nightly app-CVE scan")
|
||||
@@ -511,6 +513,14 @@ def _run_cve_scan(db: Session, asset_ids: set) -> int:
|
||||
total += cpe.scan_asset_os(db, asset, new_ids, touched=touched)
|
||||
except Exception as e:
|
||||
logger.warning("IGEL CPE scan failed for %s: %s", asset.hostname, e)
|
||||
# The vendor's own notices — the only source for the ISNs without a
|
||||
# CVE id. Between the CPE pass and the cvelistV5 one on purpose: it
|
||||
# folds the CPE hits into the UMS reconcile, and scan_asset_igel
|
||||
# folds its hits into the endpoint one.
|
||||
try:
|
||||
total += isn.scan_asset(db, asset, new_ids, touched=touched, index=isn_index)
|
||||
except Exception as e:
|
||||
logger.warning("IGEL ISN scan failed for %s: %s", asset.hostname, e)
|
||||
if not index:
|
||||
continue
|
||||
try:
|
||||
|
||||
@@ -0,0 +1,270 @@
|
||||
"""IGEL Security Notices (ISN) — run: python tests/test_igel_isn.py
|
||||
|
||||
Field report (03.09.2026): many IGEL flaws never get a CVE id, only the
|
||||
vendor's ISN id — ISN 2026-09 (UMS information disclosure, CVSS 8.6),
|
||||
ISN 2025-31 (stored XSS in UMS, 8.0), ISN 2025-24 (command execution in IGEL
|
||||
OS, 8.8). No CVE-keyed source can see them; only the vendor page states the
|
||||
CVSS at the top and the fixed versions under "Update Instructions".
|
||||
|
||||
The HTML below is the article text of the live pages, verbatim, in the
|
||||
Confluence export's heading structure (main > h1, h2 Summary / Details /
|
||||
Update Instructions / References). Exercises the production parser and the
|
||||
production scan decision, not copies of their logic.
|
||||
"""
|
||||
import json
|
||||
import os
|
||||
import sys
|
||||
|
||||
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
|
||||
|
||||
from app.services import igel_isn_service as isn
|
||||
from app.services import app_cve_scanner_service as cpe
|
||||
from app.services import cvelistv5_scan_service as c5
|
||||
|
||||
|
||||
def page(title, head, summary, details, update, refs=""):
|
||||
return (f"<html><body><nav>Breadcrumbs Home Security & Safety</nav><main>"
|
||||
f"<h1>{title}</h1><p>{head}</p>"
|
||||
f"<h2>Summary</h2><p>{summary}</p>"
|
||||
f"<h2>Details</h2><p>{details}</p>"
|
||||
f"<h2>Update Instructions</h2><p>{update}</p>"
|
||||
+ (f"<h2>References</h2><p>{refs}</p>" if refs else "")
|
||||
+ "</main><script>// Mark block images on the first level as breakout"
|
||||
"</script></body></html>")
|
||||
|
||||
|
||||
# ---- verbatim from kb.igel.com (fetched 03.09.2026) ------------------------
|
||||
UMS_2026_09 = page(
|
||||
"ISN 2026-09: UMS Information Disclosure",
|
||||
"First published 1 April 2026 CVSS:3.1: 8.6 (High) CVSS:3.1/AV:N/AC:L/PR:N/UI:N/S:U/C:H/I:L/A:L",
|
||||
"A security vulnerability has been found in the UMS. This affects the following "
|
||||
"product versions: IGEL UMS in version 12.10.100 and older",
|
||||
"It has been discovered that an HTTPS endpoint in UMS disclosed multiple files "
|
||||
"without authentication.",
|
||||
"Update to UMS 12.11.100 or newer when available.")
|
||||
|
||||
UMS_2025_31 = page(
|
||||
"ISN 2025-31: XSS Vulnerabilities in UMS",
|
||||
"First published 28 July 2025 CVSS:3.1: 8.0 (High) CVSS:3.1/AV:N/AC:L/PR:L/UI:R/S:U/C:H/I:H/A:H",
|
||||
"Multiple instances of Stored Cross-Site Scripting (XSS) vulnerabilities found that "
|
||||
"affect the following products: IGEL Universal Management Suite versions <=12.08.110",
|
||||
"After internal and external security testing, multiple instances of stored "
|
||||
"Cross-Site Scripting (XSS) vulnerabilities have been found in IGEL UMS.",
|
||||
"UMS: Update to version 12.08.130")
|
||||
|
||||
UMS_2023_27 = page(
|
||||
"ISN 2023-27: ActiveMQ in UMS HA",
|
||||
"First published 3 November 2023 CVSS 3.1: 10.0 (Critical) CVSS:3.1/AV:N/AC:L/PR:N/UI:N/S:C/C:L/I:H/A:H",
|
||||
"Apache ActiveMQ is vulnerable to a critical remote code execution vulnerability. "
|
||||
"This vulnerability affects the High Availability (HA) feature only , used in UMS "
|
||||
"in the following versions: UMS versions <= 12.02.120",
|
||||
"Apache ActiveMQ is vulnerable to a critical (10.0) remote code execution "
|
||||
"vulnerability being tracked with CVE-2023-46604.",
|
||||
"UMS 12: We are preparing an emergency release of UMS 12.02.130. "
|
||||
"UMS 6: Upgrade to UMS 12.02.130, available soon.",
|
||||
"CVE-2023-46604: https://nvd.nist.gov/vuln/detail/CVE-2023-46604")
|
||||
|
||||
OS_2026_18 = page(
|
||||
"ISN 2026-18: “Copy fail” Kernel Vulnerability",
|
||||
"First published 8 May 2026 CVSS:3.1: 7.8 (High) CVSS:3.1/AV:L/AC:L/PR:L/UI:N/S:U/C:H/I:H/A:H",
|
||||
"An important security vulnerability has been found in the Linux Kernel used in "
|
||||
"IGEL OS. This affects the following product versions: IGEL OS 12 IGEL OS 11",
|
||||
"A local privilege escalation has been discovered in the Linux Kernel. This is "
|
||||
"tracked as CVE-2026-31431 and rated as high.",
|
||||
"OS 12: Upgrade the Base System app to version 12.8.2 LTS or 12.9.0 as soon as "
|
||||
"they are available. OS 11: Upgrade to IGEL OS 11.11.150 as soon as it is available.",
|
||||
"CVE-2026-31431 https://www.cve.org/CVERecord?id=CVE-2026-31431")
|
||||
|
||||
OS_2026_08 = page(
|
||||
"ISN 2026-08: Telnetd Vulnerabilities",
|
||||
"First published 26 March 2026 CVSS:3.1: 7.8 (High) CVSS:3.1/AV:L/AC:L/PR:L/UI:N/S:U/C:H/I:H/A:H",
|
||||
"A security vulnerability has been found in Telnetd, a network service used in IGEL "
|
||||
"OS for the Secure Terminal feature. This affects the following product versions: IGEL OS 12",
|
||||
"Telnetd in GNU Inetutils through 2.7 allows a remote authentication bypass "
|
||||
"(CVE-2026-24061). In addition, there is an out-of-bounds write (CVE-2026-32746). "
|
||||
"OS 11 is not affected, as it uses a different Telnetd implementation.",
|
||||
"OS 12: Update to the base system app in version 12.7.6 or newer when available "
|
||||
"from the IGEL App Portal.",
|
||||
"CVE-2026-24061: https://nvd.nist.gov/vuln/detail/CVE-2026-24061 "
|
||||
"CVE-2026-32746: https://nvd.nist.gov/vuln/detail/CVE-2026-32746")
|
||||
|
||||
OS_2025_24 = page(
|
||||
"ISN 2025-24: Command Execution in IGEL OS",
|
||||
"First published 30 June 2025 CVSS:3.1: 8.8 (High) CVSS:3.1/AV:L/AC:L/PR:L/UI:N/S:C/C:H/I:H/A:H",
|
||||
"A security vulnerability has been found in the IGEL OS base system. This affects "
|
||||
"the following product versions: IGEL OS 12 IGEL OS 11",
|
||||
"An issue has been found in the way IGEL OS handles the LD_PRELOAD environment variable.",
|
||||
"OS 12: Update to IGEL OS 12.7.1. OS 11: Update to IGEL OS 11.10.410 when available "
|
||||
"(planned for July).")
|
||||
|
||||
# An "Updated" line BEFORE "First published" carries the fix version — it must
|
||||
# not be read as the score.
|
||||
OS_2026_07 = page(
|
||||
"ISN 2026-07: AppArmor Vulnerabilities",
|
||||
"Updated 30 March 2026 (updated OS 12 fix version to 12.8.1) First published 26 March 2026 "
|
||||
"CVSS:3.1: 7.8 (High) CVSS:3.1/AV:L/AC:L/PR:L/UI:N/S:U/C:H/I:H/A:H",
|
||||
"IGEL OS 12 IGEL OS 11",
|
||||
"Two AppArmor flaws, CVE-2026-1234 and CVE-2026-1235.",
|
||||
"OS 12: Update to IGEL OS 12.8.1. OS 11: Update to IGEL OS 11.11.150.")
|
||||
|
||||
# Chromium ISN: the OS 12 fix is an APP version — no firmware bound. OS 11 is.
|
||||
OS_CHROMIUM = page(
|
||||
"ISN 2026-34: Critical Chromium Vulnerabilities",
|
||||
"First published 1 September 2026 CVSS:3.1: 9.6 (Critical)",
|
||||
"IGEL OS 12 IGEL OS 11",
|
||||
"CVE-2026-9000.",
|
||||
"OS 12: Update the Chromium app to version 151.0.7922.137 or newer. "
|
||||
"OS 11: Update to IGEL OS 11.11.160.")
|
||||
|
||||
# No fix version at all → nothing (cannot become a per-device verdict).
|
||||
OS_PREPARING = page(
|
||||
"ISN 2024-05: Something",
|
||||
"First published 1 February 2024 CVSS:3.1: 8.4 (High)",
|
||||
"IGEL OS 12", "CVE-2024-1.", "OS 12: IGEL is preparing a fixed OS 12 base system version.")
|
||||
|
||||
|
||||
def parse(slug, html):
|
||||
e = isn._parse_isn(slug, html)
|
||||
assert e, slug
|
||||
return e
|
||||
|
||||
|
||||
def bounds(e):
|
||||
return [(b["prod"], b["release"], b["fix"], b["all"]) for b in e["affected"]]
|
||||
|
||||
|
||||
def test_parser():
|
||||
e = parse("isn-2026-09-ums-information-disclosure", UMS_2026_09)
|
||||
assert (e["id"], e["cvss"], e["sev"], e["cves"]) == ("ISN-2026-09", 8.6, "high", [])
|
||||
assert bounds(e) == [("ums", 12, "12.11.100", False)]
|
||||
assert e["url"].endswith("/isn-2026-09-ums-information-disclosure")
|
||||
assert e["desc"].startswith("A security vulnerability has been found in the UMS.")
|
||||
|
||||
e = parse("isn-2025-31-xss-vulnerabilities-in-ums", UMS_2025_31)
|
||||
assert (e["cvss"], e["sev"], e["cves"]) == (8.0, "high", [])
|
||||
assert bounds(e) == [("ums", 12, "12.08.130", False)]
|
||||
|
||||
e = parse("isn-2023-27-activemq-in-ums-ha", UMS_2023_27)
|
||||
assert (e["cvss"], e["sev"], e["cves"]) == (10.0, "critical", ["CVE-2023-46604"])
|
||||
assert bounds(e) == [("ums", 12, "12.02.130", False), ("ums", 6, "12.02.130", True)]
|
||||
|
||||
e = parse("isn-2026-18-copy-fail-kernel-vulnerability", OS_2026_18)
|
||||
assert (e["cvss"], e["sev"], e["cves"]) == (7.8, "high", ["CVE-2026-31431"])
|
||||
assert bounds(e) == [("os", 12, "12.8.2", False), ("os", 11, "11.11.150", False)]
|
||||
|
||||
e = parse("isn-2026-08-telnetd-vulnerabilities", OS_2026_08)
|
||||
assert e["cves"] == ["CVE-2026-24061", "CVE-2026-32746"]
|
||||
assert bounds(e) == [("os", 12, "12.7.6", False)] # OS 11 not affected
|
||||
|
||||
e = parse("isn-2025-24-command-execution-in-igel-os", OS_2025_24)
|
||||
assert (e["cvss"], e["cves"]) == (8.8, [])
|
||||
assert bounds(e) == [("os", 12, "12.7.1", False), ("os", 11, "11.10.410", False)]
|
||||
|
||||
e = parse("isn-2026-07-apparmor-vulnerabilities", OS_2026_07)
|
||||
assert e["cvss"] == 7.8, e["cvss"]
|
||||
assert bounds(e) == [("os", 12, "12.8.1", False), ("os", 11, "11.11.150", False)]
|
||||
|
||||
e = parse("isn-2026-34-critical-chromium-vulnerabilities", OS_CHROMIUM)
|
||||
assert bounds(e) == [("os", 11, "11.11.160", False)]
|
||||
|
||||
assert isn._parse_isn("isn-2024-05-something", OS_PREPARING) is None
|
||||
print(" parser OK")
|
||||
|
||||
|
||||
class _Asset:
|
||||
def __init__(self, os_name, version):
|
||||
self.id = 1
|
||||
self.hostname = "tc-01"
|
||||
self.operating_system = os_name
|
||||
self.os_version = version
|
||||
|
||||
|
||||
def run_scan(asset, index):
|
||||
"""scan_asset with the two DB writers stubbed — records what it would write."""
|
||||
written, closed = [], []
|
||||
orig_upsert, orig_resolve = cpe._upsert, c5._resolve_stale_appliance
|
||||
|
||||
def fake_upsert(db, a, pkg, version, c, new_ids, touched=None, vendor=None):
|
||||
written.append((c["cve"], pkg, c["fixed"], c["cvss"], c["severity"], json.loads(c["refs"])[0]))
|
||||
if touched is not None:
|
||||
touched.add(c["cve"].upper())
|
||||
new_ids.append(len(new_ids) + 1)
|
||||
|
||||
def fake_resolve(db, a, label, still, evidence, tag):
|
||||
closed.append((label, sorted(still), tag))
|
||||
return 0
|
||||
|
||||
cpe._upsert, c5._resolve_stale_appliance = fake_upsert, fake_resolve
|
||||
try:
|
||||
touched = set()
|
||||
n = isn.scan_asset(None, asset, [], touched=touched, index=index)
|
||||
finally:
|
||||
cpe._upsert, c5._resolve_stale_appliance = orig_upsert, orig_resolve
|
||||
return n, written, closed, touched
|
||||
|
||||
|
||||
def test_scan():
|
||||
index = {}
|
||||
for slug, html in (("isn-2026-09-ums-information-disclosure", UMS_2026_09),
|
||||
("isn-2025-31-xss-vulnerabilities-in-ums", UMS_2025_31),
|
||||
("isn-2023-27-activemq-in-ums-ha", UMS_2023_27),
|
||||
("isn-2026-18-copy-fail-kernel-vulnerability", OS_2026_18),
|
||||
("isn-2026-08-telnetd-vulnerabilities", OS_2026_08),
|
||||
("isn-2025-24-command-execution-in-igel-os", OS_2025_24),
|
||||
("isn-2026-34-critical-chromium-vulnerabilities", OS_CHROMIUM)):
|
||||
index[slug] = parse(slug, html)
|
||||
|
||||
# UMS 12.08.110 — the field report's server: XSS (no CVE) + info disclosure
|
||||
# (no CVE); ActiveMQ fixed in 12.02.130, so not that one.
|
||||
n, written, closed, _ = run_scan(_Asset("IGEL Universal Management Suite", "12.08.110"), index)
|
||||
ids = sorted(w[0] for w in written)
|
||||
assert ids == ["ISN-2025-31", "ISN-2026-09"], ids
|
||||
assert all(w[1] == "IGEL Universal Management Suite" for w in written)
|
||||
fixes = {w[0]: w[2] for w in written}
|
||||
assert fixes == {"ISN-2025-31": "12.08.130", "ISN-2026-09": "12.11.100"}, fixes
|
||||
assert {w[0]: w[3] for w in written} == {"ISN-2025-31": 8.0, "ISN-2026-09": 8.6}
|
||||
assert written[0][5].startswith("https://kb.igel.com/")
|
||||
# The UMS server reconciles here (nothing else ever closed its findings).
|
||||
assert closed == [("IGEL Universal Management Suite", ["ISN-2025-31", "ISN-2026-09"], "igel-isn")], closed
|
||||
|
||||
# UMS 12.11.100 — fixed everything.
|
||||
n, written, closed, _ = run_scan(_Asset("IGEL Universal Management Suite", "12.11.100"), index)
|
||||
assert written == [], written
|
||||
assert closed[0][1] == []
|
||||
|
||||
# UMS 6.10.130 — whole line affected by the ActiveMQ ISN (upgrade to 12).
|
||||
n, written, _, _ = run_scan(_Asset("IGEL Universal Management Suite", "6.10.130"), index)
|
||||
assert [(w[0], w[2]) for w in written] == [("CVE-2023-46604", "12.02.130")], written
|
||||
|
||||
# OS 12.7.0+3 (build suffix as UMS reports it): copy-fail, telnetd ×2,
|
||||
# LD_PRELOAD (no CVE, fixed 12.7.1). Chromium ISN: OS 12 line has no
|
||||
# firmware bound.
|
||||
n, written, closed, touched = run_scan(_Asset("IGEL OS", "12.7.0+3"), index)
|
||||
ids = sorted(w[0] for w in written)
|
||||
assert ids == ["CVE-2026-24061", "CVE-2026-31431", "CVE-2026-32746", "ISN-2025-24"], ids
|
||||
assert all(w[1] == "IGEL OS" for w in written)
|
||||
assert {w[0]: w[2] for w in written}["CVE-2026-31431"] == "12.8.2"
|
||||
assert touched == set(ids) # scan_asset_igel's reconcile folds these in
|
||||
assert closed == [], closed # …so the endpoint does NOT reconcile here
|
||||
|
||||
# OS 12.8.2 — copy-fail fixed, the rest fixed earlier.
|
||||
n, written, _, _ = run_scan(_Asset("IGEL OS", "12.8.2"), index)
|
||||
assert written == [], written
|
||||
|
||||
# OS 11.11.140: copy-fail (11.11.150) yes; telnetd (OS 11 not affected)
|
||||
# no; LD_PRELOAD (11.10.410) no; Chromium (11.11.160) yes. Never the
|
||||
# OS 12 bounds — 11 is a different product.
|
||||
n, written, _, _ = run_scan(_Asset("IGEL OS", "11.11.140"), index)
|
||||
got = sorted((w[0], w[2]) for w in written)
|
||||
assert got == [("CVE-2026-31431", "11.11.150"), ("CVE-2026-9000", "11.11.160")], got
|
||||
|
||||
# Not an IGEL asset / no version → nothing, no reconcile.
|
||||
assert run_scan(_Asset("Windows 11", "10.0.26100"), index)[1] == []
|
||||
assert run_scan(_Asset("IGEL OS", ""), index)[2] == []
|
||||
print(" scan OK")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
test_parser()
|
||||
test_scan()
|
||||
print("test_igel_isn OK")
|
||||
@@ -144,9 +144,56 @@ def demo():
|
||||
"[]", m365_service._load_cache), \
|
||||
"M365 cache is expected to survive the 24h boundary"
|
||||
|
||||
# --- app CVE scan: both vendor indexes rebuilt before the scan reads them
|
||||
# --- one job rebuilds every index, and it is scheduled BEFORE the first
|
||||
# sync that decides from them. The Intune sync (02:10) runs the app-CVE
|
||||
# scan per device and can take hours; it used to read indexes that the
|
||||
# 03:20 job rebuilt after it, i.e. yesterday's, every night.
|
||||
from app.services import igel_isn_service as isn
|
||||
events = []
|
||||
db = _DB()
|
||||
restore = [
|
||||
_patch(sched, SessionLocal=lambda: db),
|
||||
_patch(cvelistv5_scan_service,
|
||||
build_product_index=lambda d, force_fresh=False:
|
||||
events.append(f"c5(force={force_fresh})") or {}),
|
||||
_patch(gh, build_index=lambda d: events.append("gh.build") or {}),
|
||||
_patch(tv, build_index=lambda d: events.append("tv.build") or {}),
|
||||
_patch(isn, build_index=lambda d: events.append("isn.build") or {}),
|
||||
]
|
||||
try:
|
||||
sched.vuln_index_refresh_nightly()
|
||||
finally:
|
||||
[r() for r in restore]
|
||||
assert events == ["c5(force=True)", "gh.build", "tv.build", "isn.build"], events
|
||||
|
||||
jobs = {}
|
||||
class _Sched:
|
||||
def add_job(self, fn, trigger=None, id=None, **k):
|
||||
jobs[id] = (trigger.fields[5].__str__(), trigger.fields[6].__str__()) \
|
||||
if hasattr(trigger, "fields") else None
|
||||
def get_jobs(self):
|
||||
return []
|
||||
def start(self):
|
||||
pass
|
||||
restore = [_patch(sched, scheduler=_Sched(), HAS_APSCHEDULER=True,
|
||||
sync_schedules=lambda: None)]
|
||||
try:
|
||||
sched.start_scheduler()
|
||||
finally:
|
||||
[r() for r in restore]
|
||||
hm = lambda j: tuple(int(x) for x in jobs[j])
|
||||
for sync in ("intune_sync_nightly", "vcenter_sync_nightly",
|
||||
"igel_sync_nightly", "app_cve_scan_nightly"):
|
||||
assert hm("vuln_index_refresh_nightly") < hm(sync), (
|
||||
f"the index refresh must be scheduled before {sync}: "
|
||||
f"{jobs['vuln_index_refresh_nightly']} vs {jobs[sync]}")
|
||||
assert hm("intune_sync_nightly")[0] * 60 + hm("intune_sync_nightly")[1] \
|
||||
- hm("vuln_index_refresh_nightly")[0] * 60 - hm("vuln_index_refresh_nightly")[1] >= 30, \
|
||||
"the Intune sync needs room for the cvelistV5 ZIP walk to finish"
|
||||
|
||||
# --- the app-CVE scan itself no longer rebuilds (it would be the second
|
||||
# 557 MB walk of the night) — it reads what 01:30 stored
|
||||
events.clear()
|
||||
restore = [
|
||||
_patch(sched, SessionLocal=lambda: db),
|
||||
_patch(cvelistv5_scan_service,
|
||||
@@ -160,16 +207,9 @@ def demo():
|
||||
sched.app_cve_scan_nightly()
|
||||
finally:
|
||||
[r() for r in restore]
|
||||
assert events == ["scan"], events
|
||||
|
||||
for tag in ("gh.build", "tv.build", "scan"):
|
||||
assert tag in events, f"{tag} never ran: {events}"
|
||||
assert events.index("gh.build") < events.index("scan"), (
|
||||
"the repo-advisory index (Notepad++, Wazuh) must be rebuilt BEFORE the "
|
||||
f"scan that matches against it, got {events}")
|
||||
assert events.index("tv.build") < events.index("scan"), (
|
||||
f"the TeamViewer index must be rebuilt BEFORE the scan, got {events}")
|
||||
|
||||
# --- a vendor site that is down must not take the scan with it
|
||||
# --- a vendor site that is down must not take the other rebuilds with it
|
||||
events.clear()
|
||||
def _boom(d):
|
||||
events.append("gh.build")
|
||||
@@ -181,14 +221,13 @@ def demo():
|
||||
build_product_index=lambda d, force_fresh=False: events.append("c5") or {}),
|
||||
_patch(gh, build_index=_boom),
|
||||
_patch(tv, build_index=lambda d: events.append("tv.build") or {}),
|
||||
_patch(app_cve_scanner_service,
|
||||
run_app_cve_scan=lambda d, **k: events.append("scan") or {"findings": 0}),
|
||||
_patch(isn, build_index=lambda d: events.append("isn.build") or {}),
|
||||
]
|
||||
try:
|
||||
sched.app_cve_scan_nightly()
|
||||
sched.vuln_index_refresh_nightly()
|
||||
finally:
|
||||
[r() for r in restore]
|
||||
assert events == ["c5", "gh.build", "tv.build", "scan"], (
|
||||
assert events == ["c5", "gh.build", "tv.build", "isn.build"], (
|
||||
f"a failed index rebuild must fall back to the cache, got {events}")
|
||||
|
||||
# --- M365: the page is re-parsed before the build comparison, not after
|
||||
@@ -257,8 +296,9 @@ def demo():
|
||||
"the Intune sync must pull the M365 page itself — the 03:10 job's parse "
|
||||
f"is a night old by the time this one runs, got {events}")
|
||||
|
||||
print("OK — nightly jobs rebuild the Notepad++/Wazuh, TeamViewer and M365 "
|
||||
"indexes before the scans that decide from them")
|
||||
print("OK — the 01:30 job rebuilds cvelistV5, Notepad++/Wazuh, TeamViewer and "
|
||||
"IGEL ISN before the syncs that decide from them; M365 is re-parsed "
|
||||
"by each job that compares builds")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
|
||||
Reference in New Issue
Block a user