vulncheck/app/services/mozilla_advisory_service.py

"""
Mozilla Foundation Security Advisories (MFSA) — per-CVE severity + fix train.

Source: github.com/mozilla/foundation-security-advisories (announce/YYYY/*.yml).
Each MFSA yml maps CVE → {impact, title} plus an advisory-level `fixed_in`
(e.g. ["Firefox 128", "Firefox ESR 115.13"]). Mozilla publishes an `impact`
rating (critical/high/moderate/low) but NO numeric CVSS — so this is a
SEVERITY + description source, not a CVSS source. It fills the gap where fresh
Firefox CVEs have no score yet in NVD/cvelistV5.

`fixed_in` is the AUTHORITATIVE regular-vs-ESR discriminator (an advisory whose
fixed_in lists only "Firefox ESR …" does not affect regular Firefox) — stored
here for the Firefox-scan ESR exclusion, cf. [[ghsa-unreviewed-no-version-range]]
sibling reference on why cvelistV5 alone can't tell them apart.

The parser is deliberately line-based (no PyYAML dependency): the MFSA schema is
regular — top-level `fixed_in:` list, then an `advisories:` map of
`  CVE-…:` → `    impact:` / `    title:`.
"""
import json
import logging
import re
from datetime import datetime, timedelta
from typing import Dict, List, Optional, Tuple

from sqlalchemy.orm import Session

from app.models.setting import Setting

logger = logging.getLogger(__name__)

INDEX_SETTING = "mozilla_mfsa_index"
INDEX_TS_SETTING = "mozilla_mfsa_index_ts"
TTL_HOURS = 24
_API = "https://api.github.com/repos/mozilla/foundation-security-advisories"

_IMPACT_TO_SEV = {
    "critical": "critical",
    "high": "high",
    "moderate": "medium",
    "low": "low",
    "none": "none",
}

_CVE_KEY = re.compile(r"^  (CVE-\d{4}-\d+):\s*$")
_IMPACT = re.compile(r"^    impact:\s*([A-Za-z]+)")
_TITLE = re.compile(r"^    title:\s*(.+?)\s*$")
_LIST_ITEM = re.compile(r"^-\s*(.+?)\s*$")


def _gh_headers(db: Session) -> dict:
    h = {"Accept": "application/vnd.github+json", "X-GitHub-Api-Version": "2022-11-28"}
    try:
        from app.auth.setting_crypto import read_setting_value
        pat = (read_setting_value(db, "github_pat") or "").strip()
        if pat:
            h["Authorization"] = f"Bearer {pat}"
    except Exception:
        pass
    return h


def _is_esr_only(fixed_in: List[str]) -> bool:
    """True when every Firefox entry is an ESR build — the advisory does not
    affect regular Firefox. 'Firefox ESR 115.13' → ESR; 'Firefox 128' → regular.
    Thunderbird/other entries are ignored for the Firefox decision."""
    ff = [f for f in fixed_in if "firefox" in f.lower()]
    if not ff:
        return False
    return all("esr" in f.lower() for f in ff)


def _parse_yml(text: str) -> Tuple[List[str], Dict[str, dict]]:
    """Line-parse one MFSA yml → (fixed_in list, {cve: {impact, title}})."""
    fixed_in: List[str] = []
    cves: Dict[str, dict] = {}
    section = None          # "fixed_in" | "advisories" | None
    cur = None
    for line in text.splitlines():
        if line.startswith("fixed_in:"):
            section = "fixed_in"
            continue
        if line.startswith("advisories:"):
            section = "advisories"
            cur = None
            continue
        # A non-indented, non-list line ends the current top-level block.
        if line and not line[0].isspace() and not line.startswith("-"):
            section = None
            cur = None
        if section == "fixed_in":
            m = _LIST_ITEM.match(line)
            if m:
                fixed_in.append(m.group(1))
        elif section == "advisories":
            mc = _CVE_KEY.match(line)
            if mc:
                cur = mc.group(1).upper()
                cves[cur] = {}
                continue
            if cur:
                mi = _IMPACT.match(line)
                if mi:
                    cves[cur]["impact"] = mi.group(1).lower()
                    continue
                mt = _TITLE.match(line)
                if mt and "title" not in cves[cur]:
                    cves[cur]["title"] = mt.group(1)
    return fixed_in, cves


def build_index(db: Session, years: Optional[List[int]] = None) -> Dict[str, dict]:
    """Walk the MFSA repo for the given years, build {cve: {sev, title,
    fixed_in, esr_only}}, cache it. Defaults to current + previous year (the
    window where CVEs are fresh enough that NVD may still lag)."""
    import httpx

    if years is None:
        y = datetime.now().year
        years = [y, y - 1]

    index: Dict[str, dict] = {}
    headers = _gh_headers(db)
    with httpx.Client(timeout=20.0, follow_redirects=True, headers=headers) as client:
        for year in years:
            try:
                r = client.get(f"{_API}/contents/announce/{year}")
                if r.status_code == 403 and r.headers.get("x-ratelimit-remaining") == "0":
                    logger.warning("MFSA: GitHub rate limit hit — set github_pat for 5000/h")
                    break
                if r.status_code != 200:
                    continue
                files = [f for f in (r.json() or [])
                         if isinstance(f, dict) and str(f.get("name", "")).endswith(".yml")]
            except Exception as e:
                logger.debug("MFSA: listing %s failed: %s", year, e)
                continue
            for f in files:
                url = f.get("download_url")
                if not url:
                    continue
                try:
                    rr = client.get(url)
                    if rr.status_code != 200:
                        continue
                    fixed_in, cves = _parse_yml(rr.text)
                    esr_only = _is_esr_only(fixed_in)
                    for cve_id, data in cves.items():
                        sev = _IMPACT_TO_SEV.get(data.get("impact") or "")
                        # First writer wins per CVE (a CVE can appear in several
                        # MFSAs for different products; the Firefox one is fine).
                        if cve_id not in index:
                            index[cve_id] = {
                                "sev": sev,
                                "title": data.get("title"),
                                "fixed_in": fixed_in,
                                "esr_only": esr_only,
                            }
                except Exception as e:
                    logger.debug("MFSA: parse %s failed: %s", f.get("name"), e)

    _store(db, index)
    logger.info("MFSA index built: %d CVEs across years %s", len(index), years)
    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 load_index(db: Session) -> Optional[Dict[str, dict]]:
    ts = db.query(Setting).filter(Setting.key == INDEX_TS_SETTING).first()
    row = db.query(Setting).filter(Setting.key == INDEX_SETTING).first()
    if not ts or not row or not row.value:
        return None
    try:
        if datetime.now() - datetime.fromisoformat(ts.value) > timedelta(hours=TTL_HOURS):
            return None
        return json.loads(row.value)
    except (ValueError, json.JSONDecodeError):
        return None


def get_index(db: Session) -> Dict[str, dict]:
    """Cached index, lazily (re)built when absent/stale. Build failure → {}."""
    idx = load_index(db)
    if idx is not None:
        return idx
    try:
        return build_index(db)
    except Exception as e:
        logger.warning("MFSA index build failed: %s", e)
        return {}


def apply_mozilla_severity(db: Session, cve_ids: List[str]) -> int:
    """Fill severity + description for Firefox CVEs from Mozilla's authoritative
    impact rating. Only overrides severity when the vuln has NO CVSS-derived
    value (cvss_score is None → severity is a default placeholder); Mozilla has
    no CVSS number so it must not clobber a real score-derived severity."""
    from app.models.vulnerability import Vulnerability, VulnerabilitySeverity

    wanted = [c.upper() for c in cve_ids if c]
    if not wanted:
        return 0
    idx = get_index(db)
    if not idx:
        return 0
    hits = [c for c in wanted if c in idx]
    if not hits:
        return 0

    updated = 0
    rows = db.query(Vulnerability).filter(Vulnerability.cve_id.in_(hits)).all()
    for v in rows:
        data = idx.get((v.cve_id or "").upper())
        if not data:
            continue
        sev = data.get("sev")
        if sev and v.cvss_score is None:
            new_sev = getattr(VulnerabilitySeverity, sev, None)
            if new_sev is not None and v.severity != new_sev:
                v.severity = new_sev
                updated += 1
        if not v.description and data.get("title"):
            v.description = data["title"]
    if updated:
        db.commit()
    return updated


if __name__ == "__main__":
    # ponytail: one self-check for the line parser + ESR discriminator — the two
    # non-trivial bits. Run: python -m app.services.mozilla_advisory_service
    sample = """announced: July 9th, 2024
impact: high
fixed_in:
- Firefox 128
- Firefox ESR 115.13
title: Security Vulnerabilities fixed in Firefox 128
advisories:
  CVE-2024-6601:
    title: Race condition in permission assignment
    impact: moderate
    reporter: Andreas Farre
  CVE-2024-6602:
    title: Memory corruption in NSS
    impact: critical
"""
    fixed_in, cves = _parse_yml(sample)
    assert fixed_in == ["Firefox 128", "Firefox ESR 115.13"], fixed_in
    assert cves["CVE-2024-6601"] == {"title": "Race condition in permission assignment",
                                     "impact": "moderate"}, cves["CVE-2024-6601"]
    assert cves["CVE-2024-6602"]["impact"] == "critical"
    assert _is_esr_only(["Firefox 128", "Firefox ESR 115.13"]) is False  # has regular
    assert _is_esr_only(["Firefox ESR 115.13"]) is True                  # ESR only
    assert _is_esr_only(["Thunderbird 128"]) is False                    # no firefox
    assert _IMPACT_TO_SEV["moderate"] == "medium"
    print("mozilla_advisory_service self-check OK")