vulncheck/app/services/android_cve_service.py

"""
Android per-CVE detection from the Google Android Security Bulletin (ASB).

For an Intune-managed Android device we know its security patch level
(androidSecurityPatchLevel, e.g. "2025-03-01"). Every monthly ASB published
AFTER that level lists CVEs the device has NOT yet received. We fetch those
months from source.android.com (stable, static, per-month URLs), extract the
CVEs + severity, and raise real-CVE findings (source 'android-asb').

Why ASB and not Samsung's SMR page: Samsung's securityUpdate.smsb ignores the
year/month query param and loads the month via JS, so a plain fetch can't get
a historical month. ASB is the upstream source for the Google CVEs Samsung
ships (the security-critical bulk) and is cleanly scrapeable. Samsung-
proprietary SVE CVEs are not covered (their page is unscrapeable).

Scope guards (volume): Critical + High only, last _MAX_MONTHS months.
"""
from __future__ import annotations

import json
import logging
import re
from datetime import date, datetime
from typing import List, Optional, Tuple

from sqlalchemy.orm import Session

logger = logging.getLogger(__name__)

_ASB_BASE = "https://source.android.com/docs/security/bulletin"


def _asb_urls(month: str) -> List[str]:
    """Candidate ASB URLs for a month slug. From 2026 Google nests the page
    under a year segment (/bulletin/2026/2026-01-01); older months are flat
    (/bulletin/2025-10-01). Try the year-nested form first, then flat, so we
    survive whichever format applies (and future shifts)."""
    year = month[:4]
    return [f"{_ASB_BASE}/{year}/{month}", f"{_ASB_BASE}/{month}"]
_CACHE_PREFIX = "asb_month_v2_"        # v2: bumped when the URL/section-filter
                                        # logic changed, so stale cache entries
                                        # from before those fixes are ignored
                                        # and every month is refetched fresh.
_MAX_MONTHS = 12                      # cap lookback so a very stale device can't flood
_WANT_SEV = {"critical", "high"}      # actionable severities only
_CVE_RE = re.compile(r"CVE-\d{4}-\d{4,7}")
_SEV_RE = re.compile(r">(Critical|High|Moderate|Low)<")
# Section headers + rows, in document order, so each CVE is scoped to its
# ASB section.
_TOKEN_RE = re.compile(r"<h[23][^>]*>(.*?)</h[23]>|<tr[^>]*>(.*?)</tr>", re.S)
# SoC / third-party vendor sections — those CVEs only affect devices with that
# chipset (Qualcomm/MediaTek/etc.), so importing them onto every Android device
# produces false positives (Samsung's own SMR lists many as "Not applicable").
# We keep only the AOSP sections (Framework/System/Kernel/Runtime/Media/Play/
# Widevine) that apply to any Android device at that patch level.
_SOC_SECTION = re.compile(
    r"qualcomm|mediatek|unisoc|spreadtrum|imagination|arm component|"
    r"broadcom|nvidia|marvell|kryo|adreno", re.I)


def _parse_patch_month(raw) -> Optional[Tuple[int, int]]:
    """androidSecurityPatchLevel 'YYYY-MM-DD' → (year, month)."""
    m = re.match(r"(\d{4})-(\d{2})", str(raw or ""))
    if not m:
        return None
    return int(m.group(1)), int(m.group(2))


def _months_after(year: int, month: int, today: date) -> List[str]:
    """ASB month slugs ('YYYY-MM-01') strictly after (year,month) up to today,
    newest first, capped at _MAX_MONTHS."""
    out = []
    y, mo = year, month
    while True:
        mo += 1
        if mo > 12:
            mo = 1
            y += 1
        if (y, mo) > (today.year, today.month):
            break
        out.append(f"{y:04d}-{mo:02d}-01")
    return out[-_MAX_MONTHS:][::-1]


def _parse_asb_html(html: str) -> List[Tuple[str, str]]:
    """→ [(cve, severity)] from one ASB page, walking section headers + rows in
    order. CVEs under SoC/vendor sections (chipset-specific) are skipped so we
    don't false-positive them onto devices with a different SoC. Severity
    carries forward across rowspan rows (Android merges the severity cell)."""
    out: List[Tuple[str, str]] = []
    seen: set = set()
    last_sev = "High"
    skip = False
    for m in _TOKEN_RE.finditer(html):
        if m.group(1) is not None:  # section header
            last_sev = "High"
            skip = bool(_SOC_SECTION.search(re.sub(r"<[^>]+>", "", m.group(1))))
            continue
        cell = m.group(2)
        cm = _CVE_RE.search(cell)
        if not cm:
            continue
        sm = _SEV_RE.search(cell)
        if sm:
            last_sev = sm.group(1)
        if skip:
            continue  # SoC/vendor section → chipset-specific, not universal
        cve = cm.group(0).upper()
        if cve in seen:
            continue
        seen.add(cve)
        out.append((cve, last_sev.lower()))
    return out


# A month with CVEs is immutable → cache forever. An empty/404 month (a
# future month Google hasn't published yet) is negatively cached for this long
# so we don't re-fetch it on every device sync, but still pick it up once it
# goes live.
_NEG_TTL_DAYS = 3


def _cache_read(db: Session, key: str):
    """→ (pairs, is_fresh). pairs may be []; is_fresh False means refetch."""
    from app.models.setting import Setting
    row = db.query(Setting).filter(Setting.key == key).first()
    if not row or not row.value:
        return None, False
    try:
        blob = json.loads(row.value)
    except Exception:
        return None, False
    if isinstance(blob, list):  # legacy positive entry
        return [tuple(x) for x in blob], True
    pairs = [tuple(x) for x in (blob.get("pairs") or [])]
    if pairs:
        return pairs, True  # non-empty is immutable
    try:
        ts = datetime.fromisoformat(blob.get("ts"))
    except Exception:
        return [], False
    return [], (datetime.now() - ts).days < _NEG_TTL_DAYS


def _cache_write(db: Session, key: str, month: str, pairs: list) -> None:
    from app.models.setting import Setting
    payload = json.dumps({"ts": datetime.now().isoformat(), "pairs": pairs})
    row = db.query(Setting).filter(Setting.key == key).first()
    if row:
        row.value = payload
    else:
        db.add(Setting(key=key, value=payload,
                       description=f"Android Security Bulletin {month} (cve,severity)"))
    db.commit()


def fetch_asb_month(db: Session, month: str) -> List[Tuple[str, str]]:
    """Cached fetch+parse of one ASB month. Empty/404 months are negatively
    cached (short TTL) so a future month Google hasn't published yet doesn't
    trigger a re-fetch on every device sync."""
    key = _CACHE_PREFIX + month
    pairs, fresh = _cache_read(db, key)
    if fresh:
        return pairs
    import httpx
    pairs = []
    try:
        with httpx.Client(timeout=30.0, follow_redirects=True,
                          headers={"User-Agent": "TrueVuln/1.0"}) as c:
            for url in _asb_urls(month):
                r = c.get(url)
                if r.status_code == 200:
                    pairs = _parse_asb_html(r.text)
                    break
    except Exception as e:
        logger.debug("ASB fetch failed for %s: %s", month, e)
        pairs = []
    _cache_write(db, key, month, pairs)  # cache empties too (negative cache)
    return pairs


def _upsert(db: Session, asset, month: str, cve: str, sev: str, new_ids: list,
            source: str = "android-asb", label: str = "ASB") -> None:
    from app.models.vulnerability import Vulnerability, VulnerabilitySeverity, VulnerabilityStatus
    cve_id = cve.upper()
    existing = (db.query(Vulnerability)
                  .filter(Vulnerability.cve_id == cve_id, Vulnerability.asset_id == asset.id)
                  .first())
    if existing:
        existing.add_source(source)
        if not existing.package_name:
            existing.package_name = f"Android ({label} {month[:7]})"[:255]
        if existing.status == VulnerabilityStatus.patched:
            existing.status = VulnerabilityStatus.open
            existing.patched_at = None
        try:
            existing.refresh_scores()
        except Exception:
            pass
        return
    sevmap = {"critical": VulnerabilitySeverity.critical, "high": VulnerabilitySeverity.high,
              "moderate": VulnerabilitySeverity.medium, "low": VulnerabilitySeverity.low}
    row = Vulnerability(
        cve_id=cve_id, asset_id=asset.id, severity=sevmap.get(sev, VulnerabilitySeverity.high),
        status=VulnerabilityStatus.open,
        title=f"Android {month[:7]} security patch — {cve_id}"[:500],
        package_name=f"Android ({label} {month[:7]})"[:255],
        package_version=(asset.os_version or "")[:100] or None,
        detected_at=datetime.now(),
        sources=json.dumps([source]), first_detected_by=source,
    )
    db.add(row)
    db.flush()
    try:
        row.refresh_scores()
    except Exception:
        pass
    new_ids.append(row.id)


def check_android_cves(db: Session, asset, patch_level, manufacturer: str = "",
                       new_ids: Optional[list] = None) -> int:
    """Raise findings for the months the device is behind on. Critical+High
    only, last _MAX_MONTHS months.

    Samsung devices: prefer security.samsungmobile.com's own SMR page per
    month (excludes chipset CVEs Samsung says don't apply — avoids false
    positives the raw ASB would produce, e.g. a Qualcomm-only CVE on a
    Samsung device with a different SoC). Falls back to raw ASB (source
    'android-asb') for any month the SMR page doesn't cover or fails to
    fetch, and for all non-Samsung Android devices.
    """
    if new_ids is None:
        new_ids = []
    ym = _parse_patch_month(patch_level)
    if not ym:
        return 0
    months = _months_after(ym[0], ym[1], date.today())
    is_samsung = "samsung" in (manufacturer or "").lower()
    count = 0
    for month in months:
        pairs = None
        if is_samsung:
            try:
                from app.services import samsung_smr_service
                y, m = int(month[:4]), int(month[5:7])
                pairs = samsung_smr_service.get_smr_month(db, y, m)
            except Exception as e:
                logger.debug("Samsung SMR lookup failed for %s: %s", month, e)
                pairs = None
        if pairs is not None:
            source, label = "samsung-smr", "SMR"
        else:
            pairs = fetch_asb_month(db, month)
            source, label = "android-asb", "ASB"
        for cve, sev in pairs:
            if sev not in _WANT_SEV:
                continue
            before = len(new_ids)
            try:
                _upsert(db, asset, month, cve, sev, new_ids, source=source, label=label)
                count += 1 if len(new_ids) > before else 0
            except Exception as e:
                logger.debug("%s upsert failed (%s on %s): %s", source, cve, asset.id, e)
    return count