apply-exceptions.py

#!/usr/bin/env python3
"""Apply reviewed CVE exceptions (exceptions.json) to a grype scan JSON.

An exception says "this CVE does not affect this binary package, as built from
this exact source version", with the reason, who reviewed it, and when the
review expires. It is deliberately narrow so it can never silently cover
something nobody looked at:

  * it names the CVEs it covers explicitly (never "everything against X");
  * it is pinned to the reviewed SOURCE versions. A binNMU / Raspbian rebuild
    (+b1, +b2 ...) is the same source and stays covered; a new upstream
    release or a Debian security update (+deb13u1) is a new source version
    and is not;
  * it lists the binary packages it covers. Another binary from the same
    source is not covered unless it is listed, either as covered (`packages`)
    or as reviewed and still affected (`affectedPackages`);
  * it expires.

Matches an exception covers are moved from `matches` to `exceptedMatches` in
the scan JSON, each tagged with the exception that applied, so they drop out of
every count but stay in the published JSON and the report. Matches an
exception was written for but no longer covers (expired, new source version,
unlisted binary from the reviewed source) stay in `matches` and are tagged
with `exceptionReview` saying why, so the report can ask for a re-review.

Re-runnable: the previous run's exceptedMatches are first put back and its
tags removed, so applying to an already-processed scan (the pipeline caches
scans between runs) gives the same result as applying to a fresh one, and an
exception that has since expired or been removed stops applying.

Usage:
  apply-exceptions.py SCAN.json [--exceptions exceptions.json] [--quiet]
  apply-exceptions.py --validate [--exceptions exceptions.json]
"""
from __future__ import annotations

import argparse
import json
import os
import pathlib
import re
import sys
from datetime import date, timedelta

HERE = pathlib.Path(__file__).resolve().parent
DEFAULT_FILE = HERE / "exceptions.json"

# OpenVEX "not_affected" justifications; the only status supported is not_affected.
JUSTIFICATIONS = {
    "component_not_present",
    "vulnerable_code_not_present",
    "vulnerable_code_not_in_execute_path",
    "vulnerable_code_cannot_be_controlled_by_adversary",
    "inline_mitigations_already_exist",
}
MAX_LIFETIME = timedelta(days=731)     # an exception must be re-reviewed at least every two years
WARN_BEFORE = timedelta(days=30)       # start warning this long before an exception expires
VULN_ID = re.compile(r"^(CVE-\d{4}-\d{4,}|GHSA(-[23456789cfghjmpqrvwx]{4}){3})$")
REQUIRED = {"id", "source", "sourceVersions", "packages", "vulnerabilities", "status",
            "justification", "reason", "reviewedBy", "reviewed", "expires"}
OPTIONAL = {"affectedPackages", "evidence"}


class ExceptionsError(ValueError):
    pass


def _date(entry_id: str, field: str, value) -> date:
    try:
        return date.fromisoformat(value)
    except (TypeError, ValueError):
        raise ExceptionsError(f"{entry_id}: {field} must be an ISO date (YYYY-MM-DD), got {value!r}")


def _str_list(entry_id: str, field: str, value, pattern: re.Pattern | None = None) -> list[str]:
    if not isinstance(value, list) or not value or not all(isinstance(x, str) and x for x in value):
        raise ExceptionsError(f"{entry_id}: {field} must be a non-empty list of strings")
    if len(set(value)) != len(value):
        raise ExceptionsError(f"{entry_id}: {field} has duplicates")
    if pattern:
        bad = [x for x in value if not pattern.match(x)]
        if bad:
            raise ExceptionsError(f"{entry_id}: {field} has malformed IDs: {', '.join(bad)}")
    return value


def load_exceptions(path: pathlib.Path) -> list[dict]:
    """Load and strictly validate the exceptions file.

    Strict on purpose: a typo in a field name or package name would otherwise
    make an exception silently match nothing (or the wrong thing)."""
    try:
        doc = json.loads(path.read_text())
    except (OSError, json.JSONDecodeError) as e:
        raise ExceptionsError(f"cannot read {path}: {e}")
    if not isinstance(doc, dict) or doc.get("schema") != 1 or not isinstance(doc.get("exceptions"), list):
        raise ExceptionsError(f"{path}: expected {{\"schema\": 1, \"exceptions\": [...]}}")
    unknown_top = set(doc) - {"schema", "comment", "exceptions"}
    if unknown_top:
        raise ExceptionsError(f"{path}: unknown top-level keys: {', '.join(sorted(unknown_top))}")

    seen_ids: set[str] = set()
    seen_pairs: dict[tuple[str, str], str] = {}
    out = []
    for i, e in enumerate(doc["exceptions"]):
        eid = e.get("id") if isinstance(e, dict) else None
        label = eid if isinstance(eid, str) and eid else f"exceptions[{i}]"
        if not isinstance(e, dict):
            raise ExceptionsError(f"{label}: must be an object")
        missing = REQUIRED - set(e)
        if missing:
            raise ExceptionsError(f"{label}: missing {', '.join(sorted(missing))}")
        unknown = set(e) - REQUIRED - OPTIONAL
        if unknown:
            raise ExceptionsError(f"{label}: unknown keys {', '.join(sorted(unknown))}")
        if not re.match(r"^[a-z0-9][a-z0-9._-]*$", eid):
            raise ExceptionsError(f"{label}: id must be lower-case letters, digits, '.', '_' or '-'")
        if eid in seen_ids:
            raise ExceptionsError(f"{label}: duplicate id")
        seen_ids.add(eid)
        if not isinstance(e["source"], str) or not e["source"]:
            raise ExceptionsError(f"{label}: source must be the Debian source package name")
        _str_list(label, "sourceVersions", e["sourceVersions"])
        pkgs = _str_list(label, "packages", e["packages"])
        affected = _str_list(label, "affectedPackages", e["affectedPackages"]) if "affectedPackages" in e else []
        both = set(pkgs) & set(affected)
        if both:
            raise ExceptionsError(f"{label}: {', '.join(sorted(both))} in both packages and affectedPackages")
        vulns = _str_list(label, "vulnerabilities", e["vulnerabilities"], VULN_ID)
        if e["status"] != "not_affected":
            raise ExceptionsError(f"{label}: status must be \"not_affected\"")
        if e["justification"] not in JUSTIFICATIONS:
            raise ExceptionsError(f"{label}: justification must be one of {', '.join(sorted(JUSTIFICATIONS))}")
        for f in ("reason", "reviewedBy"):
            if not isinstance(e[f], str) or not e[f].strip():
                raise ExceptionsError(f"{label}: {f} must be a non-empty string")
        if "evidence" in e:
            _str_list(label, "evidence", e["evidence"])
        reviewed = _date(label, "reviewed", e["reviewed"])
        expires = _date(label, "expires", e["expires"])
        if expires <= reviewed:
            raise ExceptionsError(f"{label}: expires must be after reviewed")
        if expires - reviewed > MAX_LIFETIME:
            raise ExceptionsError(f"{label}: expires is more than two years after reviewed; re-review instead")
        for p in pkgs:
            for v in vulns:
                prev = seen_pairs.get((p, v))
                if prev:
                    raise ExceptionsError(f"{label}: {p} / {v} is already covered by {prev}")
                seen_pairs[(p, v)] = eid
        out.append({**e, "affectedPackages": affected, "_reviewed": reviewed, "_expires": expires})
    return out


def deb_source(artifact: dict) -> tuple[str, str]:
    """(source name, source version) of a deb artifact as grype reports it.

    syft records the source in `upstreams`: name only when the source version
    equals the binary version, name + version when they differ (binNMUs such
    as 0.2.13-1.1+b3 come from source 0.2.13-1.1). No upstream at all means
    the binary has the source's own name and version."""
    ups = artifact.get("upstreams") or []
    name = artifact.get("name", "")
    version = artifact.get("version", "")
    if ups:
        name = ups[0].get("name") or name
        version = ups[0].get("version") or version
    return name, version


def restore(scan: dict) -> None:
    """Undo a previous application, so this one starts from grype's own result."""
    back = scan.pop("exceptedMatches", None) or []
    for m in back:
        m.pop("exception", None)
    scan["matches"] = (scan.get("matches") or []) + back
    for m in scan["matches"]:
        m.pop("exceptionReview", None)
    scan.pop("exceptions", None)


def apply(scan: dict, exceptions: list[dict], today: date) -> dict:
    """Apply in place. Returns {"excepted": [...], "review": [...]} for reporting."""
    restore(scan)
    by_vuln: dict[str, list[dict]] = {}
    for e in exceptions:
        for v in e["vulnerabilities"]:
            by_vuln.setdefault(v, []).append(e)

    kept, excepted, review = [], [], []
    for m in scan["matches"]:
        a = m.get("artifact") or {}
        vid = (m.get("vulnerability") or {}).get("id", "")
        cands = by_vuln.get(vid)
        if not cands or a.get("type") != "deb":
            kept.append(m)
            continue
        name = a.get("name", "")
        src, src_ver = deb_source(a)
        verdict = None          # ("except", e) | ("review", e, why) | None
        for e in cands:
            listed = name in e["packages"]
            if not listed and (src != e["source"] or name in e["affectedPackages"]):
                continue        # not this exception's concern
            if not listed:
                verdict = ("review", e, f"{name} is built from {src} but this exception does not list it "
                                        f"(add it to packages or affectedPackages after reviewing it)")
            elif src != e["source"]:
                verdict = ("review", e, f"{name} is now built from source {src}, not {e['source']}")
            elif src_ver not in e["sourceVersions"]:
                verdict = ("review", e, f"source {src} {src_ver} has not been reviewed "
                                        f"(reviewed: {', '.join(e['sourceVersions'])})")
            elif today > e["_expires"]:
                verdict = ("review", e, f"exception expired on {e['expires']}")
            else:
                verdict = ("except", e)
                break
        if verdict is None:
            kept.append(m)
        elif verdict[0] == "except":
            e = verdict[1]
            m["exception"] = {"id": e["id"], "justification": e["justification"], "reason": e["reason"],
                              "reviewedBy": e["reviewedBy"], "reviewed": e["reviewed"], "expires": e["expires"]}
            excepted.append(m)
        else:
            _, e, why = verdict
            m["exceptionReview"] = {"id": e["id"], "why": why}
            kept.append(m)
            review.append(m)

    scan["matches"] = kept
    scan["exceptedMatches"] = excepted
    scan["exceptions"] = {"applied": today.isoformat(),
                          "ids": sorted({m["exception"]["id"] for m in excepted})}
    return {"excepted": excepted, "review": review}


def main() -> int:
    ap = argparse.ArgumentParser(description=__doc__.splitlines()[0])
    ap.add_argument("scan", nargs="?", help="grype JSON scan, rewritten in place")
    ap.add_argument("--exceptions", default=str(DEFAULT_FILE), help=f"exceptions file (default: {DEFAULT_FILE.name} beside this script)")
    ap.add_argument("--validate", action="store_true", help="only validate the exceptions file and report expiries")
    ap.add_argument("--today", default=None, help="evaluate expiry as of this date (YYYY-MM-DD); for testing")
    ap.add_argument("--quiet", action="store_true")
    args = ap.parse_args()
    if not args.validate and not args.scan:
        ap.error("a scan file is required unless --validate is given")

    gha = os.environ.get("GITHUB_ACTIONS") == "true"
    def warn(title: str, msg: str):
        print(f"::warning title={title}::{msg}" if gha else f"warning: {msg}", file=sys.stderr)

    try:
        today = date.fromisoformat(args.today) if args.today else date.today()
        exceptions = load_exceptions(pathlib.Path(args.exceptions))
    except (ExceptionsError, ValueError) as e:
        print(f"::error title=exceptions.json invalid::{e}" if gha else f"error: {e}", file=sys.stderr)
        return 2

    if args.validate:
        for e in exceptions:
            if today > e["_expires"]:
                warn("Exception expired", f"{e['id']} expired on {e['expires']}; its CVEs are reported again until it is re-reviewed")
            elif e["_expires"] - today <= WARN_BEFORE:
                warn("Exception expiring", f"{e['id']} expires on {e['expires']}")
        print(f"{args.exceptions}: {len(exceptions)} exception(s) valid")
        return 0

    path = pathlib.Path(args.scan)
    scan = json.loads(path.read_text())
    res = apply(scan, exceptions, today)
    path.write_text(json.dumps(scan))

    reviews: dict[tuple[str, str], set[str]] = {}
    for m in res["review"]:
        r = m["exceptionReview"]
        reviews.setdefault((r["id"], r["why"]), set()).add(m["vulnerability"]["id"])
    for (eid, why), vids in sorted(reviews.items()):
        warn("Exception needs review", f"{path.name}: {eid}: {why} [{', '.join(sorted(vids))}]")
    if not args.quiet:
        n_cves = len({m["vulnerability"]["id"] for m in res["excepted"]})
        print(f"apply-exceptions: {len(res['excepted'])} finding(s) / {n_cves} CVE(s) excepted, "
              f"{len(res['review'])} finding(s) need review")
    return 0


if __name__ == "__main__":
    sys.exit(main())