Initial curated ITWorx_MCP_Hub source from 368fba691ea8d23403b4d56b53b5ec670d7c5020

scripts/check_cve_exceptions.py

@@ -0,0 +1,181 @@
+#!/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())