ITWorx_MCP_Hub-Public/scripts/check_cve_exceptions.py

#!/usr/bin/env python3
"""Enforce the CVE exception policy against a vulnerability scan.

Every CRITICAL/HIGH finding in --scan-summary must be covered by a current,
explicit, fully-populated, *approved* exception in --exceptions, or the gate
fails. This never suppresses scanner output -- it cross-references it.
Specifically fails on:

- a CRITICAL/HIGH finding (package, CVE) with no matching exception at all;
- a matching exception whose approvalStatus is not exactly "approved"
  (covers "pending", "rejected", or any other value -- there is no default
  grant);
- a matching exception that has expired (expiryDate in the past) or is past
  its mandatory rescan date;
- a matching exception whose recorded affectedImageDigests disagree with
  --image-digests for any service both documents name (the digest changed --
  the exception was reviewed against a build that no longer exists);
- an exception missing any required field.

See docs/09-release/CVE_EXCEPTIONS.md for the full policy and
contracts/schemas/ (release-record.schema.json's sibling documents) for
context. This script has no "--force"/override flag by design.
"""

from __future__ import annotations

import argparse
import sys
from datetime import date
from pathlib import Path
from typing import Any

_ROOT_DIR = Path(__file__).resolve().parents[1]
if str(_ROOT_DIR) not in sys.path:
    sys.path.insert(0, str(_ROOT_DIR))

from scripts.packlib import load_json  # noqa: E402

REQUIRED_FIELDS = (
    "id",
    "cve",
    "package",
    "installedVersion",
    "severity",
    "scanner",
    "scanDate",
    "affectedImageDigests",
    "reachabilityAnalysis",
    "compensatingControls",
    "riskOwner",
    "approver",
    "approvalStatus",
    "expiryDate",
    "mandatoryRescanDate",
    "remediationTrigger",
    "evidenceReference",
)


def _findings(scan: dict[str, Any]) -> list[tuple[str, str, str]]:
    """Return (package, cve, severity) for every CRITICAL/HIGH finding."""
    findings = []
    for row in scan.get("unresolvedCriticalOrHigh", []):
        for cve in row["cves"]:
            findings.append((row["package"], cve, row.get("severity", "UNKNOWN")))
    return findings


def check(
    scan: dict[str, Any],
    exceptions_doc: dict[str, Any],
    image_digests: dict[str, str] | None,
    today: date,
) -> list[str]:
    errors: list[str] = []
    by_key: dict[tuple[str, str], dict[str, Any]] = {}
    for exception in exceptions_doc.get("exceptions", []):
        missing = [field for field in REQUIRED_FIELDS if not exception.get(field)]
        # riskOwner/approver/compensatingControls may be legitimately empty
        # only while approvalStatus is "pending" -- once "approved" they are
        # mandatory. Re-check those three explicitly below rather than
        # excluding them from REQUIRED_FIELDS entirely, so a pending draft
        # from scripts/draft_cve_exceptions.py isn't flagged as malformed.
        missing = [f for f in missing if f not in {"riskOwner", "approver", "compensatingControls"}]
        if missing:
            errors.append(
                f"exception {exception.get('id', '<no id>')} is missing required fields: {missing}"
            )
            continue
        key = (exception["package"], exception["cve"])
        by_key[key] = exception

    for package, cve, severity in _findings(scan):
        key = (package, cve)
        exception = by_key.get(key)
        if exception is None:
            errors.append(
                f"UNAPPROVED: {severity} finding {package}/{cve} has no exception recorded at all"
            )
            continue

        status = exception.get("approvalStatus")
        if status != "approved":
            errors.append(
                f"UNAPPROVED: {package}/{cve} exception status is {status!r}, "
                "not 'approved' -- no default grant"
            )
            continue
        for role in ("riskOwner", "approver"):
            if not exception.get(role):
                errors.append(
                    f"UNAPPROVED: {package}/{cve} is marked approved but has no {role} recorded"
                )
        if not exception.get("compensatingControls"):
            errors.append(
                f"UNAPPROVED: {package}/{cve} is marked approved but records no "
                "compensating controls"
            )

        expiry = date.fromisoformat(exception["expiryDate"])
        if expiry < today:
            errors.append(f"EXPIRED: {package}/{cve} exception expired on {expiry.isoformat()}")

        rescan_due = date.fromisoformat(exception["mandatoryRescanDate"])
        if rescan_due < today:
            errors.append(
                f"RESCAN OVERDUE: {package}/{cve} exception's mandatory rescan date "
                f"{rescan_due.isoformat()} has passed"
            )

        if image_digests:
            recorded = exception.get("affectedImageDigests", {})
            for service, current_digest in image_digests.items():
                recorded_digest = recorded.get(service)
                if recorded_digest and recorded_digest != current_digest:
                    errors.append(
                        f"DIGEST CHANGED: {package}/{cve} exception recorded "
                        f"{service}={recorded_digest}, but the current build is "
                        f"{current_digest} -- the exception was reviewed against a "
                        "build that no longer exists and must be re-approved"
                    )

    return errors


def main() -> int:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--scan-summary", required=True, type=Path)
    parser.add_argument("--exceptions", required=True, type=Path)
    parser.add_argument(
        "--image-digests",
        type=Path,
        help="evidence/<packet>/image-digests.json; when given, exceptions whose "
        "recorded digest for a named service disagrees with the current build fail.",
    )
    args = parser.parse_args()

    scan = load_json(args.scan_summary)
    exceptions_doc = load_json(args.exceptions)
    image_digests: dict[str, str] | None = None
    if args.image_digests:
        digest_doc = load_json(args.image_digests)
        image_digests = {
            service: (entry.get("digest") or entry.get("imageId"))
            for service, entry in digest_doc.get("images", {}).items()
        }

    errors = check(scan, exceptions_doc, image_digests, date.today())
    if errors:
        print("CVE exception gate: FAIL")
        for error in errors:
            print(f"  - {error}")
        return 1
    print(
        "CVE exception gate: PASS (every CRITICAL/HIGH finding has a current, approved exception)"
    )
    return 0


if __name__ == "__main__":
    raise SystemExit(main())