181 lines
7.4 KiB
Python
181 lines
7.4 KiB
Python
"""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())
|