mirror of
https://github.com/elder-plinius/OBLITERATUS.git
synced 2026-08-17 16:37:30 +02:00
ci: add reproducible supply-chain gates
This commit is contained in:
Joseph Magly
@@ -0,0 +1,382 @@
|
||||
"""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())
|
||||
Reference in New Issue
Block a user