"""Validate and summarize one Trivy image report without accepting ambiguous JSON. Trivy returning zero is not sufficient release evidence: an empty, duplicate-key or unrelated report can otherwise look like a clean scan. This parser binds the report to the exact image ID, requires package coverage, and fails on every fixable or unreviewed HIGH/CRITICAL finding. An upstream-unfixed finding is non-blocking only when its complete identity matches an explicit reviewed baseline. """ from __future__ import annotations import argparse import json from collections import Counter from pathlib import Path from typing import Any VulnerabilityIdentity = tuple[str, str, str, str] def reject_duplicate_keys(pairs: list[tuple[str, Any]]) -> dict[str, Any]: result: dict[str, Any] = {} for key, value in pairs: if key in result: raise ValueError(f"duplicate JSON key: {key}") result[key] = value return result def load_reviewed_unfixed(path: Path | None) -> set[VulnerabilityIdentity]: if path is None: return set() data = json.loads( path.read_text(encoding="utf-8"), object_pairs_hook=reject_duplicate_keys ) if not isinstance(data, dict) or data.get("schema_version") != 1: raise ValueError("reviewed-unfixed baseline must use schema_version 1") entries = data.get("vulnerabilities") if not isinstance(entries, list): raise ValueError("reviewed-unfixed vulnerabilities must be an array") reviewed: set[VulnerabilityIdentity] = set() for entry in entries: if not isinstance(entry, dict): raise ValueError("reviewed-unfixed entry must be an object") identity = tuple( entry.get(field) for field in ("vulnerability_id", "package", "installed_version", "severity") ) if not all(isinstance(value, str) and value for value in identity): raise ValueError("reviewed-unfixed identity fields must be non-empty strings") typed_identity = (identity[0], identity[1], identity[2], identity[3].upper()) if typed_identity[3] not in {"HIGH", "CRITICAL"}: raise ValueError("reviewed-unfixed severity must be HIGH or CRITICAL") if typed_identity in reviewed: raise ValueError(f"duplicate reviewed-unfixed identity: {typed_identity!r}") reviewed.add(typed_identity) return reviewed def validate( report: Path, expected_image_id: str, reviewed_unfixed_path: Path | None = None, ) -> dict[str, Any]: data = json.loads( report.read_text(encoding="utf-8"), object_pairs_hook=reject_duplicate_keys ) if not isinstance(data, dict): raise ValueError("Trivy report root must be an object") if data.get("ArtifactType") != "container_image": raise ValueError("Trivy report is not a container-image scan") metadata = data.get("Metadata") if not isinstance(metadata, dict) or metadata.get("ImageID") != expected_image_id: observed = metadata.get("ImageID") if isinstance(metadata, dict) else None raise ValueError( f"Trivy report image ID {observed!r} does not match {expected_image_id!r}" ) results = data.get("Results") if not isinstance(results, list) or not results: raise ValueError("Trivy report has no package result coverage") covered_targets: list[str] = [] severities: Counter[str] = Counter() reviewed_unfixed = load_reviewed_unfixed(reviewed_unfixed_path) observed_reviewed: set[VulnerabilityIdentity] = set() fixable_high_critical = 0 unreviewed_unfixed_high_critical = 0 findings = 0 for result in results: if not isinstance(result, dict): raise ValueError("Trivy result entry must be an object") target = result.get("Target") result_class = result.get("Class") if isinstance(target, str) and target and result_class in {"os-pkgs", "lang-pkgs"}: covered_targets.append(target) vulnerabilities = result.get("Vulnerabilities") if vulnerabilities is None: continue if not isinstance(vulnerabilities, list): raise ValueError("Trivy Vulnerabilities must be an array or null") for finding in vulnerabilities: if not isinstance(finding, dict): raise ValueError("Trivy vulnerability entry must be an object") severity = finding.get("Severity") if not isinstance(severity, str) or not severity: raise ValueError("Trivy vulnerability is missing Severity") normalized_severity = severity.upper() severities[normalized_severity] += 1 findings += 1 if normalized_severity not in {"HIGH", "CRITICAL"}: continue vulnerability_id = finding.get("VulnerabilityID") package = finding.get("PkgName") installed_version = finding.get("InstalledVersion") if not all( isinstance(value, str) and value for value in (vulnerability_id, package, installed_version) ): raise ValueError("HIGH/CRITICAL finding is missing its exact identity") fixed_version = finding.get("FixedVersion") if isinstance(fixed_version, str) and fixed_version.strip(): fixable_high_critical += 1 continue identity = ( vulnerability_id, package, installed_version, normalized_severity, ) if identity in reviewed_unfixed: observed_reviewed.add(identity) else: unreviewed_unfixed_high_critical += 1 if not covered_targets: raise ValueError("Trivy report covers no OS or language package target") return { "schema_version": 2, "image_id": expected_image_id, "artifact_name": data.get("ArtifactName"), "covered_targets": sorted(set(covered_targets)), "finding_count": findings, "severities": dict(sorted(severities.items())), "high_critical_findings": severities["HIGH"] + severities["CRITICAL"], "reviewed_unfixed_high_critical": len(observed_reviewed), "fixable_high_critical": fixable_high_critical, "unreviewed_unfixed_high_critical": unreviewed_unfixed_high_critical, "stale_reviewed_unfixed_entries": len(reviewed_unfixed - observed_reviewed), "release_blockers": fixable_high_critical + unreviewed_unfixed_high_critical, } def main() -> int: parser = argparse.ArgumentParser() parser.add_argument("--report", type=Path, required=True) parser.add_argument("--image-id", required=True) parser.add_argument("--summary", type=Path, required=True) parser.add_argument("--reviewed-unfixed", type=Path) args = parser.parse_args() summary = validate(args.report, args.image_id, args.reviewed_unfixed) args.summary.parent.mkdir(parents=True, exist_ok=True) args.summary.write_text( json.dumps(summary, indent=2, sort_keys=True) + "\n", encoding="utf-8", newline="\n" ) print( f"Trivy coverage: {len(summary['covered_targets'])} target(s), " f"{summary['finding_count']} finding(s), " f"{summary['reviewed_unfixed_high_critical']} reviewed upstream-unfixed, " f"{summary['release_blockers']} HIGH/CRITICAL blocker(s)" ) return 1 if summary["release_blockers"] else 0 if __name__ == "__main__": raise SystemExit(main())