Initial public ModelForge release
This commit is contained in:
Jens
@@ -0,0 +1,180 @@
|
||||
"""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())
|
||||
Reference in New Issue
Block a user