"""Validate supply-chain policy and convert scanner evidence into strict decisions.""" from __future__ import annotations import argparse import hashlib import json from datetime import date from pathlib import Path from typing import Any ALL_SEVERITIES = {"unknown", "low", "medium", "high", "critical"} def _read_json(path: Path) -> Any: return json.loads(path.read_text(encoding="utf-8")) def _parse_date(value: object, field: str, failures: list[str]) -> date | None: if not isinstance(value, str): failures.append(f"{field} must be an ISO date") return None try: return date.fromisoformat(value) except ValueError: failures.append(f"{field} must be an ISO date") return None def validate_policy(policy: object, *, today: date | None = None) -> list[str]: """Return failures for malformed, expired, or overlong policy exceptions.""" current = today or date.today() if not isinstance(policy, dict): return ["policy must be a JSON object"] failures: list[str] = [] if policy.get("schema_version") != 1: failures.append("policy schema_version must be 1") vulnerability = policy.get("vulnerability") if not isinstance(vulnerability, dict): failures.append("policy is missing vulnerability settings") else: severities = vulnerability.get("fail_severities") if not isinstance(severities, list) or set(severities) != ALL_SEVERITIES: failures.append("vulnerability policy must fail every normalized severity") failures.extend( _validate_suppressions( vulnerability.get("suppressions"), identifier="id", current=current, fixed_limit=vulnerability.get("max_fixed_suppression_days"), unfixed_limit=vulnerability.get("max_unfixed_suppression_days"), ), ) secret = policy.get("secret") if not isinstance(secret, dict): failures.append("policy is missing secret settings") else: if secret.get("report_redaction_percent") != 100: failures.append("secret reports must use 100 percent redaction") failures.extend( _validate_suppressions( secret.get("suppressions"), identifier="fingerprint", current=current, fixed_limit=secret.get("max_suppression_days"), ), ) license_policy = policy.get("license") if not isinstance(license_policy, dict): failures.append("policy is missing license settings") else: allowed = license_policy.get("allowed_expressions") if not isinstance(allowed, list) or not allowed or not all( isinstance(item, str) and item for item in allowed ): failures.append("license allowed_expressions must be a non-empty string list") return failures def _validate_suppressions( suppressions: object, *, identifier: str, current: date, fixed_limit: object, unfixed_limit: object | None = None, ) -> list[str]: if not isinstance(suppressions, list): return [f"{identifier} suppressions must be a list"] failures: list[str] = [] seen: set[str] = set() for index, item in enumerate(suppressions): prefix = f"suppression[{index}]" if not isinstance(item, dict): failures.append(f"{prefix} must be an object") continue value = item.get(identifier) if not isinstance(value, str) or not value: failures.append(f"{prefix}.{identifier} must be a non-empty string") elif value in seen: failures.append(f"duplicate suppression {identifier}: {value}") else: seen.add(value) if not isinstance(item.get("reason"), str) or not item["reason"].strip(): failures.append(f"{prefix}.reason must be a non-empty string") approved = _parse_date(item.get("approved_on"), f"{prefix}.approved_on", failures) expires = _parse_date(item.get("expires"), f"{prefix}.expires", failures) if approved is None or expires is None: continue if expires < current: failures.append(f"{prefix} expired on {expires.isoformat()}") if expires < approved: failures.append(f"{prefix}.expires precedes approved_on") continue limit = fixed_limit if unfixed_limit is not None: fix_available = item.get("fix_available") if not isinstance(fix_available, bool): failures.append(f"{prefix}.fix_available must be boolean") continue limit = fixed_limit if fix_available else unfixed_limit if not isinstance(limit, int) or isinstance(limit, bool) or limit < 1: failures.append(f"{prefix} suppression limit must be a positive integer") elif (expires - approved).days > limit: failures.append(f"{prefix} exceeds its {limit}-day maximum") return failures def evaluate_audit( policy: dict[str, Any], evidence: object, *, scanner_status: int, today: date | None = None, ) -> dict[str, Any]: """Evaluate uv audit JSON without discarding its meaningful exit status.""" failures = validate_policy(policy, today=today) if scanner_status not in (0, 1): failures.append(f"uv audit failed operationally with status {scanner_status}") if not isinstance(evidence, dict): failures.append("uv audit evidence must be a JSON object") vulnerabilities: list[object] = [] adverse_statuses: list[object] = [] else: vulnerabilities = evidence.get("vulnerabilities", []) adverse_statuses = evidence.get("adverse_statuses", []) if not isinstance(vulnerabilities, list): failures.append("uv audit vulnerabilities must be a list") vulnerabilities = [] if not isinstance(adverse_statuses, list): failures.append("uv audit adverse_statuses must be a list") adverse_statuses = [] suppressions = policy.get("vulnerability", {}).get("suppressions", []) matched: set[str] = set() suppressed: list[str] = [] for item in vulnerabilities: if not isinstance(item, dict) or not isinstance(item.get("id"), str): failures.append("uv audit contains a malformed vulnerability") continue identifiers = {item["id"]} aliases = item.get("aliases", []) if isinstance(aliases, list): identifiers.update(alias for alias in aliases if isinstance(alias, str)) suppression = next( ( candidate for candidate in suppressions if isinstance(candidate, dict) and candidate.get("id") in identifiers ), None, ) if suppression is None: failures.append(f"unsuppressed vulnerability: {item['id']}") continue has_fix = bool(item.get("fix_versions")) if suppression.get("fix_available") != has_fix: failures.append( f"suppression fixability is stale for {suppression.get('id')}: " f"scanner reports fix_available={has_fix}", ) continue matched.add(suppression["id"]) suppressed.append(item["id"]) for suppression in suppressions: if isinstance(suppression, dict) and suppression.get("id") not in matched: failures.append(f"unused vulnerability suppression: {suppression.get('id')}") if adverse_statuses: failures.append(f"uv audit reported {len(adverse_statuses)} adverse package statuses") if scanner_status == 0 and vulnerabilities: failures.append("uv audit returned success while reporting vulnerabilities") if scanner_status == 1 and not vulnerabilities and not adverse_statuses: failures.append("uv audit returned failure without reviewable findings") return { "passed": not failures, "scanner_status": scanner_status, "vulnerability_count": len(vulnerabilities), "suppressed": sorted(suppressed), "failures": failures, } def evaluate_licenses(policy: dict[str, Any], inventory: object) -> dict[str, Any]: """Enforce the exact license-expression allow list.""" failures = validate_policy(policy) if not isinstance(inventory, list): failures.append("license inventory must be a JSON list") inventory = [] license_policy = policy.get("license", {}) allowed = set(license_policy.get("allowed_expressions", [])) excluded = set(license_policy.get("excluded_packages", [])) checked = 0 for item in inventory: if not isinstance(item, dict): failures.append("license inventory contains a malformed row") continue name = item.get("Name") expression = item.get("License") if name in excluded: continue checked += 1 if expression not in allowed: failures.append(f"unapproved license expression for {name}: {expression}") if checked == 0: failures.append("license inventory contains no packaged dependencies") return {"passed": not failures, "checked_packages": checked, "failures": failures} def evaluate_secrets( policy: dict[str, Any], report: object, *, scanner_status: int, today: date | None = None, ) -> dict[str, Any]: """Require redacted Gitleaks evidence and match only bounded fingerprints.""" failures = validate_policy(policy, today=today) if scanner_status not in (0, 1): failures.append(f"gitleaks failed operationally with status {scanner_status}") if not isinstance(report, list): failures.append("gitleaks report must be a JSON list") report = [] suppressions = policy.get("secret", {}).get("suppressions", []) matched: set[str] = set() suppressed: list[str] = [] for finding in report: if not isinstance(finding, dict): failures.append("gitleaks report contains a malformed finding") continue fingerprint = finding.get("Fingerprint") if finding.get("Secret") != "REDACTED" or "REDACTED" not in str( finding.get("Match", ""), ): failures.append(f"gitleaks finding is not fully redacted: {fingerprint}") suppression = next( ( candidate for candidate in suppressions if isinstance(candidate, dict) and candidate.get("fingerprint") == fingerprint ), None, ) if suppression is None: failures.append(f"unsuppressed secret finding: {fingerprint}") else: matched.add(fingerprint) suppressed.append(fingerprint) for suppression in suppressions: if isinstance(suppression, dict) and suppression.get("fingerprint") not in matched: failures.append( f"unused secret suppression: {suppression.get('fingerprint')}", ) if scanner_status == 0 and report: failures.append("gitleaks returned success while reporting findings") if scanner_status == 1 and not report: failures.append("gitleaks returned failure without reviewable findings") return { "passed": not failures, "scanner_status": scanner_status, "finding_count": len(report), "suppressed": sorted(suppressed), "failures": failures, } def bind_sbom(sbom: object, wheel: Path) -> dict[str, Any]: """Bind a CycloneDX SBOM to the exact built wheel by SHA-256.""" if not isinstance(sbom, dict) or sbom.get("bomFormat") != "CycloneDX": raise ValueError("SBOM must be a CycloneDX JSON object") if sbom.get("specVersion") != "1.5": raise ValueError("SBOM must use CycloneDX 1.5") metadata = sbom.get("metadata") if not isinstance(metadata, dict): raise ValueError("SBOM is missing metadata") component = metadata.get("component") if not isinstance(component, dict) or component.get("name") != "obliteratus": raise ValueError("SBOM metadata must describe obliteratus") digest = hashlib.sha256(wheel.read_bytes()).hexdigest() component["hashes"] = [{"alg": "SHA-256", "content": digest}] properties = component.setdefault("properties", []) if not isinstance(properties, list): raise ValueError("SBOM component properties must be a list") properties.append({"name": "obliteratus:distribution-file", "value": wheel.name}) return sbom def _status(path: Path) -> int: return int(path.read_text(encoding="utf-8").strip()) def _write_decision(path: Path, decision: dict[str, Any]) -> int: path.write_text(json.dumps(decision, indent=2, sort_keys=True) + "\n", encoding="utf-8") for failure in decision.get("failures", []): print(f"supply-chain gate failed: {failure}") if decision.get("passed"): print("supply-chain gate passed") return 0 return 1 def _parser() -> argparse.ArgumentParser: parser = argparse.ArgumentParser(description=__doc__) commands = parser.add_subparsers(dest="command", required=True) policy = commands.add_parser("policy") policy.add_argument("--policy", type=Path, required=True) for name in ("audit", "secrets"): command = commands.add_parser(name) command.add_argument("--policy", type=Path, required=True) command.add_argument("--evidence", type=Path, required=True) command.add_argument("--scanner-status", type=Path, required=True) command.add_argument("--decision", type=Path, required=True) licenses = commands.add_parser("licenses") licenses.add_argument("--policy", type=Path, required=True) licenses.add_argument("--evidence", type=Path, required=True) licenses.add_argument("--decision", type=Path, required=True) sbom = commands.add_parser("sbom") sbom.add_argument("--input", type=Path, required=True) sbom.add_argument("--wheel", type=Path, required=True) sbom.add_argument("--output", type=Path, required=True) return parser def main() -> int: args = _parser().parse_args() if args.command == "policy": failures = validate_policy(_read_json(args.policy)) return _write_decision(Path("/dev/null"), {"passed": not failures, "failures": failures}) if args.command == "sbom": bound = bind_sbom(_read_json(args.input), args.wheel) args.output.write_text(json.dumps(bound, indent=2) + "\n", encoding="utf-8") print(f"bound SBOM to {args.wheel.name}") return 0 policy = _read_json(args.policy) evidence = _read_json(args.evidence) if args.command == "audit": decision = evaluate_audit( policy, evidence, scanner_status=_status(args.scanner_status), ) elif args.command == "secrets": decision = evaluate_secrets( policy, evidence, scanner_status=_status(args.scanner_status), ) else: decision = evaluate_licenses(policy, evidence) return _write_decision(args.decision, decision) if __name__ == "__main__": raise SystemExit(main())