vulncheck/app/services/igel_isn_service.py

"""
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