#!/usr/bin/env python3 """Fail-closed evaluator and report normalizers for the managed security gate. The scanners deliberately collect findings without deciding release eligibility. This standard-library-only program validates that every expected target produced a complete report, applies the narrowly scoped accepted-risk policy, and always writes one machine-readable decision. """ from __future__ import annotations import argparse import hashlib import ipaddress import json import os import re import stat import subprocess import sys import tempfile from dataclasses import asdict, dataclass from datetime import UTC, date, datetime, timedelta from pathlib import Path from typing import Any, Callable, Final from urllib.parse import urlsplit import tomllib SCHEMA_VERSION: Final = 1 PINNED_SCANNER_VERSIONS: Final = { "pip-audit": "2.10.1", "gitleaks": "8.30.1", "trivy": "0.74.0", } SOURCE_COMPONENTS: Final = { "backend": ("backend/pyproject.toml", "modelforge-api"), "node-agent": ("node-agent/pyproject.toml", "modelforge-node-agent"), "runtime-worker": ("runtime-worker/pyproject.toml", "modelforge-runtime-worker"), } CANONICAL_SOURCE_REPOSITORY: Final = "https://git.example.com/example/modelforge" PUBLIC_SOURCE_LABEL_PLACEHOLDER: Final = "https://git.example.com/example/modelforge" PUBLIC_SOURCE_MANIFEST: Final = "PUBLIC_SOURCE_MANIFEST.json" BUILD_TOOLCHAIN_LOCK: Final = "config/security/build-toolchain.lock" BUILD_TOOLCHAIN_CONTAINER_PATH: Final = "/tmp/modelforge-build-toolchain.lock" BUILD_TOOLCHAIN_PINS: Final = { "editables": "0.5", "hatchling": "1.32.4", "msgpack": "1.2.2", "packaging": "26.3", "pathspec": "1.1.1", "pip": "26.2.1", "pluggy": "1.6.0", "setuptools": "84.0.0", "tomlkit": "0.14.0", "trove-classifiers": "2026.9.21.13", "wheel": "0.48.0", } PEP517_BUILD_SYSTEMS: Final = { "backend": ("hatchling.build", {"hatchling": "1.32.4"}), "node-agent": ("setuptools.build_meta", {"setuptools": "84.0.0"}), "runtime-worker": ("setuptools.build_meta", {"setuptools": "84.0.0"}), } IMAGE_DOCKERFILES: Final = { "api": "backend/Dockerfile", "web": "frontend/Dockerfile", "node-agent": "node-agent/Dockerfile", "runtime-worker": "runtime-worker/Dockerfile", } PYTHON_BUILD_DOCKERFILES: Final = frozenset({"api", "node-agent", "runtime-worker"}) IMAGE_FIRST_PARTY: Final = { "api": {"modelforge-api"}, "node-agent": {"modelforge-api", "modelforge-node-agent"}, "runtime-worker": {"modelforge-runtime-worker"}, } IMAGE_TARGETS: Final = ("api", "web", "node-agent", "runtime-worker") PYTHON_IMAGE_TARGETS: Final = ("api", "node-agent", "runtime-worker") SECURITY_PROFILES: Final = ("full", "security") # pip-audit omits some packaging tools from resolved requirement reports. The source-only # exceptions remain hash-locked and checked by Trivy; shipping-image inventory must still # account for packaging when present. No other runtime dependency receives this exception. PIP_AUDIT_SOURCE_PACKAGING_TOOL_OMISSIONS: Final = frozenset( {"packaging", "pip", "setuptools", "wheel"} ) PIP_AUDIT_PACKAGING_TOOL_OMISSIONS: Final = frozenset({"setuptools", "wheel"}) IMAGE_FORBIDDEN_PACKAGES: Final = { "api": frozenset( { "editables", "hatchling", "pathspec", "pip", "pluggy", "setuptools", "tomlkit", "trove-classifiers", "wheel", } ), "runtime-worker": frozenset( { "editables", "hatchling", "pathspec", "pip", "pluggy", "setuptools", "tomlkit", "trove-classifiers", "wheel", } ), } # The digest-pinned PyTorch base exposes Conda's own distribution metadata, but Conda is not a # PyPI project and pip-audit correctly refuses to invent advisory coverage for it. Keep the exact # observed base version in the complete image inventory and let Trivy scan the immutable image; # exclude only this exact target/name/version tuple from the PyPI-specific requirements input. IMAGE_NON_PYPI_AUDIT_EXCEPTIONS: Final = { "runtime-worker": { "conda": "25.5.0", "conda-build": "25.5.0", "conda-libmamba-solver": "25.3.0", "libmambapy": "2.1.1", "menuinst": "2.2.0", "torch": "2.7.1+cu128", "torchaudio": "2.7.1+cu128", "torchvision": "0.22.1+cu128", }, } WEB_SECURITY_ORIGIN: Final = "https://security-scan.invalid" WEB_BUNDLE_PATH: Final = "/usr/share/nginx/html" WEB_PROVENANCE_PATH: Final = ".well-known/modelforge-build-provenance.json" WEB_PROVENANCE_FIELDS: Final = { "schema_version", "target", "api_base_url", "binding_sha256", } WEB_EXPECTED_NON_API_ORIGINS: Final = ( "http://www.w3.org", "https://git.example.com", "https://react.dev", ) TRIVY_DB_MAX_AGE: Final = timedelta(hours=48) TRIVY_DB_DOWNLOAD_MAX_AGE: Final = timedelta(hours=24) SOURCE_STATUS_CODES: Final = { "history-fetch": {0}, "history": {0}, "release-version": {0}, "pip-audit-download": {0}, "pip-audit-hash": {0}, "pip-audit-install": {0}, "pip-audit-version": {0}, "gitleaks-download": {0}, "gitleaks-hash": {0}, "gitleaks-extract": {0}, "gitleaks-version": {0}, "gitleaks-self-test": {0}, "trivy-download": {0}, "trivy-hash": {0}, "trivy-extract": {0}, "trivy-version": {0}, "source-inventory": {0}, "gitleaks-default": {0, 1}, "gitleaks-custom": {0, 1}, "gitleaks-merge": {0}, "gitleaks-cleanup": {0}, "gitleaks": {0, 1}, "gitleaks-sanitize": {0}, "pip-source-backend": {0, 1}, "pip-source-node-agent": {0, 1}, "pip-source-runtime-worker": {0, 1}, "npm-version": {0}, "npm-audit-all": {0, 1}, "npm-audit-production": {0, 1}, "trivy-fs": {0}, } IMAGE_STATUS_CODES: Final = { "docker": {0}, "image-build-api": {0}, "image-build-web": {0}, "image-build-node-agent": {0}, "image-build-runtime-worker": {0}, "image-identity-api": {0}, "image-identity-web": {0}, "image-identity-node-agent": {0}, "image-identity-runtime-worker": {0}, "web-bundle-origin": {0}, "image-inventory-api": {0}, "image-inventory-node-agent": {0}, "image-inventory-runtime-worker": {0}, "pip-image-api": {0, 1}, "pip-image-node-agent": {0, 1}, "pip-image-runtime-worker": {0, 1}, "trivy-image-api": {0}, "trivy-image-web": {0}, "trivy-image-node-agent": {0}, "trivy-image-runtime-worker": {0}, "image-cleanup": {0}, } SOURCE_EXPECTED_TARGETS: Final = { "history": ["full-history"], "gitleaks": ["full-history"], "python_source": sorted(SOURCE_COMPONENTS), "python_source_first_party": sorted(value[1] for value in SOURCE_COMPONENTS.values()), "npm": ["frontend:all", "frontend:production"], "trivy_filesystem": ["repository"], } IMAGE_EXPECTED_TARGETS: Final = { "image_build": sorted(IMAGE_TARGETS), "image_identity": sorted(IMAGE_TARGETS), "image_scan": sorted(IMAGE_TARGETS), "python_image_inventory": sorted(PYTHON_IMAGE_TARGETS), "python_image_audit": sorted(PYTHON_IMAGE_TARGETS), "web_bundle": ["web"], } RISK_FIELDS: Final = { "scanner", "target", "advisory_id", "package", "installed_version", "classification", "owner", "accepted_on", "expires_on", "tracking", "reason", } RISK_SCANNERS: Final = {"pip-audit", "npm-audit", "trivy"} GITLEAKS_FALSE_POSITIVE_FIELDS: Final = { "scanner", "rule_id", "fingerprint_sha256", "classification", "owner", "classified_on", "review_on", "tracking", "reason", } TIMEOUT_EXIT_CODES: Final = {124, 137, 143} EXACT_PIN_RE: Final = re.compile( r"^\s*([A-Za-z0-9][A-Za-z0-9._-]*)(?:\[[^\]]+\])?\s*==\s*([^\s;]+)(?:\s*;.*)?$" ) GITLEAKS_REPORT_FIELDS: Final = { "schema_version", "scanner", "target", "redaction_percent", "findings", } GITLEAKS_FINDING_FIELDS: Final = {"rule_id", "fingerprint_sha256"} GITLEAKS_RULE_ID_RE: Final = re.compile(r"[a-z0-9][a-z0-9._-]{0,127}") HISTORY_REPORT_FIELDS: Final = { "schema_version", "target", "head", "expected_commit", "is_shallow", "head_commit_count", "all_commit_count", "root_commits", "remote_name", "remote_url", "advertised_remote_heads", "local_remote_heads", "advertised_tags", "local_tags", "all_ref_count", "all_objects_reachable", } def expected_status_codes(profile: str) -> dict[str, set[int]]: if profile not in SECURITY_PROFILES: raise GateInputError(f"unsupported security profile: {profile}") values = dict(SOURCE_STATUS_CODES) if profile == "security": values.update(IMAGE_STATUS_CODES) return values def expected_targets(profile: str) -> dict[str, list[str]]: if profile not in SECURITY_PROFILES: raise GateInputError(f"unsupported security profile: {profile}") values = {name: list(targets) for name, targets in SOURCE_EXPECTED_TARGETS.items()} if profile == "security": values.update({name: list(targets) for name, targets in IMAGE_EXPECTED_TARGETS.items()}) return values def expected_evidence_paths(profile: str) -> set[str]: """Return the complete, profile-scoped evidence manifest before any bytes are read. The gate must not recursively consume arbitrary files placed beside managed reports. This manifest intentionally includes the retained scanner inputs because the evaluator verifies those exact inputs against the project manifests. """ paths = { "git-history.json", "source-inventory.json", "gitleaks.json", "npm-audit-all.json", "npm-audit-production.json", "trivy-fs.json", "versions/pip-audit.txt", "versions/gitleaks.txt", "versions/npm.txt", "versions/trivy.json", *(f"status/{name}.exit" for name in expected_status_codes(profile)), *(f"source-requirements/{target}.txt" for target in SOURCE_COMPONENTS), *(f"pip-source-{target}.json" for target in SOURCE_COMPONENTS), } if profile == "security": paths.update( { "web-bundle-origin.json", *(f"image-identity-{target}.json" for target in IMAGE_TARGETS), *(f"trivy-image-{target}.json" for target in IMAGE_TARGETS), *(f"image-inventory-{target}.json" for target in PYTHON_IMAGE_TARGETS), *(f"pip-image-{target}.json" for target in PYTHON_IMAGE_TARGETS), *(f"image-requirements/{target}.txt" for target in PYTHON_IMAGE_TARGETS), } ) return paths class GateInputError(ValueError): """An expected report or policy input is absent or structurally invalid.""" @dataclass(frozen=True) class Finding: scanner: str target: str advisory_id: str package: str installed_version: str severity: str fix_versions: tuple[str, ...] finding_type: str = "vulnerability" @property def identity(self) -> tuple[str, str, str, str, str]: return ( self.scanner, self.target, self.advisory_id, self.package, self.installed_version, ) def as_json(self, *, reason: str) -> dict[str, Any]: value = asdict(self) value["fix_versions"] = list(self.fix_versions) value["reason"] = reason return value @dataclass(frozen=True) class EvaluationRequest: reports_dir: Path accepted_risks: Path gitleaks_false_positives: Path repository: Path commit: str ref: str run_id: str source_repository: str profile: str output: Path as_of: date evaluation_time: datetime def _normalize_name(value: str) -> str: return re.sub(r"[-_.]+", "-", value).lower() def _matches_first_party_version(installed: str | None, source: str | None) -> bool: if installed is None or source is None: return False if installed == source: return True # Wheel metadata canonicalizes only the RC separator for these local packages. release = re.fullmatch(r"([0-9]+\.[0-9]+\.[0-9]+)-rc\.([0-9]+)", source) return release is not None and installed == f"{release.group(1)}rc{release.group(2)}" def _read_json(path: Path) -> Any: try: raw = path.read_bytes() except OSError as exc: raise GateInputError(f"missing report: {path}") from exc if not raw.strip(): raise GateInputError(f"empty report: {path}") try: return json.loads(raw, object_pairs_hook=_reject_duplicate_json_object_keys) except (UnicodeDecodeError, json.JSONDecodeError) as exc: raise GateInputError(f"malformed JSON report: {path}: {exc}") from exc def _read_json_bytes(raw: bytes, label: str) -> Any: if not raw.strip(): raise GateInputError(f"empty report: {label}") try: return json.loads(raw, object_pairs_hook=_reject_duplicate_json_object_keys) except (UnicodeDecodeError, json.JSONDecodeError) as exc: raise GateInputError(f"malformed JSON report: {label}: {exc}") from exc def _write_json(path: Path, value: Any) -> None: path.parent.mkdir(parents=True, exist_ok=True) # Replacing the directory entry does not follow an existing symlink or hard link. The gate # writes its final decision after inspecting evidence supplied by scanners, so following a # pre-created output path here would otherwise turn result persistence into a write primitive. descriptor, temporary_name = tempfile.mkstemp( prefix=f".{path.name}.", suffix=".tmp", dir=path.parent, text=True ) temporary = Path(temporary_name) try: with os.fdopen(descriptor, "w", encoding="utf-8", newline="\n") as handle: handle.write(json.dumps(value, indent=2, sort_keys=True) + "\n") handle.flush() os.fsync(handle.fileno()) os.replace(temporary, path) except BaseException: temporary.unlink(missing_ok=True) raise def _reject_duplicate_json_object_keys(pairs: list[tuple[str, Any]]) -> dict[str, Any]: """Decode JSON objects without letting a later duplicate silently replace evidence.""" result: dict[str, Any] = {} for key, value in pairs: if key in result: raise GateInputError(f"duplicate JSON object key: {key!r}") result[key] = value return result def _sha256(path: Path) -> str: try: metadata = os.lstat(path) except OSError as exc: raise GateInputError(f"cannot inspect file before hashing: {path}: {exc}") from exc if _is_link_or_reparse(path) or not stat.S_ISREG(metadata.st_mode): raise GateInputError(f"refusing to hash a link, reparse point, or non-file: {path}") digest = hashlib.sha256() with path.open("rb") as handle: for chunk in iter(lambda: handle.read(1024 * 1024), b""): digest.update(chunk) return digest.hexdigest() def _is_link_or_reparse(path: Path) -> bool: metadata = os.lstat(path) if stat.S_ISLNK(metadata.st_mode): return True reparse_flag = getattr(stat, "FILE_ATTRIBUTE_REPARSE_POINT", 0) return bool(getattr(metadata, "st_file_attributes", 0) & reparse_flag) def _metadata_identity(metadata: os.stat_result) -> tuple[int, int, int, int]: return ( metadata.st_dev, metadata.st_ino, stat.S_IFMT(metadata.st_mode), metadata.st_uid, ) def _metadata_stability(metadata: os.stat_result) -> tuple[int, int, int, int, int, int, int]: # NTFS exposes creation-time aliases through st_ctime and can round them differently between # an open handle and a later path lookup. Size+mtime+file identity remain stable and detect # in-place mutation there; POSIX ctime additionally detects metadata-only changes. change_time = metadata.st_ctime_ns if os.name == "posix" else 0 return ( *_metadata_identity(metadata), metadata.st_size, metadata.st_mtime_ns, change_time, ) def _metadata_is_reparse(metadata: os.stat_result) -> bool: reparse_flag = getattr(stat, "FILE_ATTRIBUTE_REPARSE_POINT", 0) return stat.S_ISLNK(metadata.st_mode) or bool( getattr(metadata, "st_file_attributes", 0) & reparse_flag ) def _validate_evidence_metadata( metadata: os.stat_result, *, path: str, expected_kind: str, root: bool = False, enforce_posix_permissions: bool | None = None, ) -> None: expected_type = stat.S_IFDIR if expected_kind == "directory" else stat.S_IFREG if _metadata_is_reparse(metadata) or stat.S_IFMT(metadata.st_mode) != expected_type: raise GateInputError(f"evidence {path} is not a real {expected_kind}") link_count = getattr(metadata, "st_nlink", 1) if expected_kind == "file" and link_count != 1: raise GateInputError(f"evidence {path} has {link_count} hard links") if enforce_posix_permissions is None: enforce_posix_permissions = os.name == "posix" if enforce_posix_permissions and hasattr(os, "geteuid") and metadata.st_uid != os.geteuid(): raise GateInputError(f"evidence {path} is not owned by the evaluator user") mode = stat.S_IMODE(metadata.st_mode) if enforce_posix_permissions and root and mode & 0o077: raise GateInputError(f"evidence root {path} permissions are broader than 0700") if enforce_posix_permissions and mode & 0o022: raise GateInputError(f"evidence {path} is group/world-writable ({mode:04o})") EvidenceRaceHook = Callable[[str, Path], None] class SecureEvidenceSnapshot: """Read and hash one immutable evidence view without following replaceable paths.""" def __init__( self, root: Path, output: Path, *, allowed_paths: set[str] | None = None, race_hook: EvidenceRaceHook | None = None, ) -> None: self.root = root.absolute() self.output_relative = self._relative(output.absolute()) self._allowed_paths = ( None if allowed_paths is None else {self._validate_allowed_path(path) for path in allowed_paths} ) self._race_hook = race_hook self._values: dict[str, bytes] = {} self._hashes: dict[str, str] = {} self._path_identities: dict[Path, tuple[int, int, int, int]] = {} self._path_stabilities: dict[Path, tuple[int, int, int, int, int, int, int]] = {} self._ancestry = self._capture_ancestry(self.root) if os.name == "posix" and os.open in os.supports_dir_fd: self._capture_posix() else: self._capture_path_based() self.revalidate() def _relative(self, path: Path) -> str: try: relative = path.relative_to(self.root).as_posix() except ValueError as exc: raise GateInputError(f"evidence path escapes its root: {path}") from exc if relative in {"", "."}: raise GateInputError("evaluator output cannot replace the evidence root") return relative @staticmethod def _validate_allowed_path(path: str) -> str: if not path or path.startswith("/") or "\\" in path: raise GateInputError(f"invalid allowed evidence path: {path!r}") parts = path.split("/") if any(part in {"", ".", ".."} for part in parts): raise GateInputError(f"invalid allowed evidence path: {path!r}") return path def _validate_allowed_entry(self, relative: str, *, directory: bool) -> None: if self._allowed_paths is None: return if directory: if any(path.startswith(f"{relative}/") for path in self._allowed_paths): return elif relative in self._allowed_paths: return raise GateInputError(f"unexpected evidence path: {relative}") def _record_output_entry(self, relative: str, path: Path, metadata: os.stat_result) -> None: """Validate the bootstrap result without consuming it as evidence. It is not report input and therefore is not hashed, but it is still an attacker-visible path later replaced by the evaluator. Track it so a link or substitution blocks before that replacement occurs. """ _validate_evidence_metadata(metadata, path=relative, expected_kind="file") self._path_identities[path] = _metadata_identity(metadata) self._path_stabilities[path] = _metadata_stability(metadata) @staticmethod def _capture_ancestry(path: Path) -> dict[Path, tuple[int, int, int, int]]: values: dict[Path, tuple[int, int, int, int]] = {} candidates = [path, *path.parents] for candidate in reversed(candidates): try: metadata = os.lstat(candidate) except OSError as exc: raise GateInputError(f"evidence ancestry is unavailable: {candidate}: {exc}") from exc if _metadata_is_reparse(metadata) or not stat.S_ISDIR(metadata.st_mode): raise GateInputError(f"evidence ancestry is not a real directory: {candidate}") values[candidate] = _metadata_identity(metadata) return values def _hook(self, event: str, path: Path) -> None: if self._race_hook is not None: self._race_hook(event, path) @staticmethod def _read_fd(file_descriptor: int) -> bytes: chunks: list[bytes] = [] while True: chunk = os.read(file_descriptor, 1024 * 1024) if not chunk: return b"".join(chunks) chunks.append(chunk) def _store(self, relative: str, value: bytes) -> None: if relative == self.output_relative: return self._values[relative] = value self._hashes[relative] = hashlib.sha256(value).hexdigest() def _capture_posix(self) -> None: flags = os.O_RDONLY | getattr(os, "O_CLOEXEC", 0) | getattr(os, "O_NOFOLLOW", 0) directory_flags = flags | getattr(os, "O_DIRECTORY", 0) root_before = os.lstat(self.root) _validate_evidence_metadata( root_before, path=str(self.root), expected_kind="directory", root=True ) self._hook("after_lstat", self.root) try: root_fd = os.open(self.root, directory_flags) except OSError as exc: raise GateInputError(f"cannot securely open evidence root: {exc}") from exc try: root_open = os.fstat(root_fd) _validate_evidence_metadata( root_open, path=str(self.root), expected_kind="directory", root=True ) if _metadata_identity(root_before) != _metadata_identity(root_open): raise GateInputError("evidence root changed between validation and open") self._path_identities[self.root] = _metadata_identity(root_open) self._path_stabilities[self.root] = _metadata_stability(root_open) self._walk_posix(root_fd, self.root, "", directory_flags, flags) if _metadata_identity(os.fstat(root_fd)) != _metadata_identity(root_open): raise GateInputError("opened evidence root identity changed during capture") finally: os.close(root_fd) def _walk_posix( self, directory_fd: int, directory_path: Path, relative_directory: str, directory_flags: int, file_flags: int, ) -> None: try: names = sorted(os.listdir(directory_fd)) except OSError as exc: raise GateInputError(f"cannot list evidence directory {directory_path}: {exc}") from exc for name in names: if name in {"", ".", ".."} or "/" in name: raise GateInputError(f"invalid evidence entry name: {name!r}") relative = f"{relative_directory}/{name}".lstrip("/") display_path = directory_path / name before = os.stat(name, dir_fd=directory_fd, follow_symlinks=False) if relative == self.output_relative: self._record_output_entry(relative, display_path, before) continue if stat.S_ISDIR(before.st_mode): self._validate_allowed_entry(relative, directory=True) _validate_evidence_metadata( before, path=relative, expected_kind="directory" ) self._hook("after_lstat", display_path) try: child_fd = os.open(name, directory_flags, dir_fd=directory_fd) except OSError as exc: raise GateInputError(f"cannot securely open evidence directory {relative}: {exc}") from exc try: opened = os.fstat(child_fd) _validate_evidence_metadata( opened, path=relative, expected_kind="directory" ) if _metadata_identity(before) != _metadata_identity(opened): raise GateInputError(f"evidence directory changed before open: {relative}") self._path_identities[display_path] = _metadata_identity(opened) self._path_stabilities[display_path] = _metadata_stability(opened) self._walk_posix( child_fd, display_path, relative, directory_flags, file_flags ) after_entry = os.stat(name, dir_fd=directory_fd, follow_symlinks=False) if _metadata_identity(after_entry) != _metadata_identity(opened): raise GateInputError(f"evidence directory changed after read: {relative}") finally: os.close(child_fd) continue self._validate_allowed_entry(relative, directory=False) _validate_evidence_metadata(before, path=relative, expected_kind="file") self._hook("after_lstat", display_path) try: file_fd = os.open(name, file_flags, dir_fd=directory_fd) except OSError as exc: raise GateInputError(f"cannot securely open evidence file {relative}: {exc}") from exc try: opened = os.fstat(file_fd) _validate_evidence_metadata(opened, path=relative, expected_kind="file") if _metadata_identity(before) != _metadata_identity(opened): raise GateInputError(f"evidence file changed before open: {relative}") self._hook("after_open", display_path) value = self._read_fd(file_fd) after_open = os.fstat(file_fd) after_entry = os.stat(name, dir_fd=directory_fd, follow_symlinks=False) if ( _metadata_stability(opened) != _metadata_stability(after_open) or _metadata_identity(opened) != _metadata_identity(after_entry) ): raise GateInputError(f"evidence file changed during read: {relative}") self._path_identities[display_path] = _metadata_identity(opened) self._path_stabilities[display_path] = _metadata_stability(opened) self._store(relative, value) finally: os.close(file_fd) def _capture_path_based(self) -> None: root_before = os.lstat(self.root) _validate_evidence_metadata( root_before, path=str(self.root), expected_kind="directory", root=True ) self._path_identities[self.root] = _metadata_identity(root_before) self._path_stabilities[self.root] = _metadata_stability(root_before) self._walk_path_based(self.root, "") def _walk_path_based(self, directory: Path, relative_directory: str) -> None: before_directory = os.lstat(directory) _validate_evidence_metadata( before_directory, path=relative_directory or str(self.root), expected_kind="directory", root=not relative_directory, ) try: names = sorted(entry.name for entry in os.scandir(directory)) except OSError as exc: raise GateInputError(f"cannot list evidence directory {directory}: {exc}") from exc for name in names: relative = f"{relative_directory}/{name}".lstrip("/") path = directory / name before = os.lstat(path) if relative == self.output_relative: self._record_output_entry(relative, path, before) continue if stat.S_ISDIR(before.st_mode) and not _metadata_is_reparse(before): self._validate_allowed_entry(relative, directory=True) _validate_evidence_metadata(before, path=relative, expected_kind="directory") self._hook("after_lstat", path) opened = os.lstat(path) if _metadata_identity(before) != _metadata_identity(opened): raise GateInputError(f"evidence directory changed before traversal: {relative}") self._path_identities[path] = _metadata_identity(opened) self._path_stabilities[path] = _metadata_stability(opened) self._walk_path_based(path, relative) if _metadata_identity(os.lstat(path)) != _metadata_identity(opened): raise GateInputError(f"evidence directory changed after traversal: {relative}") continue self._validate_allowed_entry(relative, directory=False) _validate_evidence_metadata(before, path=relative, expected_kind="file") self._hook("after_lstat", path) flags = os.O_RDONLY | getattr(os, "O_BINARY", 0) try: file_fd = os.open(path, flags) except OSError as exc: raise GateInputError(f"cannot securely open evidence file {relative}: {exc}") from exc try: opened = os.fstat(file_fd) _validate_evidence_metadata(opened, path=relative, expected_kind="file") if _metadata_identity(before) != _metadata_identity(opened): raise GateInputError(f"evidence file changed before open: {relative}") self._hook("after_open", path) value = self._read_fd(file_fd) after_open = os.fstat(file_fd) after_entry = os.lstat(path) if ( _metadata_stability(opened) != _metadata_stability(after_open) or _metadata_identity(opened) != _metadata_identity(after_entry) ): raise GateInputError(f"evidence file changed during read: {relative}") self._path_identities[path] = _metadata_identity(opened) self._path_stabilities[path] = _metadata_stability(opened) self._store(relative, value) finally: os.close(file_fd) if _metadata_identity(os.lstat(directory)) != _metadata_identity(before_directory): raise GateInputError(f"evidence directory changed while read: {directory}") def revalidate(self) -> None: for path, expected in self._ancestry.items(): metadata = os.lstat(path) if _metadata_is_reparse(metadata) or _metadata_identity(metadata) != expected: raise GateInputError(f"evidence ancestry identity changed: {path}") for path, expected in self._path_identities.items(): metadata = os.lstat(path) expected_kind = "directory" if expected[2] == stat.S_IFDIR else "file" _validate_evidence_metadata( metadata, path=str(path), expected_kind=expected_kind, root=path == self.root, ) if _metadata_identity(metadata) != expected: raise GateInputError(f"evidence entry identity changed: {path}") if _metadata_stability(metadata) != self._path_stabilities[path]: raise GateInputError(f"evidence entry changed after secure capture: {path}") def read_bytes(self, relative: str) -> bytes: self.revalidate() try: return self._values[relative] except KeyError as exc: raise GateInputError(f"missing report: {self.root / relative}") from exc def hashes(self) -> dict[str, str]: self.revalidate() return dict(sorted(self._hashes.items())) def _string(value: Any, field: str) -> str: if not isinstance(value, str) or not value.strip(): raise GateInputError(f"{field} must be a non-empty string") return value.strip() def _list(value: Any, field: str) -> list[Any]: if not isinstance(value, list): raise GateInputError(f"{field} must be a list") return value def _dict(value: Any, field: str) -> dict[str, Any]: if not isinstance(value, dict) or not all(isinstance(key, str) for key in value): raise GateInputError(f"{field} must be an object with string keys") return value def _utc_timestamp(value: Any, field: str) -> datetime: text = _string(value, field) normalized = text[:-1] + "+00:00" if text.endswith("Z") else text try: parsed = datetime.fromisoformat(normalized) except ValueError as exc: raise GateInputError(f"{field} is not an ISO-8601 timestamp") from exc if parsed.tzinfo is None or parsed.utcoffset() != timedelta(0): raise GateInputError(f"{field} must be an explicit UTC timestamp") return parsed.astimezone(UTC) def _canonical_repository_url(value: Any, field: str) -> str: text = _string(value, field) parsed = urlsplit(text) try: port = parsed.port except ValueError as exc: raise GateInputError(f"{field} has an invalid port") from exc if ( parsed.scheme != "https" or parsed.hostname is None or parsed.username is not None or parsed.password is not None or parsed.query or parsed.fragment or parsed.path in {"", "/"} or any(character.isspace() for character in text) ): raise GateInputError(f"{field} must be a credential-free canonical HTTPS repository URL") host = parsed.hostname.lower() host_text = f"[{host}]" if ":" in host else host path = parsed.path.rstrip("/") if path.endswith(".git"): path = path.removesuffix(".git") if path in {"", "/"}: raise GateInputError(f"{field} has no repository path") return f"https://{host_text}{f':{port}' if port is not None else ''}{path}" def _parse_cli_utc_timestamp(value: str) -> datetime: try: return _utc_timestamp(value, "evaluation time") except GateInputError as exc: raise argparse.ArgumentTypeError(str(exc)) from exc def _parse_requirement(requirement: str, field: str) -> tuple[str, str]: match = EXACT_PIN_RE.fullmatch(requirement) if match is None: raise GateInputError(f"{field} is not pinned with one exact == version: {requirement!r}") return _normalize_name(match.group(1)), match.group(2) def _manifest_requirement_lines( text: str, manifest: str ) -> dict[str, list[tuple[str, int]]]: """Locate dependency-array declarations without confusing unrelated TOML strings.""" result: dict[str, list[tuple[str, int]]] = {} literal_pattern = re.compile( r'''^\s*(?P"(?:\\.|[^"\\])*"|'[^']*')\s*,?\s*(?:#.*)?$''' ) section = "" active_source: str | None = None for line_number, line in enumerate(text.splitlines(), start=1): stripped = line.strip() section_match = re.fullmatch(r"\[([^]]+)]", stripped) if section_match is not None: section = section_match.group(1) active_source = None continue if active_source is None: assignment = re.fullmatch(r"([A-Za-z0-9_-]+)\s*=\s*\[\s*", stripped) if assignment is None: continue key = assignment.group(1) if section == "build-system" and key == "requires": active_source = "build-system.requires" elif section == "project" and key == "dependencies": active_source = "project.dependencies" elif section == "project.optional-dependencies": active_source = f"project.optional-dependencies.{key}" if active_source is not None: result[active_source] = [] continue if re.fullmatch(r"]\s*,?\s*(?:#.*)?", stripped): active_source = None continue if not stripped or stripped.startswith("#"): continue match = literal_pattern.fullmatch(line) if match is None: raise GateInputError( f"unsupported dependency declaration syntax at {manifest}:{line_number}" ) try: value = tomllib.loads(f"value = {match.group('literal')}")["value"] except (KeyError, tomllib.TOMLDecodeError) as exc: raise GateInputError( f"cannot parse dependency declaration at {manifest}:{line_number}: {exc}" ) from exc if not isinstance(value, str): raise GateInputError(f"dependency at {manifest}:{line_number} is not a string") result[active_source].append((value, line_number)) return result def _read_build_toolchain_lock(repository: Path) -> list[dict[str, Any]]: lock_path = repository / BUILD_TOOLCHAIN_LOCK try: text = lock_path.read_text(encoding="utf-8") except (OSError, UnicodeDecodeError) as exc: raise GateInputError(f"cannot read {BUILD_TOOLCHAIN_LOCK}: {exc}") from exc entries: list[dict[str, Any]] = [] seen: dict[str, str] = {} for line_number, line in enumerate(text.splitlines(), start=1): stripped = line.strip() if not stripped or stripped.startswith("#"): continue fields = stripped.split() name, version = _parse_requirement(fields[0], f"{BUILD_TOOLCHAIN_LOCK}:{line_number}") if name in seen: raise GateInputError(f"{BUILD_TOOLCHAIN_LOCK} contains duplicate package {name}") hashes = fields[1:] if not hashes or any( re.fullmatch(r"--hash=sha256:[0-9a-f]{64}", value) is None for value in hashes ): raise GateInputError( f"{BUILD_TOOLCHAIN_LOCK}:{line_number} must contain SHA-256 hashes only" ) if len(set(hashes)) != len(hashes): raise GateInputError(f"{BUILD_TOOLCHAIN_LOCK}:{line_number} repeats a hash") if version != BUILD_TOOLCHAIN_PINS.get(name): raise GateInputError( f"{BUILD_TOOLCHAIN_LOCK}:{line_number} pins {name} to {version!r}; " f"expected {BUILD_TOOLCHAIN_PINS.get(name)!r}" ) seen[name] = version entries.append( { "name": name, "version": version, "requirement": fields[0], "line": line_number, } ) if seen != BUILD_TOOLCHAIN_PINS: missing = sorted(set(BUILD_TOOLCHAIN_PINS) - set(seen)) unexpected = sorted(set(seen) - set(BUILD_TOOLCHAIN_PINS)) raise GateInputError( f"{BUILD_TOOLCHAIN_LOCK} package set differs from policy: " f"missing={missing}, unexpected={unexpected}" ) return entries def _validate_release_dockerfiles(repository: Path) -> None: dockerfiles: dict[str, str] = {} for target, relative_path in IMAGE_DOCKERFILES.items(): path = repository / relative_path try: text = path.read_text(encoding="utf-8") except (OSError, UnicodeDecodeError) as exc: raise GateInputError(f"cannot read {relative_path}: {exc}") from exc dockerfiles[target] = text expected_source = CANONICAL_SOURCE_REPOSITORY public_manifest_path = repository / PUBLIC_SOURCE_MANIFEST if public_manifest_path.exists(): # The curated exporter redacts private hostnames in its output. Accept its existing # example label only when the generated manifest binds all four transformed Dockerfiles. manifest = _dict(_read_json(public_manifest_path), PUBLIC_SOURCE_MANIFEST) if ( manifest.get("schemaVersion") != 1 or not isinstance(manifest.get("sourceRevision"), str) or re.fullmatch(r"[0-9a-f]{40}", manifest["sourceRevision"]) is None ): raise GateInputError(f"{PUBLIC_SOURCE_MANIFEST} has an invalid export identity") public_source = _string( manifest.get("sourceRepository"), f"{PUBLIC_SOURCE_MANIFEST}.sourceRepository", ) if public_source != PUBLIC_SOURCE_LABEL_PLACEHOLDER: raise GateInputError( f"{PUBLIC_SOURCE_MANIFEST}.sourceRepository is not the approved sanitized identity" ) file_entries = _list(manifest.get("files"), f"{PUBLIC_SOURCE_MANIFEST}.files") manifest_files: dict[str, dict[str, Any]] = {} for index, value in enumerate(file_entries): entry = _dict(value, f"{PUBLIC_SOURCE_MANIFEST}.files[{index}]") path_value = _string(entry.get("path"), f"{PUBLIC_SOURCE_MANIFEST}.files[{index}].path") if path_value in manifest_files: raise GateInputError(f"{PUBLIC_SOURCE_MANIFEST} repeats {path_value!r}") manifest_files[path_value] = entry for target, relative_path in IMAGE_DOCKERFILES.items(): entry = manifest_files.get(relative_path) if entry is None: raise GateInputError( f"{PUBLIC_SOURCE_MANIFEST} does not bind {relative_path}" ) dockerfile_bytes = dockerfiles[target].encode("utf-8") expected_digest = hashlib.sha256(dockerfile_bytes).hexdigest() if ( entry.get("bytes") != len(dockerfile_bytes) or entry.get("sha256") != expected_digest ): raise GateInputError( f"{PUBLIC_SOURCE_MANIFEST} does not bind the current {relative_path}" ) expected_source = public_source for target, text in dockerfiles.items(): labels = re.findall( r'(?m)^LABEL\s+org\.opencontainers\.image\.source="([^"]+)"\s*$', text ) if labels != [expected_source]: raise GateInputError( f"{IMAGE_DOCKERFILES[target]} must declare one source label bound to the " f"canonical source identity {expected_source!r}" ) for target in PYTHON_BUILD_DOCKERFILES: relative_path = IMAGE_DOCKERFILES[target] text = dockerfiles[target] flattened = re.sub(r"\\\r?\n\s*", " ", text) lock_copy = re.escape(f"COPY {BUILD_TOOLCHAIN_LOCK} {BUILD_TOOLCHAIN_CONTAINER_PATH}") if len(re.findall(rf"(?m)^{lock_copy}\s*$", text)) != 1: raise GateInputError( f"{relative_path} must copy {BUILD_TOOLCHAIN_LOCK} exactly once" ) run_statements = re.findall(r"(?m)^RUN\s+([^\r\n]+)", flattened) commands = [command.strip() for run in run_statements for command in run.split("&&")] installs = [command for command in commands if re.search(r"\bpip install\b", command)] bootstrap = [ command for command in installs if f"-r {BUILD_TOOLCHAIN_CONTAINER_PATH}" in command ] project = [ command for command in installs if f"-r {BUILD_TOOLCHAIN_CONTAINER_PATH}" not in command ] required_bootstrap = ( "--require-hashes", "--only-binary=:all:", "--index-url https://pypi.org/simple", f"-r {BUILD_TOOLCHAIN_CONTAINER_PATH}", ) missing = [value for value in required_bootstrap if not bootstrap or value not in bootstrap[0]] if len(installs) != 2 or len(bootstrap) != 1 or len(project) != 1 or missing: raise GateInputError( f"{relative_path} must use one hash-locked bootstrap install and one project install; " f"missing bootstrap options={missing}" ) if "--no-build-isolation" not in project[0] or "--upgrade" in flattened: raise GateInputError( f"{relative_path} project install must disable build isolation after the locked bootstrap" ) if target == "node-agent" and "--ignore-installed" not in project[0]: raise GateInputError( f"{relative_path} target install must not satisfy runtime dependencies from build tooling" ) def build_source_inventory(repository: Path) -> dict[str, Any]: """Read project dependencies, PEP 517 requirements and the hashed build toolchain.""" _validate_release_dockerfiles(repository) build_toolchain = _read_build_toolchain_lock(repository) component_values: list[dict[str, Any]] = [] project_versions: dict[str, tuple[str, str]] = {} parsed_projects: dict[str, tuple[Path, dict[str, Any], str]] = {} for target, (relative_manifest, expected_name) in SOURCE_COMPONENTS.items(): manifest = repository / relative_manifest try: manifest_text = manifest.read_text(encoding="utf-8") document = tomllib.loads(manifest_text) except (OSError, UnicodeDecodeError, tomllib.TOMLDecodeError) as exc: raise GateInputError(f"cannot parse {relative_manifest}: {exc}") from exc project = _dict(document.get("project"), f"{relative_manifest} [project]") name = _normalize_name(_string(project.get("name"), f"{relative_manifest} project.name")) version = _string(project.get("version"), f"{relative_manifest} project.version") if name != expected_name: raise GateInputError( f"{relative_manifest} project name is {name!r}, expected {expected_name!r}" ) project_versions[name] = (target, version) parsed_projects[target] = (manifest, project, manifest_text) for target, (manifest, project, manifest_text) in parsed_projects.items(): relative_manifest = manifest.relative_to(repository).as_posix() project_name = _normalize_name(str(project["name"])) project_version = str(project["version"]) collected: dict[tuple[str, str], dict[str, Any]] = {} declaration_lines = _manifest_requirement_lines(manifest_text, relative_manifest) declarations_seen: dict[str, tuple[str, int]] = {} build_system = _dict( tomllib.loads(manifest_text).get("build-system"), f"{relative_manifest} [build-system]", ) expected_backend, expected_build_requirements = PEP517_BUILD_SYSTEMS[target] backend = _string( build_system.get("build-backend"), f"{relative_manifest} build-system.build-backend" ) if backend != expected_backend: raise GateInputError( f"{relative_manifest} build backend is {backend!r}; expected {expected_backend!r}" ) build_requirements = _list( build_system.get("requires"), f"{relative_manifest} build-system.requires" ) actual_build_requirements: dict[str, str] = {} for index, requirement_value in enumerate(build_requirements): requirement = _string( requirement_value, f"{relative_manifest} build-system.requires[{index}]" ) name, version = _parse_requirement( requirement, f"{relative_manifest} build-system.requires[{index}]" ) if name in actual_build_requirements: raise GateInputError( f"{relative_manifest} build-system repeats build requirement {name}" ) actual_build_requirements[name] = version if actual_build_requirements != expected_build_requirements: raise GateInputError( f"{relative_manifest} PEP 517 requirements differ from policy: " f"expected={expected_build_requirements}, actual={actual_build_requirements}" ) lock_versions = {entry["name"]: entry["version"] for entry in build_toolchain} if any(lock_versions.get(name) != version for name, version in actual_build_requirements.items()): raise GateInputError( f"{relative_manifest} PEP 517 requirements do not match the hashed toolchain lock" ) def add( requirement: str, group: str, *, source: str, own_project: bool = False, declaration_line: int | None = None, ) -> None: name, version = _parse_requirement(requirement, f"{relative_manifest}:{group}") line_number: int | None = None if not own_project: if declaration_line is not None: line_number = declaration_line else: source_match = re.fullmatch(r"(.+)\[([0-9]+)]", source) if source_match is None: raise GateInputError(f"invalid manifest source identity: {source}") source_key = source_match.group(1) source_index = int(source_match.group(2)) records = declaration_lines.get(source_key, []) if source_index >= len(records) or records[source_index][0] != requirement: raise GateInputError( f"cannot locate {source} declaration {requirement!r} in " f"{relative_manifest}" ) line_number = records[source_index][1] if not source.startswith("build-system."): prior = declarations_seen.get(requirement) if prior is not None: prior_source, prior_line = prior raise GateInputError( f"identical manifest declaration {requirement!r} is duplicated: " f"{relative_manifest}:{prior_line} ({prior_source}) and " f"{relative_manifest}:{line_number} ({source})" ) declarations_seen[requirement] = (source, line_number) key = (name, version) conflicting = [item for item in collected if item[0] == name and item[1] != version] if conflicting: raise GateInputError( f"{relative_manifest} pins {name} to conflicting versions " f"{conflicting[0][1]!r} and {version!r}" ) record = collected.setdefault( key, { "name": name, "version": version, "requirements": [], "declarations": [], "groups": [], "first_party": name in project_versions, "own_project": own_project, }, ) record["requirements"].append(requirement) if line_number is not None: record["declarations"].append( { "requirement": requirement, "source": source, "line": line_number, } ) record["groups"].append(group) record["own_project"] = bool(record["own_project"] or own_project) add( f"{project_name}=={project_version}", "first-party-source", source="project.name/project.version", own_project=True, ) for index, requirement_value in enumerate(build_requirements): requirement = _string( requirement_value, f"{relative_manifest} build-system.requires[{index}]" ) add( requirement, "build-system", source=f"build-system.requires[{index}]", ) for entry in build_toolchain: add( str(entry["requirement"]), "build-toolchain", source=BUILD_TOOLCHAIN_LOCK, declaration_line=int(entry["line"]), ) for index, requirement in enumerate(_list(project.get("dependencies", []), "dependencies")): add( _string(requirement, f"{relative_manifest} dependencies[{index}]"), "runtime", source=f"project.dependencies[{index}]", ) optional = _dict(project.get("optional-dependencies", {}), "optional-dependencies") for group in sorted(optional): for index, requirement in enumerate(_list(optional[group], f"optional group {group}")): add( _string(requirement, f"{relative_manifest} {group}[{index}]"), f"optional:{group}", source=f"project.optional-dependencies.{group}[{index}]", ) requirements = [] for record in sorted(collected.values(), key=lambda item: (item["name"], item["version"])): record["groups"] = sorted(set(record["groups"])) record["requirements"] = sorted(set(record["requirements"])) record["declarations"] = sorted( record["declarations"], key=lambda item: (item["line"], item["source"]) ) requirements.append(record) component_values.append( { "target": target, "manifest": relative_manifest, "project_name": project_name, "project_version": project_version, "optional_groups": sorted(optional), "requirements": requirements, } ) first_party_coverage: list[dict[str, str]] = [] for component in component_values: for requirement in component["requirements"]: if not requirement["first_party"]: continue provider_target, provider_version = project_versions[requirement["name"]] first_party_coverage.append( { "consumer_target": component["target"], "package": requirement["name"], "required_version": requirement["version"], "provider_target": provider_target, "provider_version": provider_version, "coverage": "provider-source-and-shipping-image-audited", } ) return { "schema_version": SCHEMA_VERSION, "components": sorted(component_values, key=lambda item: item["target"]), "first_party_coverage": sorted( first_party_coverage, key=lambda item: (item["consumer_target"], item["package"]), ), } def collect_source_inventory(repository: Path, output: Path, requirements_directory: Path) -> None: inventory = build_source_inventory(repository) _write_json(output, inventory) requirements_directory.mkdir(mode=0o700, parents=False, exist_ok=False) for component in inventory["components"]: # Feed pip-audit the exact PEP 508 declarations. Reconstructing only name==version silently # drops extras such as uvicorn[standard] and psycopg[binary], so their dependencies would # never be resolved or audited. lines = sorted( { declaration for requirement in component["requirements"] if not requirement["first_party"] for declaration in requirement["requirements"] } ) target = str(component["target"]) (requirements_directory / f"{target}.txt").write_text( "\n".join(lines) + "\n", encoding="utf-8" ) def sanitize_gitleaks(input_path: Path, output: Path) -> None: """Reduce a raw Gitleaks report to fields that cannot contain matched secret material.""" raw = _read_json(input_path) findings = raw if isinstance(raw, list) else _dict(raw, "Gitleaks report").get("findings") finding_list = _list(findings, "Gitleaks findings") sanitized: list[dict[str, Any]] = [] for index, value in enumerate(finding_list): finding = _dict(value, f"Gitleaks findings[{index}]") rule_id = _string(finding.get("RuleID"), f"Gitleaks findings[{index}].RuleID") if GITLEAKS_RULE_ID_RE.fullmatch(rule_id) is None: raise GateInputError(f"Gitleaks findings[{index}].RuleID is not safe rule metadata") fingerprint = _string( finding.get("Fingerprint"), f"Gitleaks findings[{index}].Fingerprint" ) sanitized.append( { "rule_id": rule_id, "fingerprint_sha256": hashlib.sha256( fingerprint.encode("utf-8") ).hexdigest(), } ) _write_json( output, { "schema_version": SCHEMA_VERSION, "scanner": "gitleaks", "target": "full-history", "redaction_percent": 100, "findings": sanitized, }, ) def normalize_image_inventory( target: str, input_path: Path, output: Path, requirements_output: Path ) -> None: if target not in PYTHON_IMAGE_TARGETS: raise GateInputError(f"unexpected Python image target: {target}") raw = _read_json(input_path) document = _dict(raw, "image inventory") packages = _list(document.get("packages"), "image inventory packages") normalized: dict[str, str] = {} for index, value in enumerate(packages): package = _dict(value, f"packages[{index}]") name = _normalize_name(_string(package.get("name"), f"packages[{index}].name")) version = _string(package.get("version"), f"packages[{index}].version") previous = normalized.get(name) if previous is not None: if previous != version: raise GateInputError(f"image {target} contains conflicting {name} versions") raise GateInputError(f"image {target} contains duplicate {name} package rows") normalized[name] = version if not normalized: raise GateInputError(f"image {target} inventory is empty") non_pypi = IMAGE_NON_PYPI_AUDIT_EXCEPTIONS.get(target, {}) mismatched_non_pypi = sorted( name for name, version in non_pypi.items() if normalized.get(name) != version ) if mismatched_non_pypi: raise GateInputError( f"image {target} non-PyPI audit exceptions are absent or version-drifted: " f"{mismatched_non_pypi}" ) output_value = { "schema_version": SCHEMA_VERSION, "target": target, "python_version": _string(document.get("python_version"), "python_version"), "packages": [ {"name": name, "version": version} for name, version in sorted(normalized.items()) ], } _write_json(output, output_value) requirements_output.parent.mkdir(parents=True, exist_ok=True) first_party_names = {project_name for _, project_name in SOURCE_COMPONENTS.values()} requirements_output.write_text( "".join( f"{name}=={version}\n" for name, version in sorted(normalized.items()) if name not in first_party_names and name not in non_pypi ), encoding="utf-8", ) def record_image_identity( target: str, tag: str, expected_revision: str, expected_source: str, expected_version: str, output: Path, ) -> None: """Record only immutable, non-secret identity fields from Docker inspection output.""" if target not in IMAGE_TARGETS: raise GateInputError(f"unexpected image target: {target}") if _canonical_repository_url(expected_source, "expected image source") != expected_source: raise GateInputError("expected image source is not canonical") expected_tag = f"modelforge-security-{target}:{expected_revision}" if tag != expected_tag: raise GateInputError(f"image tag is {tag!r}; expected {expected_tag!r}") result = subprocess.run( # noqa: S603 ["docker", "image", "inspect", tag], check=False, capture_output=True, text=True, ) if result.returncode != 0: raise GateInputError(f"docker image inspect failed ({result.returncode})") try: documents = json.loads(result.stdout, object_pairs_hook=_reject_duplicate_json_object_keys) except (json.JSONDecodeError, GateInputError) as exc: raise GateInputError("docker image inspect returned malformed JSON") from exc values = _list(documents, "docker image inspect") if len(values) != 1: raise GateInputError("docker image inspect did not return exactly one image") document = _dict(values[0], "docker image identity") image_id = _string(document.get("Id"), "docker image Id") if re.fullmatch(r"sha256:[0-9a-f]{64}", image_id) is None: raise GateInputError("docker image Id is not an immutable SHA-256 identity") repo_tags_value = document.get("RepoTags") or [] repo_digests_value = document.get("RepoDigests") or [] repo_tags = [_string(value, "docker RepoTag") for value in _list(repo_tags_value, "RepoTags")] repo_digests = [ _string(value, "docker RepoDigest") for value in _list(repo_digests_value, "RepoDigests") ] if repo_tags != [tag]: raise GateInputError( "docker image identity must contain exactly its one approved unique commit tag" ) if repo_digests: raise GateInputError("local release image unexpectedly has repository digest aliases") config = _dict(document.get("Config"), "docker image Config") labels = _dict(config.get("Labels"), "docker image labels") revision = _string( labels.get("org.opencontainers.image.revision"), "org.opencontainers.image.revision", ) source = _string( labels.get("org.opencontainers.image.source"), "org.opencontainers.image.source", ) version = _string( labels.get("org.opencontainers.image.version"), "org.opencontainers.image.version", ) if revision != expected_revision: raise GateInputError( f"image revision label is {revision!r}; expected {expected_revision!r}" ) if source != expected_source: raise GateInputError(f"image source label is {source!r}; expected {expected_source!r}") if version != expected_version: raise GateInputError(f"image version label is {version!r}; expected {expected_version!r}") bundle_label = labels.get("io.modelforge.security.bundle-path") bundle_path: str | None if target == "web": bundle_path = _string(bundle_label, "io.modelforge.security.bundle-path") if bundle_path != WEB_BUNDLE_PATH: raise GateInputError(f"Web bundle path is not canonical: {bundle_path!r}") else: if bundle_label is not None: raise GateInputError(f"non-Web image {target} unexpectedly has a bundle path") bundle_path = None _write_json( output, { "schema_version": SCHEMA_VERSION, "target": target, "tag": tag, "image_id": image_id, "repo_tags": repo_tags, "repo_digests": repo_digests, "revision": revision, "source": source, "version": version, "bundle_path": bundle_path, }, ) def bind_trivy_image_report( input_path: Path, output: Path, image_id: str, expected_tag: str ) -> None: """Retain Trivy component context while binding every result to the scanned immutable image.""" if re.fullmatch(r"sha256:[0-9a-f]{64}", image_id) is None: raise GateInputError("Trivy binding image ID is not an immutable SHA-256 identity") if re.fullmatch( r"modelforge-security-(?:api|web|node-agent|runtime-worker):[0-9a-f]{40}", expected_tag, ) is None: raise GateInputError("Trivy binding tag is not an exact security image identity") document = _dict(_read_json(input_path), "raw Trivy image report") if "ScannedImageID" in document: raise GateInputError("raw Trivy report unexpectedly contains gate-owned binding data") if ( document.get("SchemaVersion") != 2 or document.get("ArtifactType") != "container_image" or document.get("ArtifactName") != image_id ): raise GateInputError("raw Trivy report is not bound to the exact scanned image") metadata = _dict(document.get("Metadata"), "raw Trivy image Metadata") if metadata.get("ImageID") != image_id or metadata.get("Reference") not in { None, image_id, expected_tag, }: raise GateInputError("raw Trivy metadata is not bound to the exact scanned image") artifact_id = document.get("ArtifactID") if artifact_id is not None and re.fullmatch(r"sha256:[0-9a-f]{64}", str(artifact_id)) is None: raise GateInputError("raw Trivy ArtifactID is not an immutable SHA-256 identity") results = _list(document.get("Results"), "raw Trivy Results") normalized_results: list[dict[str, Any]] = [] for index, value in enumerate(results): result = _dict(value, f"raw Trivy Results[{index}]") if "ComponentTarget" in result: raise GateInputError("raw Trivy result unexpectedly contains gate-owned binding data") component_target = _string(result.get("Target"), f"raw Trivy Results[{index}].Target") normalized_results.append( {**result, "Target": image_id, "ComponentTarget": component_target} ) _write_json( output, {**document, "ScannedImageID": image_id, "Results": normalized_results}, ) def _extract_http_origins(content: bytes) -> set[str]: """Conservatively find direct or string-assembled HTTP/WebSocket origins.""" normalized = ( content.replace(b"\\/", b"/") .replace(b"\\u002f", b"/") .replace(b"\\u002F", b"/") .replace(b"\\x2f", b"/") .replace(b"\\x2F", b"/") ) # A second representation closes the common minified/source-map bypasses where a scheme, # separator, or hostname is assembled from quoted/template fragments. collapsed = re.sub(rb'''["'`\s+${}()]''', b"", normalized) pattern = rb"(?:https?|wss?)://(?:\[[0-9A-Fa-f:.]+\]|[A-Za-z0-9._~-]+)(?::[0-9]{1,5})?" matches = re.findall(pattern, normalized, flags=re.IGNORECASE) matches.extend(re.findall(pattern, collapsed, flags=re.IGNORECASE)) origins: set[str] = set() for raw in matches: value = raw.decode("ascii") parsed = urlsplit(value) try: port = parsed.port except ValueError as exc: raise GateInputError(f"bundle contains an invalid HTTP origin: {value!r}") from exc if parsed.scheme not in {"http", "https", "ws", "wss"} or parsed.hostname is None: raise GateInputError(f"bundle contains an invalid network origin: {value!r}") host = parsed.hostname.rstrip(".").lower() try: address = ipaddress.ip_address(host) except ValueError: address = None if ( host == "localhost" or host.startswith("localhost.") or host.endswith(".localhost") or (address is not None and address.is_loopback) ): raise GateInputError(f"Web bundle contains a loopback API origin: {value!r}") host_text = f"[{host}]" if ":" in host else host origins.add(f"{parsed.scheme}://{host_text}{f':{port}' if port is not None else ''}") return origins def _validate_non_api_origin_contexts(content: bytes) -> None: """Allow only pinned framework namespace/error-document URL contexts, never arbitrary use.""" normalized = ( content.replace(b"\\/", b"/") .replace(b"\\u002f", b"/") .replace(b"\\u002F", b"/") .replace(b"\\x2f", b"/") .replace(b"\\x2F", b"/") ) collapsed = re.sub(rb'''["'`\s+${}()]''', b"", normalized) allowed_suffixes = { b"http://www.w3.org": ( b"/1998/Math/MathML", b"/1999/xhtml", b"/1999/xlink", b"/2000/svg", b"/XML/1998/namespace", ), b"https://react.dev": (b"/errors/",), b"https://git.example.com": ( b"/Jens/ITWorx-ModelForge/src/branch/master/docs/", ), } for representation in (normalized, collapsed): for origin, suffixes in allowed_suffixes.items(): start = 0 while (position := representation.find(origin, start)) >= 0: tail = representation[position + len(origin) :] if not any(tail.startswith(suffix) for suffix in suffixes): raise GateInputError( f"Web bundle uses expected non-API origin outside its pinned context: " f"{origin.decode('ascii')}" ) start = position + len(origin) def _web_provenance_binding(api_base_url: str) -> str: binding = json.dumps( { "api_base_url": api_base_url, "schema_version": SCHEMA_VERSION, "target": "modelforge-web", }, sort_keys=True, separators=(",", ":"), ) return hashlib.sha256(binding.encode("utf-8")).hexdigest() def _bundle_files(root: Path) -> list[Path]: root_resolved = root.resolve(strict=True) files: list[Path] = [] for current, directory_names, file_names in os.walk(root, topdown=True, followlinks=False): current_path = Path(current) for name in directory_names: path = current_path / name metadata = os.lstat(path) if _is_link_or_reparse(path) or not stat.S_ISDIR(metadata.st_mode): raise GateInputError(f"Web bundle directory is unsafe: {path}") for name in file_names: path = current_path / name metadata = os.lstat(path) if _is_link_or_reparse(path) or not stat.S_ISREG(metadata.st_mode): raise GateInputError(f"Web bundle file is unsafe: {path}") try: path.resolve(strict=True).relative_to(root_resolved) except ValueError as exc: raise GateInputError(f"Web bundle file escapes its root: {path}") from exc files.append(path) return sorted(files, key=lambda value: value.relative_to(root).as_posix()) def record_web_bundle_evidence( image_id: str, bundle_path: str, expected_origin: str, output: Path ) -> None: """Prove the exact Web image embeds only the intentional security-scan API origin.""" if re.fullmatch(r"sha256:[0-9a-f]{64}", image_id) is None: raise GateInputError("Web image identity is not an immutable SHA-256") if expected_origin != WEB_SECURITY_ORIGIN: raise GateInputError(f"unexpected Web security origin: {expected_origin!r}") if bundle_path != WEB_BUNDLE_PATH: raise GateInputError(f"unexpected Web bundle path: {bundle_path!r}") create = subprocess.run( # noqa: S603 ["docker", "create", image_id], check=False, capture_output=True, text=True, ) if create.returncode != 0: raise GateInputError(f"docker create failed ({create.returncode})") container_id = create.stdout.strip() if re.fullmatch(r"[0-9a-f]{12,64}", container_id) is None: raise GateInputError("docker create returned an invalid container identity") evidence: dict[str, Any] | None = None primary_error: GateInputError | None = None try: with tempfile.TemporaryDirectory(prefix="modelforge-web-bundle-") as temporary: bundle_root = Path(temporary) / "bundle" bundle_root.mkdir(mode=0o700) copied = subprocess.run( # noqa: S603 ["docker", "cp", f"{container_id}:{bundle_path}/.", str(bundle_root)], check=False, capture_output=True, text=True, ) if copied.returncode != 0: raise GateInputError(f"docker cp of Web bundle failed ({copied.returncode})") started = subprocess.run( # noqa: S603 ["docker", "start", container_id], check=False, capture_output=True, text=True, timeout=30, ) if started.returncode != 0: raise GateInputError(f"docker start failed ({started.returncode})") served = subprocess.run( # noqa: S603 [ "docker", "exec", container_id, "sh", "-ec", "i=0; until wget -q -O - " "http://127.0.0.1:3000/.well-known/modelforge-build-provenance.json; " "do i=$((i+1)); [ \"${i}\" -lt 20 ] || exit 1; sleep 1; done", ], check=False, capture_output=True, timeout=30, ) if served.returncode != 0: raise GateInputError( f"querying served Web provenance failed ({served.returncode})" ) evidence = _analyze_web_bundle( bundle_root, image_id=image_id, bundle_path=bundle_path, expected_origin=expected_origin, served_provenance=served.stdout, ) except (GateInputError, OSError, subprocess.SubprocessError) as exc: primary_error = GateInputError(str(exc)) removed = subprocess.run( # noqa: S603 ["docker", "rm", "--force", container_id], check=False, capture_output=True, text=True, ) if removed.returncode != 0: raise GateInputError(f"docker container cleanup failed ({removed.returncode})") if primary_error is not None: raise primary_error if evidence is None: raise GateInputError("Web bundle evidence was not produced") _write_json(output, evidence) def _analyze_web_bundle( bundle_root: Path, *, image_id: str, bundle_path: str, expected_origin: str, served_provenance: bytes, ) -> dict[str, Any]: """Bind the served provenance asset and its one effective runtime constant to a bundle.""" files = _bundle_files(bundle_root) if not files: raise GateInputError("Web bundle is empty") by_relative = {path.relative_to(bundle_root).as_posix(): path for path in files} provenance_path = by_relative.get(WEB_PROVENANCE_PATH) if provenance_path is None: raise GateInputError("Web bundle has no canonical build provenance asset") provenance_bytes = provenance_path.read_bytes() if served_provenance != provenance_bytes: raise GateInputError("served Web provenance differs from the hashed bundle asset") provenance = _read_json_bytes(provenance_bytes, WEB_PROVENANCE_PATH) if not isinstance(provenance, dict) or set(provenance) != WEB_PROVENANCE_FIELDS: raise GateInputError("Web build provenance does not have the exact version-1 schema") api_base_url = _string(provenance.get("api_base_url"), "Web provenance api_base_url") binding = _string(provenance.get("binding_sha256"), "Web provenance binding_sha256") if ( provenance.get("schema_version") != SCHEMA_VERSION or provenance.get("target") != "modelforge-web" or api_base_url != expected_origin or binding != _web_provenance_binding(api_base_url) or re.fullmatch(r"[0-9a-f]{64}", binding) is None ): raise GateInputError("Web build provenance is malformed or not bound to the expected origin") bundle_digest = hashlib.sha256() origins: set[str] = set() runtime_candidates: list[tuple[str, str]] = [] for relative, path in sorted(by_relative.items()): content = path.read_bytes() file_hash = hashlib.sha256(content).hexdigest() bundle_digest.update(relative.encode("utf-8") + b"\0" + bytes.fromhex(file_hash)) if path.suffix.lower() in {".js", ".mjs", ".cjs", ".map", ".html", ".css"}: _validate_non_api_origin_contexts(content) origins.update(_extract_http_origins(content)) if ( path.suffix.lower() in {".js", ".mjs", ".cjs"} and api_base_url.encode("utf-8") in content and binding.encode("ascii") in content ): runtime_candidates.append((relative, file_hash)) expected_origins = {expected_origin, *WEB_EXPECTED_NON_API_ORIGINS} if origins != expected_origins: raise GateInputError( f"Web bundle network origins are not exact: expected={sorted(expected_origins)!r}, " f"actual={sorted(origins)!r}" ) if len(runtime_candidates) != 1: raise GateInputError( "Web bundle must have exactly one runtime chunk bound to the provenance constant" ) runtime_path, runtime_hash = runtime_candidates[0] return { "schema_version": SCHEMA_VERSION, "target": "web", "image_id": image_id, "bundle_path": bundle_path, "bundle_sha256": bundle_digest.hexdigest(), "bundle_file_count": len(files), "provenance_path": WEB_PROVENANCE_PATH, "provenance_sha256": hashlib.sha256(provenance_bytes).hexdigest(), "served_provenance_sha256": hashlib.sha256(served_provenance).hexdigest(), "provenance_binding_sha256": binding, "runtime_chunk_path": runtime_path, "runtime_chunk_sha256": runtime_hash, "api_origins": [expected_origin], "non_api_origins": list(WEB_EXPECTED_NON_API_ORIGINS), } def record_history(repository: Path, expected_commit: str, output: Path) -> None: def git(*arguments: str) -> str: result = subprocess.run( # noqa: S603 ["git", "-C", str(repository), *arguments], check=False, capture_output=True, text=True, ) if result.returncode != 0: raise GateInputError( f"git {' '.join(arguments)} failed ({result.returncode}): {result.stderr.strip()}" ) return result.stdout.strip() def refs(*prefixes: str) -> dict[str, str]: lines = git("for-each-ref", "--format=%(refname)%09%(objectname)", *prefixes) result: dict[str, str] = {} for line in lines.splitlines(): try: ref, object_id = line.split("\t", maxsplit=1) except ValueError as exc: raise GateInputError(f"malformed local ref record: {line!r}") from exc if re.fullmatch(r"[0-9a-f]{40}", object_id) is None: raise GateInputError(f"malformed object id for {ref}: {object_id!r}") result[ref] = object_id return result def advertised_refs() -> tuple[dict[str, str], dict[str, str]]: lines = git("ls-remote", "--refs", "--heads", "--tags", "origin") heads: dict[str, str] = {} tags: dict[str, str] = {} for line in lines.splitlines(): try: object_id, ref = line.split("\t", maxsplit=1) except ValueError as exc: raise GateInputError(f"malformed advertised ref record: {line!r}") from exc if re.fullmatch(r"[0-9a-f]{40}", object_id) is None: raise GateInputError(f"malformed advertised object id for {ref}") if ref.startswith("refs/heads/"): heads[ref] = object_id elif ref.startswith("refs/tags/"): tags[ref] = object_id return heads, tags head = git("rev-parse", "HEAD") shallow_text = git("rev-parse", "--is-shallow-repository") head_commit_count_text = git("rev-list", "--count", "HEAD") all_commit_count_text = git("rev-list", "--count", "--all") roots = [line for line in git("rev-list", "--max-parents=0", "--all").splitlines() if line] remote_url = git("remote", "get-url", "origin") advertised_heads, advertised_tags = advertised_refs() local_remote_raw = refs("refs/remotes/origin") local_heads = { "refs/heads/" + ref.removeprefix("refs/remotes/origin/"): object_id for ref, object_id in local_remote_raw.items() if ref != "refs/remotes/origin/HEAD" } local_tags = refs("refs/tags") if local_heads != advertised_heads: raise GateInputError("local remote-tracking heads do not match all advertised remote heads") if local_tags != advertised_tags: raise GateInputError("local tags do not match all advertised remote tags") if shallow_text != "false": raise GateInputError("repository is shallow") if head != expected_commit: raise GateInputError("repository HEAD does not equal the expected commit") try: head_commit_count = int(head_commit_count_text) all_commit_count = int(all_commit_count_text) except ValueError as exc: raise GateInputError("invalid git commit count") from exc all_refs = refs("refs") object_ids = {head, *advertised_heads.values(), *advertised_tags.values()} for object_id in object_ids: git("cat-file", "-e", object_id) def ref_values(values: dict[str, str]) -> list[dict[str, str]]: return [ {"ref": ref, "object": object_id} for ref, object_id in sorted(values.items()) ] _write_json( output, { "schema_version": SCHEMA_VERSION, "target": "full-history", "head": head, "expected_commit": expected_commit, "is_shallow": False, "head_commit_count": head_commit_count, "all_commit_count": all_commit_count, "root_commits": roots, "remote_name": "origin", "remote_url": remote_url, "advertised_remote_heads": ref_values(advertised_heads), "local_remote_heads": ref_values(local_heads), "advertised_tags": ref_values(advertised_tags), "local_tags": ref_values(local_tags), "all_ref_count": len(all_refs), "all_objects_reachable": True, }, ) class Evaluator: def __init__(self, request: EvaluationRequest) -> None: self.request = request if request.profile not in SECURITY_PROFILES: raise GateInputError(f"unsupported security profile: {request.profile}") if ( _canonical_repository_url(request.source_repository, "source_repository") != request.source_repository ): raise GateInputError("source_repository must be one canonical HTTPS repository URL") if ( request.evaluation_time.tzinfo is None or request.evaluation_time.utcoffset() != timedelta(0) ): raise GateInputError("evaluation_time must be an explicit UTC timestamp") self.expected_status_codes = expected_status_codes(request.profile) self.expected_targets = expected_targets(request.profile) self.policy_blocks: list[dict[str, Any]] = [] self.findings: list[Finding] = [] self.actual_targets: dict[str, set[str]] = { category: set() for category in self.expected_targets } self.exit_codes: dict[str, int | None] = {} self.scanner_versions: dict[str, str | None] = { "pip-audit": None, "gitleaks": None, "trivy": None, "npm": None, } self.database_times: dict[str, Any] = {"trivy": None} self.gitleaks_finding_identities: list[tuple[str, str]] = [] self.resolved_transitives: dict[str, list[dict[str, str]]] = {} self.image_identities: dict[str, dict[str, Any]] = {} try: self.evidence_snapshot: SecureEvidenceSnapshot | None = SecureEvidenceSnapshot( request.reports_dir, request.output, allowed_paths=expected_evidence_paths(request.profile), ) self.evidence_tree_error: str | None = None except (GateInputError, OSError) as exc: self.evidence_snapshot = None self.evidence_tree_error = str(exc) def _evidence_relative(self, path: Path) -> str: try: return path.absolute().relative_to(self.request.reports_dir.absolute()).as_posix() except ValueError as exc: raise GateInputError(f"evidence path escapes its root: {path}") from exc def read_evidence_bytes(self, path: Path) -> bytes: if self.evidence_snapshot is None: raise GateInputError(self.evidence_tree_error or "secure evidence snapshot is absent") return self.evidence_snapshot.read_bytes(self._evidence_relative(path)) def read_evidence_text(self, path: Path, *, encoding: str) -> str: try: return self.read_evidence_bytes(path).decode(encoding) except UnicodeDecodeError as exc: raise GateInputError(f"evidence text has invalid {encoding}: {path}") from exc def read_evidence_json(self, path: Path) -> Any: return _read_json_bytes(self.read_evidence_bytes(path), str(path)) def block(self, target: str, advisory_id: str, reason: str) -> None: self.policy_blocks.append( Finding( scanner="security-gate", target=target, advisory_id=advisory_id, package="", installed_version="", severity="ERROR", fix_versions=(), finding_type="policy", ).as_json(reason=reason) ) def guarded(self, target: str, advisory_id: str, operation: Any) -> Any | None: try: return operation() except (GateInputError, OSError, ValueError, TypeError, KeyError) as exc: self.block(target, advisory_id, str(exc)) return None def evaluate_statuses(self) -> None: status_dir = self.request.reports_dir / "status" for name, allowed in self.expected_status_codes.items(): path = status_dir / f"{name}.exit" try: text = self.read_evidence_text(path, encoding="ascii").strip() if not re.fullmatch(r"-?[0-9]+", text): raise GateInputError(f"malformed exit-code report: {path}") code = int(text) except (OSError, UnicodeDecodeError, GateInputError) as exc: self.exit_codes[name] = None self.block(name, "GATE-STATUS-MISSING", str(exc)) continue self.exit_codes[name] = code if code not in allowed: category = "timeout" if code in TIMEOUT_EXIT_CODES else "scanner-or-network-failure" self.block( name, "GATE-STATUS-FAILED", f"{name} returned {code} ({category}); allowed={sorted(allowed)}", ) if self.request.profile == "security": for target in IMAGE_TARGETS: if self.exit_codes.get(f"image-build-{target}") == 0: self.actual_targets["image_build"].add(target) def evaluate_history(self) -> None: path = self.request.reports_dir / "git-history.json" document = self.guarded( "full-history", "GATE-HISTORY-REPORT", lambda: self.read_evidence_json(path) ) if document is None: return report = self.guarded("full-history", "GATE-HISTORY-SCHEMA", lambda: _dict(document, "history")) if report is None: return errors: list[str] = [] if set(report) != HISTORY_REPORT_FIELDS or report.get("schema_version") != SCHEMA_VERSION: errors.append("history report does not have the exact version-1 schema") if report.get("target") != "full-history": errors.append("history target is not full-history") if report.get("is_shallow") is not False: errors.append("repository is shallow") if report.get("head") != self.request.commit: errors.append("history HEAD does not equal the evaluated commit") if report.get("expected_commit") != self.request.commit: errors.append("history expected_commit does not equal the evaluated commit") head_count = report.get("head_commit_count") all_count = report.get("all_commit_count") roots = report.get("root_commits") if not isinstance(head_count, int) or head_count <= 0: errors.append("history head_commit_count is not positive") if not isinstance(all_count, int) or all_count < (head_count if isinstance(head_count, int) else 1): errors.append("history all_commit_count does not cover HEAD history") if not isinstance(roots, list) or not roots or not all( isinstance(root, str) and re.fullmatch(r"[0-9a-f]{40}", root) for root in roots ): errors.append("history root commit proof is absent or malformed") if report.get("remote_name") != "origin" or not isinstance(report.get("remote_url"), str): errors.append("history origin metadata is absent") else: try: remote_url = _canonical_repository_url(report["remote_url"], "history origin URL") if remote_url != self.request.source_repository: errors.append("history origin URL is not the evaluated source repository") except GateInputError as exc: errors.append(str(exc)) def validate_ref_records( field: str, expected_prefix: str ) -> list[dict[str, str]] | None: value = report.get(field) if not isinstance(value, list): errors.append(f"history {field} is not a list") return None previous = "" for item in value: if ( not isinstance(item, dict) or set(item) != {"ref", "object"} or not isinstance(item.get("ref"), str) or not isinstance(item.get("object"), str) or re.fullmatch(r"[0-9a-f]{40}", item["object"]) is None or not item["ref"].startswith(expected_prefix) or item["ref"] == expected_prefix or item["ref"] <= previous ): errors.append(f"history {field} contains malformed or unsorted refs") return None previous = item["ref"] return value advertised_heads = validate_ref_records("advertised_remote_heads", "refs/heads/") local_heads = validate_ref_records("local_remote_heads", "refs/heads/") advertised_tags = validate_ref_records("advertised_tags", "refs/tags/") local_tags = validate_ref_records("local_tags", "refs/tags/") if advertised_heads == []: errors.append("history has no advertised remote heads") if advertised_heads is not None and local_heads is not None and advertised_heads != local_heads: errors.append("local remote heads differ from advertised heads") if advertised_tags is not None and local_tags is not None and advertised_tags != local_tags: errors.append("local tags differ from advertised tags") if not isinstance(report.get("all_ref_count"), int) or report["all_ref_count"] < ( len(local_heads or []) + len(local_tags or []) ): errors.append("history all_ref_count is incomplete") if report.get("all_objects_reachable") is not True: errors.append("history object reachability proof is absent") if errors: for error in errors: self.block("full-history", "GATE-HISTORY-INCOMPLETE", error) return self.actual_targets["history"].add("full-history") def evaluate_source_inventory(self) -> dict[str, dict[str, str]]: path = self.request.reports_dir / "source-inventory.json" actual = self.guarded( "python-source", "GATE-SOURCE-INVENTORY", lambda: self.read_evidence_json(path) ) expected = self.guarded( "python-source", "GATE-SOURCE-MANIFESTS", lambda: build_source_inventory(self.request.repository), ) expected_packages: dict[str, dict[str, str]] = {} if expected is not None: for component in expected["components"]: expected_packages[component["target"]] = { item["name"]: item["version"] for item in component["requirements"] if not item["first_party"] } if actual is None or expected is None: return expected_packages if actual != expected: self.block( "python-source", "GATE-SOURCE-INVENTORY-MISMATCH", "source inventory does not exactly match all project dependencies and optional groups", ) return expected_packages for component in expected["components"]: expected_lines = sorted( { declaration for requirement in component["requirements"] if not requirement["first_party"] for declaration in requirement["requirements"] } ) requirements_path = ( self.request.reports_dir / "source-requirements" / f"{component['target']}.txt" ) try: actual_lines = self.read_evidence_text( requirements_path, encoding="utf-8" ).splitlines() except (OSError, UnicodeDecodeError) as exc: self.block( str(component["target"]), "GATE-SOURCE-REQUIREMENTS", f"cannot read retained pip input {requirements_path}: {exc}", ) continue if actual_lines != expected_lines: self.block( str(component["target"]), "GATE-SOURCE-REQUIREMENTS", "retained pip input does not exactly preserve all declared PEP 508 requirements", ) continue self.actual_targets["python_source"].add(component["target"]) own_names = { requirement["package"] for requirement in expected["first_party_coverage"] if requirement["consumer_target"] == requirement["provider_target"] } self.actual_targets["python_source_first_party"].update(own_names) return expected_packages def evaluate_pip_report( self, path: Path, target: str, expected_packages: dict[str, str], category: str, *, allow_resolved_transitives: bool, ) -> None: document = self.guarded( target, "GATE-PIP-REPORT", lambda: self.read_evidence_json(path) ) if document is None: return report = self.guarded(target, "GATE-PIP-SCHEMA", lambda: _dict(document, "pip-audit report")) if report is None: return dependencies_value = report.get("dependencies") try: dependencies = _list(dependencies_value, "pip-audit dependencies") except GateInputError as exc: self.block(target, "GATE-PIP-SCHEMA", str(exc)) return if not dependencies: self.block(target, "GATE-PIP-EMPTY", "pip-audit dependencies are empty") return seen: dict[str, str] = {} vulnerability_count = 0 for index, value in enumerate(dependencies): try: dependency = _dict(value, f"pip dependency {index}") name = _normalize_name(_string(dependency.get("name"), "pip dependency name")) version = _string(dependency.get("version"), "pip dependency version") except GateInputError as exc: self.block(target, "GATE-PIP-SCHEMA", str(exc)) continue if name in seen: qualifier = "conflicting" if seen[name] != version else "identical" self.block( target, "GATE-PIP-DUPLICATE", f"{qualifier} duplicate pip row for {name}", ) continue seen[name] = version skip_reason = dependency.get("skip_reason") if skip_reason is not None and name not in { project_name for _, project_name in SOURCE_COMPONENTS.values() }: self.block( target, "GATE-PIP-UNEXPECTED-SKIP", f"pip-audit skipped non-first-party package {name}: {skip_reason}", ) vulns = dependency.get("vulns", []) if not isinstance(vulns, list): self.block(target, "GATE-PIP-SCHEMA", f"vulns for {name} is not a list") continue for vulnerability in vulns: try: item = _dict(vulnerability, f"pip vulnerability for {name}") advisory_id = _string(item.get("id"), "pip vulnerability id") fixes_value = item.get("fix_versions", []) fixes = tuple( _string(fix, "pip fix version") for fix in _list(fixes_value, "fix_versions") ) except GateInputError as exc: self.block(target, "GATE-PIP-SCHEMA", str(exc)) continue vulnerability_count += 1 self.findings.append( Finding( scanner="pip-audit", target=target, advisory_id=advisory_id, package=name, installed_version=version, severity="UNKNOWN", fix_versions=fixes, ) ) missing_all = set(expected_packages) - set(seen) packaging_omissions = ( PIP_AUDIT_SOURCE_PACKAGING_TOOL_OMISSIONS if category == "python_source" else PIP_AUDIT_PACKAGING_TOOL_OMISSIONS ) missing = sorted(missing_all - packaging_omissions) unexpected = sorted(set(seen) - set(expected_packages)) changed = sorted( name for name in set(seen) & set(expected_packages) if seen[name] != expected_packages[name] ) if missing or changed or (unexpected and not allow_resolved_transitives): self.block( target, "GATE-PIP-COMPONENT-GAP", f"pip inventory mismatch: missing={missing}, unexpected={unexpected}, changed={changed}", ) return if allow_resolved_transitives: self.resolved_transitives[target] = [ {"name": name, "version": seen[name]} for name in unexpected ] status_name = f"pip-source-{target}" if category == "python_source" else f"pip-image-{target}" expected_exit = 1 if vulnerability_count else 0 if self.exit_codes.get(status_name) != expected_exit: self.block( target, "GATE-PIP-EXIT-MISMATCH", f"pip-audit exit {self.exit_codes.get(status_name)!r} disagrees with " f"{vulnerability_count} findings", ) self.actual_targets[category].add(target) def evaluate_gitleaks(self) -> None: path = self.request.reports_dir / "gitleaks.json" document = self.guarded( "full-history", "GATE-GITLEAKS-REPORT", lambda: self.read_evidence_json(path) ) if document is None: return report = self.guarded( "full-history", "GATE-GITLEAKS-SCHEMA", lambda: _dict(document, "Gitleaks report") ) if report is None: return if set(report) != GITLEAKS_REPORT_FIELDS or report.get("schema_version") != SCHEMA_VERSION: self.block( "full-history", "GATE-GITLEAKS-SCHEMA", "sanitized Gitleaks report does not have the exact version-1 schema", ) return if report.get("scanner") != "gitleaks" or report.get("target") != "full-history": self.block("full-history", "GATE-GITLEAKS-TARGET", "invalid Gitleaks target metadata") return if report.get("redaction_percent") != 100: self.block("full-history", "GATE-GITLEAKS-REDACTION", "Gitleaks redaction is not 100%") findings = report.get("findings") if not isinstance(findings, list): self.block("full-history", "GATE-GITLEAKS-SCHEMA", "findings is not a list") return for index, finding in enumerate(findings): if ( not isinstance(finding, dict) or set(finding) != GITLEAKS_FINDING_FIELDS or not isinstance(finding.get("rule_id"), str) or GITLEAKS_RULE_ID_RE.fullmatch(finding["rule_id"]) is None or not isinstance(finding.get("fingerprint_sha256"), str) or re.fullmatch(r"[0-9a-f]{64}", finding["fingerprint_sha256"]) is None ): self.block( "full-history", "GATE-GITLEAKS-SCHEMA", f"finding {index} lacks an exact rule/fingerprint hash", ) continue identity = (finding["rule_id"], finding["fingerprint_sha256"]) self.gitleaks_finding_identities.append(identity) self.findings.append( Finding( scanner="gitleaks", target="full-history", advisory_id=finding["rule_id"], package=finding["fingerprint_sha256"], installed_version="", severity="CRITICAL", fix_versions=(), finding_type="secret", ) ) expected_exit = 1 if findings else 0 if self.exit_codes.get("gitleaks") != expected_exit: self.block( "full-history", "GATE-GITLEAKS-EXIT-MISMATCH", f"Gitleaks exit {self.exit_codes.get('gitleaks')!r} disagrees with findings", ) self.actual_targets["gitleaks"].add("full-history") def evaluate_gitleaks_false_positives( self, ) -> tuple[list[dict[str, Any]], list[dict[str, Any]]]: """Apply exact, review-bounded FALSE_POSITIVE classifications after sanitization.""" document = self.guarded( "gitleaks-false-positives", "GATE-GITLEAKS-FP-REPORT", lambda: _read_json(self.request.gitleaks_false_positives), ) if document is None: return [], [] if not isinstance(document, list): self.block( "gitleaks-false-positives", "GATE-GITLEAKS-FP-SCHEMA", "Gitleaks false positives must be a JSON list", ) return [], [] valid: list[dict[str, Any]] = [] invalid_or_unused: list[dict[str, Any]] = [] identities: set[tuple[str, str]] = set() for index, value in enumerate(document): try: entry = _dict(value, f"Gitleaks false positive {index}") if set(entry) != GITLEAKS_FALSE_POSITIVE_FIELDS: raise GateInputError( f"Gitleaks false positive {index} fields differ: " f"expected={sorted(GITLEAKS_FALSE_POSITIVE_FIELDS)}, actual={sorted(entry)}" ) if entry["scanner"] != "gitleaks" or entry["classification"] != "FALSE_POSITIVE": raise GateInputError( f"Gitleaks false positive {index} has an invalid scanner/classification" ) rule_id = _string(entry["rule_id"], f"Gitleaks false positive {index}.rule_id") fingerprint = _string( entry["fingerprint_sha256"], f"Gitleaks false positive {index}.fingerprint_sha256", ) if re.fullmatch(r"[0-9a-f]{64}", fingerprint) is None: raise GateInputError( f"Gitleaks false positive {index} fingerprint is not an exact SHA-256" ) if re.search(r"[*?\[\]]", rule_id): raise GateInputError( f"Gitleaks false positive {index} rule contains a wildcard" ) identity = (rule_id, fingerprint) if identity in identities: raise GateInputError( f"Gitleaks false positive {index} duplicates an exact fingerprint" ) identities.add(identity) _string(entry["owner"], f"Gitleaks false positive {index}.owner") _string(entry["tracking"], f"Gitleaks false positive {index}.tracking") _string(entry["reason"], f"Gitleaks false positive {index}.reason") classified_on = date.fromisoformat( _string( entry["classified_on"], f"Gitleaks false positive {index}.classified_on", ) ) review_on = date.fromisoformat( _string(entry["review_on"], f"Gitleaks false positive {index}.review_on") ) duration = (review_on - classified_on).days if duration < 0 or duration > 30: raise GateInputError( f"Gitleaks false positive {index} review window is {duration} days; " "maximum is 30" ) if classified_on > self.request.as_of: raise GateInputError( f"Gitleaks false positive {index} is not active yet" ) if review_on < self.request.as_of: raise GateInputError(f"Gitleaks false positive {index} review is stale") valid.append({**entry, "_identity": identity, "_index": index}) except (GateInputError, ValueError) as exc: invalid_or_unused.append({"index": index, "entry": value, "reason": str(exc)}) self.block("gitleaks-false-positives", "GATE-GITLEAKS-FP-INVALID", str(exc)) applied: list[dict[str, Any]] = [] accepted_identities: set[tuple[str, str]] = set() for entry in valid: matches = [ identity for identity in self.gitleaks_finding_identities if identity == entry["_identity"] ] public_entry = {key: value for key, value in entry.items() if not key.startswith("_")} if not matches: invalid_or_unused.append( {"index": entry["_index"], "entry": public_entry, "reason": "unused or drifted"} ) self.block( "gitleaks-false-positives", "GATE-GITLEAKS-FP-UNUSED", f"Gitleaks false positive {entry['_index']} matches no exact fingerprint", ) continue if len(matches) != 1: invalid_or_unused.append( {"index": entry["_index"], "entry": public_entry, "reason": "ambiguous"} ) self.block( "gitleaks-false-positives", "GATE-GITLEAKS-FP-AMBIGUOUS", f"Gitleaks false positive {entry['_index']} matches {len(matches)} findings", ) continue accepted_identities.add(entry["_identity"]) applied.append({"entry": public_entry, "fingerprint_match_count": 1}) self.findings = [ finding for finding in self.findings if not ( finding.scanner == "gitleaks" and (finding.advisory_id, finding.package) in accepted_identities ) ] return applied, invalid_or_unused def _npm_lock_versions(self) -> tuple[dict[str, set[str]], dict[str, str]]: lock = _dict( _read_json(self.request.repository / "frontend" / "package-lock.json"), "package-lock.json", ) packages = _dict(lock.get("packages"), "package-lock packages") by_name: dict[str, set[str]] = {} by_path: dict[str, str] = {} for path, value in packages.items(): marker = "node_modules/" if marker not in path: continue name = path.rsplit(marker, maxsplit=1)[-1] if not name: continue package = _dict(value, f"package-lock {path}") version = package.get("version") if isinstance(version, str) and version: by_name.setdefault(name, set()).add(version) by_path[path] = version return by_name, by_path def evaluate_npm(self, filename: str, target: str, status_name: str) -> None: path = self.request.reports_dir / filename document = self.guarded( target, "GATE-NPM-REPORT", lambda: self.read_evidence_json(path) ) if document is None: return report = self.guarded(target, "GATE-NPM-SCHEMA", lambda: _dict(document, "npm audit")) if report is None: return if "error" in report: self.block(target, "GATE-NPM-NETWORK", "npm audit returned an error document") return if not isinstance(report.get("auditReportVersion"), int): self.block(target, "GATE-NPM-SCHEMA", "npm auditReportVersion is absent") return vulnerabilities = report.get("vulnerabilities") metadata = report.get("metadata") if not isinstance(vulnerabilities, dict) or not isinstance(metadata, dict): self.block(target, "GATE-NPM-SCHEMA", "npm vulnerability metadata is absent") return lock_inventory = self.guarded(target, "GATE-NPM-LOCK", self._npm_lock_versions) if lock_inventory is None: return lock_versions, lock_paths = lock_inventory count = 0 for package_name, raw_vulnerability in vulnerabilities.items(): try: vulnerability = _dict(raw_vulnerability, f"npm vulnerability {package_name}") severity = _string(vulnerability.get("severity"), "npm severity").upper() via = _list(vulnerability.get("via", []), "npm via") except GateInputError as exc: self.block(target, "GATE-NPM-SCHEMA", str(exc)) continue nodes_value = vulnerability.get("nodes", []) node_versions: set[str] = set() if isinstance(nodes_value, list): for node in nodes_value: if isinstance(node, str) and node in lock_paths: node_versions.add(lock_paths[node]) versions = node_versions or lock_versions.get(package_name, set()) if not versions: versions = {"unknown"} self.block( target, "GATE-NPM-INVENTORY", f"npm finding {package_name} has no exact installed version in package-lock", ) advisory_ids: list[str] = [] for source in via: if isinstance(source, dict): url = source.get("url") source_id = source.get("source") if isinstance(url, str) and url.rstrip("/").rsplit("/", maxsplit=1)[-1]: advisory_ids.append(url.rstrip("/").rsplit("/", maxsplit=1)[-1]) elif isinstance(source_id, (str, int)): advisory_ids.append(str(source_id)) if not advisory_ids: advisory_ids = [f"NPM-{package_name}"] fix_available = vulnerability.get("fixAvailable") is not False for advisory_id in sorted(set(advisory_ids)): for version in sorted(versions): count += 1 self.findings.append( Finding( scanner="npm-audit", target=target, advisory_id=advisory_id, package=package_name, installed_version=version, severity=severity, fix_versions=("available",) if fix_available else (), ) ) expected_exit = 1 if vulnerabilities else 0 if self.exit_codes.get(status_name) != expected_exit: self.block( target, "GATE-NPM-EXIT-MISMATCH", f"npm exit {self.exit_codes.get(status_name)!r} disagrees with {count} findings", ) self.actual_targets["npm"].add(target) def _validate_filesystem_result_target(self, value: Any, index: int) -> str: target = _string(value, f"Trivy result {index}.Target") repository = self.request.repository.resolve() candidate = Path(target) resolved = candidate.resolve() if candidate.is_absolute() else (repository / candidate).resolve() try: resolved.relative_to(repository) except ValueError as exc: raise GateInputError( f"Trivy filesystem result {index} escapes the scanned repository: {target!r}" ) from exc return target def evaluate_trivy(self, filename: str, target: str, category: str) -> None: path = self.request.reports_dir / filename document = self.guarded( target, "GATE-TRIVY-REPORT", lambda: self.read_evidence_json(path) ) if document is None: return report = self.guarded(target, "GATE-TRIVY-SCHEMA", lambda: _dict(document, "Trivy report")) if report is None: return if report.get("SchemaVersion") != 2: self.block(target, "GATE-TRIVY-SCHEMA", "Trivy SchemaVersion must be exactly 2") return artifact_type = report.get("ArtifactType") artifact_name = report.get("ArtifactName") if category == "trivy_filesystem": expected_name = str(self.request.repository.resolve()) if artifact_type != "repository" or artifact_name != expected_name: self.block( target, "GATE-TRIVY-TARGET", "filesystem report is not bound to the exact repository path: " f"ArtifactType={artifact_type!r}, ArtifactName={artifact_name!r}, " f"expected={expected_name!r}", ) return else: identity = self.image_identities.get(target) if identity is None: self.block(target, "GATE-TRIVY-TARGET", "validated image identity is absent") return image_id = identity["image_id"] if artifact_type != "container_image" or artifact_name != image_id: self.block( target, "GATE-TRIVY-TARGET", "image report ArtifactType/ArtifactName does not match the scanned image ID", ) return if report.get("ScannedImageID") != image_id: self.block( target, "GATE-TRIVY-TARGET", "gate-owned Trivy scan binding does not match the recorded image ID", ) return try: metadata = _dict(report.get("Metadata"), "Trivy image Metadata") metadata_image_id = _string(metadata.get("ImageID"), "Trivy Metadata.ImageID") repo_tags = [ _string(value, "Trivy Metadata.RepoTag") for value in _list(metadata.get("RepoTags") or [], "Trivy Metadata.RepoTags") ] repo_digests = [ _string(value, "Trivy Metadata.RepoDigest") for value in _list( metadata.get("RepoDigests") or [], "Trivy Metadata.RepoDigests" ) ] except GateInputError as exc: self.block(target, "GATE-TRIVY-SCHEMA", str(exc)) return if metadata_image_id != image_id: self.block( target, "GATE-TRIVY-TARGET", "Trivy Metadata.ImageID does not match the recorded image identity", ) return if repo_tags != identity["repo_tags"] or repo_digests != identity["repo_digests"]: self.block( target, "GATE-TRIVY-TARGET", "Trivy image repository metadata does not match Docker identity evidence", ) return reference = metadata.get("Reference") if reference not in {None, image_id, identity["tag"]}: self.block( target, "GATE-TRIVY-TARGET", "Trivy Metadata.Reference does not equal the exact scanned image ID or tag", ) return artifact_id = report.get("ArtifactID") if artifact_id is not None and re.fullmatch( r"sha256:[0-9a-f]{64}", str(artifact_id) ) is None: self.block( target, "GATE-TRIVY-SCHEMA", "Trivy ArtifactID is not an immutable SHA-256 identity", ) return results = report.get("Results") if not isinstance(results, list): self.block(target, "GATE-TRIVY-SCHEMA", "Trivy Results is not a list") return for result_index, raw_result in enumerate(results): if not isinstance(raw_result, dict): self.block(target, "GATE-TRIVY-SCHEMA", f"Trivy result {result_index} is invalid") continue try: result_target = _string( raw_result.get("Target"), f"Trivy result {result_index}.Target" ) except GateInputError as exc: self.block(target, "GATE-TRIVY-SCHEMA", str(exc)) continue try: if category == "trivy_filesystem": result_target = self._validate_filesystem_result_target( result_target, result_index ) component_target = result_target else: image_id = str(self.image_identities[target]["image_id"]) if result_target != image_id: raise GateInputError( f"Trivy image result {result_index}.Target does not equal " "the exact scanned image ID" ) component_target = _string( raw_result.get("ComponentTarget"), f"Trivy result {result_index}.ComponentTarget", ) except GateInputError as exc: self.block(target, "GATE-TRIVY-TARGET", str(exc)) continue vulnerabilities = raw_result.get("Vulnerabilities", []) or [] misconfigurations = raw_result.get("Misconfigurations", []) or [] if not isinstance(vulnerabilities, list) or not isinstance(misconfigurations, list): self.block(target, "GATE-TRIVY-SCHEMA", "Trivy finding lists are malformed") continue for raw_finding in vulnerabilities: if not isinstance(raw_finding, dict): self.block(target, "GATE-TRIVY-SCHEMA", "Trivy vulnerability is malformed") continue try: advisory_id = _string( raw_finding.get("VulnerabilityID"), "Trivy VulnerabilityID" ) package = _string(raw_finding.get("PkgName"), "Trivy PkgName") installed_version = _string( raw_finding.get("InstalledVersion"), "Trivy InstalledVersion" ) severity = _string(raw_finding.get("Severity"), "Trivy Severity").upper() if severity not in {"UNKNOWN", "LOW", "MEDIUM", "HIGH", "CRITICAL"}: raise GateInputError(f"Trivy Severity is invalid: {severity!r}") fixed_value = raw_finding.get("FixedVersion", "") if not isinstance(fixed_value, str): raise GateInputError("Trivy FixedVersion must be a string when present") except GateInputError as exc: self.block(target, "GATE-TRIVY-SCHEMA", str(exc)) continue if severity not in {"UNKNOWN", "HIGH", "CRITICAL"}: continue fixed = fixed_value.strip() fix_versions = tuple(part.strip() for part in fixed.split(",") if part.strip()) self.findings.append( Finding( scanner="trivy", target=target, advisory_id=advisory_id, package=package, installed_version=installed_version, severity=severity, fix_versions=fix_versions, ) ) for raw_finding in misconfigurations: if not isinstance(raw_finding, dict): self.block(target, "GATE-TRIVY-SCHEMA", "Trivy misconfiguration is malformed") continue try: advisory_id = _string(raw_finding.get("ID"), "Trivy misconfiguration ID") severity = _string( raw_finding.get("Severity"), "Trivy misconfiguration Severity" ).upper() if severity not in {"UNKNOWN", "LOW", "MEDIUM", "HIGH", "CRITICAL"}: raise GateInputError( f"Trivy misconfiguration Severity is invalid: {severity!r}" ) except GateInputError as exc: self.block(target, "GATE-TRIVY-SCHEMA", str(exc)) continue if severity not in {"UNKNOWN", "HIGH", "CRITICAL"}: continue self.findings.append( Finding( scanner="trivy", target=target, advisory_id=advisory_id, package=component_target, installed_version="not-applicable", severity=severity, fix_versions=(), finding_type="misconfiguration", ) ) self.actual_targets[category].add(target) def evaluate_image_identity(self, target: str) -> dict[str, Any] | None: path = self.request.reports_dir / f"image-identity-{target}.json" document = self.guarded( target, "GATE-IMAGE-IDENTITY", lambda: self.read_evidence_json(path) ) if document is None: return None try: report = _dict(document, "image identity") if set(report) != { "schema_version", "target", "tag", "image_id", "repo_tags", "repo_digests", "revision", "source", "version", "bundle_path", }: raise GateInputError(f"image identity {target} fields are not exact") if report.get("schema_version") != SCHEMA_VERSION or report.get("target") != target: raise GateInputError(f"image identity metadata mismatch for {target}") expected_tag = f"modelforge-security-{target}:{self.request.commit}" tag = _string(report.get("tag"), "image identity tag") image_id = _string(report.get("image_id"), "image identity image_id") revision = _string(report.get("revision"), "image identity revision") source = _string(report.get("source"), "image identity source") version = _string(report.get("version"), "image identity version") repo_tags = [ _string(value, "image identity RepoTag") for value in _list(report.get("repo_tags"), "image identity repo_tags") ] repo_digests = [ _string(value, "image identity RepoDigest") for value in _list(report.get("repo_digests"), "image identity repo_digests") ] if tag != expected_tag or repo_tags != [expected_tag]: raise GateInputError( f"image identity {target} must have exactly one approved commit tag" ) if repo_digests: raise GateInputError(f"image identity {target} has unapproved digest aliases") if re.fullmatch(r"sha256:[0-9a-f]{64}", image_id) is None: raise GateInputError(f"image identity {target} has no immutable image ID") if revision != self.request.commit: raise GateInputError(f"image identity {target} revision label mismatch") if source != self.request.source_repository: raise GateInputError(f"image identity {target} source label mismatch") expected_version = _string( (self.request.repository / "VERSION").read_text(encoding="utf-8"), "repository VERSION", ) if version != expected_version: raise GateInputError(f"image identity {target} version label mismatch") bundle_path = report.get("bundle_path") if target == "web": if bundle_path != WEB_BUNDLE_PATH: raise GateInputError("Web image identity has no canonical bundle path") elif bundle_path is not None: raise GateInputError(f"non-Web image identity {target} has a bundle path") normalized = {**report, "repo_tags": repo_tags, "repo_digests": repo_digests} except GateInputError as exc: self.block(target, "GATE-IMAGE-IDENTITY-SCHEMA", str(exc)) return None self.image_identities[target] = normalized self.actual_targets["image_identity"].add(target) return normalized def evaluate_image_identity_set(self) -> None: expected = set(IMAGE_TARGETS) actual = set(self.image_identities) identity_files: set[str] = set() if self.evidence_snapshot is not None: identity_files = { name for relative in self.evidence_snapshot.hashes() if (match := re.fullmatch(r"image-identity-(.+)\.json", relative)) for name in (match.group(1),) } if actual != expected or identity_files != expected: self.block( "shipping-images", "GATE-IMAGE-SET", "release identity set is not exact: " f"expected={sorted(expected)}, validated={sorted(actual)}, " f"evidence={sorted(identity_files)}", ) return tags = [str(self.image_identities[target]["tag"]) for target in IMAGE_TARGETS] image_ids = [str(self.image_identities[target]["image_id"]) for target in IMAGE_TARGETS] if len(set(tags)) != len(IMAGE_TARGETS) or len(set(image_ids)) != len(IMAGE_TARGETS): self.block( "shipping-images", "GATE-IMAGE-SET", "release identities contain duplicate tags or image IDs", ) def evaluate_web_bundle_evidence(self) -> None: path = self.request.reports_dir / "web-bundle-origin.json" document = self.guarded( "web", "GATE-WEB-BUNDLE", lambda: self.read_evidence_json(path) ) identity = self.image_identities.get("web") if document is None or identity is None: return try: report = _dict(document, "Web bundle evidence") if set(report) != { "schema_version", "target", "image_id", "bundle_path", "bundle_sha256", "bundle_file_count", "provenance_path", "provenance_sha256", "served_provenance_sha256", "provenance_binding_sha256", "runtime_chunk_path", "runtime_chunk_sha256", "api_origins", "non_api_origins", }: raise GateInputError("Web bundle evidence fields are not exact") if report.get("schema_version") != SCHEMA_VERSION or report.get("target") != "web": raise GateInputError("Web bundle evidence target/schema mismatch") if report.get("image_id") != identity["image_id"]: raise GateInputError("Web bundle evidence is not bound to the recorded image ID") if report.get("bundle_path") != identity["bundle_path"]: raise GateInputError("Web bundle evidence path is not bound to image identity") if ( not isinstance(report.get("bundle_sha256"), str) or re.fullmatch(r"[0-9a-f]{64}", report["bundle_sha256"]) is None or not isinstance(report.get("bundle_file_count"), int) or report["bundle_file_count"] <= 0 ): raise GateInputError("Web bundle digest or file count is malformed") if report.get("provenance_path") != WEB_PROVENANCE_PATH: raise GateInputError("Web provenance asset path is not canonical") runtime_chunk_path = _string( report.get("runtime_chunk_path"), "Web runtime chunk path" ) if ( Path(runtime_chunk_path).is_absolute() or ".." in Path(runtime_chunk_path).parts or Path(runtime_chunk_path).suffix.lower() not in {".js", ".mjs", ".cjs"} ): raise GateInputError("Web runtime chunk path escapes the bundle or is not JavaScript") for field in ( "provenance_sha256", "served_provenance_sha256", "provenance_binding_sha256", "runtime_chunk_sha256", ): if not isinstance(report.get(field), str) or re.fullmatch( r"[0-9a-f]{64}", report[field] ) is None: raise GateInputError(f"Web bundle evidence {field} is malformed") if report.get("served_provenance_sha256") != report.get( "provenance_sha256" ): raise GateInputError("served Web provenance digest differs from the bundle asset") if report.get("provenance_binding_sha256") != _web_provenance_binding( WEB_SECURITY_ORIGIN ): raise GateInputError("Web provenance binding does not identify the security origin") if report.get("api_origins") != [WEB_SECURITY_ORIGIN]: raise GateInputError("Web bundle API origins are not exactly the security origin") if report.get("non_api_origins") != list(WEB_EXPECTED_NON_API_ORIGINS): raise GateInputError("Web framework non-API origins are not the exact pinned set") except GateInputError as exc: self.block("web", "GATE-WEB-BUNDLE-SCHEMA", str(exc)) return self.actual_targets["web_bundle"].add("web") def evaluate_image_inventory(self, target: str) -> dict[str, str] | None: path = self.request.reports_dir / f"image-inventory-{target}.json" document = self.guarded( target, "GATE-IMAGE-INVENTORY", lambda: self.read_evidence_json(path) ) if document is None: return None try: report = _dict(document, "image inventory") if report.get("target") != target: raise GateInputError(f"image inventory target mismatch for {target}") packages = _list(report.get("packages"), "image inventory packages") if not packages: raise GateInputError(f"image inventory {target} is empty") result: dict[str, str] = {} for value in packages: package = _dict(value, "image package") name = _normalize_name(_string(package.get("name"), "package name")) version = _string(package.get("version"), "package version") if name in result: raise GateInputError(f"image inventory {target} duplicates package {name}") result[name] = version except GateInputError as exc: self.block(target, "GATE-IMAGE-INVENTORY-SCHEMA", str(exc)) return None missing_first_party = sorted(IMAGE_FIRST_PARTY[target] - set(result)) if missing_first_party: self.block( target, "GATE-IMAGE-FIRST-PARTY-GAP", f"shipping inventory omits first-party packages: {missing_first_party}", ) return None source = self.guarded( target, "GATE-IMAGE-FIRST-PARTY-VERSION", lambda: build_source_inventory(self.request.repository), ) if source is None: return None source_versions = { component["project_name"]: component["project_version"] for component in source["components"] } wrong_versions = sorted( name for name in IMAGE_FIRST_PARTY[target] if not _matches_first_party_version(result.get(name), source_versions.get(name)) ) if wrong_versions: self.block( target, "GATE-IMAGE-FIRST-PARTY-VERSION", f"shipping inventory has non-exact first-party versions: {wrong_versions}", ) return None non_pypi = IMAGE_NON_PYPI_AUDIT_EXCEPTIONS.get(target, {}) wrong_non_pypi = sorted( name for name, version in non_pypi.items() if result.get(name) != version ) if wrong_non_pypi: self.block( target, "GATE-IMAGE-NON-PYPI-VERSION", f"shipping inventory has absent or drifted non-PyPI packages: {wrong_non_pypi}", ) return None forbidden = sorted(IMAGE_FORBIDDEN_PACKAGES.get(target, frozenset()) & set(result)) if forbidden: self.block( target, "GATE-IMAGE-FORBIDDEN-PACKAGE", f"shipping inventory retains forbidden installer packages: {forbidden}", ) return None self.actual_targets["python_image_inventory"].add(target) return result def evaluate_versions(self) -> None: versions = self.request.reports_dir / "versions" for scanner, filename in ( ("pip-audit", "pip-audit.txt"), ("gitleaks", "gitleaks.txt"), ("npm", "npm.txt"), ): path = versions / filename def read_version() -> str: text = self.read_evidence_text(path, encoding="utf-8").strip() if not text: raise GateInputError(f"empty version report: {path}") pattern = r"(? evaluation_time: raise GateInputError("Trivy DB UpdatedAt is in the future") if downloaded_at > evaluation_time: raise GateInputError("Trivy DB DownloadedAt is in the future") if downloaded_at < updated_at: raise GateInputError("Trivy DB DownloadedAt predates UpdatedAt") updated_age = evaluation_time - updated_at downloaded_age = evaluation_time - downloaded_at if updated_age > TRIVY_DB_MAX_AGE: raise GateInputError( f"Trivy DB is stale ({updated_age}); maximum age is " f"{TRIVY_DB_MAX_AGE}" ) if downloaded_age > TRIVY_DB_DOWNLOAD_MAX_AGE: raise GateInputError( f"Trivy DB download is stale ({downloaded_age}); maximum age is " f"{TRIVY_DB_DOWNLOAD_MAX_AGE}" ) self.database_times["trivy"] = { "version": database_version, "updated_at": updated_at.isoformat().replace("+00:00", "Z"), "downloaded_at": downloaded_at.isoformat().replace("+00:00", "Z"), "maximum_updated_age_hours": int( TRIVY_DB_MAX_AGE.total_seconds() // 3600 ), "maximum_downloaded_age_hours": int( TRIVY_DB_DOWNLOAD_MAX_AGE.total_seconds() // 3600 ), } except GateInputError as exc: self.block("trivy", "GATE-TRIVY-DB-TIME", str(exc)) else: self.block("trivy", "GATE-TRIVY-DB-TIME", "Trivy vulnerability DB metadata absent") for scanner, pinned in PINNED_SCANNER_VERSIONS.items(): actual = self.scanner_versions.get(scanner) if actual != pinned: self.block( scanner, "GATE-SCANNER-VERSION-MISMATCH", f"{scanner} version is {actual!r}; required exact version is {pinned}", ) def evaluate_target_coverage(self) -> None: for category, expected_values in self.expected_targets.items(): expected = set(expected_values) actual = self.actual_targets[category] if actual != expected: self.block( category, "GATE-TARGET-GAP", f"expected={sorted(expected)}, actual={sorted(actual)}", ) def evaluate_risks(self) -> tuple[list[dict[str, Any]], list[dict[str, Any]]]: document = self.guarded( "accepted-risks", "GATE-RISK-REPORT", lambda: _read_json(self.request.accepted_risks) ) if document is None: return [], [] if not isinstance(document, dict) or set(document) != { "schema_version", "accepted_risks", }: self.block( "accepted-risks", "GATE-RISK-SCHEMA", "accepted-risk registry must be an exact versioned object", ) return [], [] if document.get("schema_version") != SCHEMA_VERSION: self.block( "accepted-risks", "GATE-RISK-SCHEMA", f"accepted-risk schema_version must be {SCHEMA_VERSION}", ) return [], [] entries = document.get("accepted_risks") if not isinstance(entries, list): self.block( "accepted-risks", "GATE-RISK-SCHEMA", "accepted_risks must be a JSON list", ) return [], [] valid: list[dict[str, Any]] = [] invalid_or_unused: list[dict[str, Any]] = [] identities: set[tuple[str, str, str, str, str]] = set() for index, value in enumerate(entries): entry_result: dict[str, Any] = {"index": index, "entry": value} try: entry = _dict(value, f"accepted risk {index}") if set(entry) != RISK_FIELDS: raise GateInputError( f"accepted risk {index} fields differ: expected={sorted(RISK_FIELDS)}, " f"actual={sorted(entry)}" ) identity_values = ( _string(entry["scanner"], f"accepted risk {index}.scanner"), _string(entry["target"], f"accepted risk {index}.target"), _string( entry["advisory_id"], f"accepted risk {index}.advisory_id" ), _string(entry["package"], f"accepted risk {index}.package"), _string( entry["installed_version"], f"accepted risk {index}.installed_version", ), ) if entry["scanner"] not in RISK_SCANNERS: raise GateInputError(f"accepted risk {index} scanner is not allowlisted") if entry["classification"] != "ACCEPTED_TEMPORARY": raise GateInputError( f"accepted risk {index} classification must be ACCEPTED_TEMPORARY" ) for field_value in identity_values: if re.search(r"[*?\[\]]", field_value) or field_value.lower() in { "all", "any", }: raise GateInputError(f"accepted risk {index} identity contains a wildcard") if identity_values in identities: raise GateInputError(f"accepted risk {index} duplicates an exact risk tuple") identities.add(identity_values) owner = _string(entry["owner"], f"accepted risk {index}.owner") tracking = _string(entry["tracking"], f"accepted risk {index}.tracking") reason = _string(entry["reason"], f"accepted risk {index}.reason") if not tracking.startswith("https://"): raise GateInputError(f"accepted risk {index} tracking must be an HTTPS URL") accepted_on = date.fromisoformat( _string(entry["accepted_on"], f"accepted risk {index}.accepted_on") ) expires_on = date.fromisoformat( _string(entry["expires_on"], f"accepted risk {index}.expires_on") ) duration = (expires_on - accepted_on).days if duration < 0 or duration > 30: raise GateInputError( f"accepted risk {index} duration is {duration} days; maximum is 30" ) if accepted_on > self.request.as_of: raise GateInputError(f"accepted risk {index} is not active yet") # expires_on is the first invalid day. With accepted_on=Sep 2 and # expires_on=Oct 2 the risk is usable on exactly 30 calendar dates. if expires_on <= self.request.as_of: raise GateInputError(f"accepted risk {index} is expired") valid.append( { **entry, "owner": owner, "tracking": tracking, "reason": reason, "_identity": identity_values, "_index": index, } ) except (GateInputError, ValueError) as exc: entry_result["reason"] = str(exc) invalid_or_unused.append(entry_result) self.block("accepted-risks", "GATE-RISK-INVALID", str(exc)) applied: list[dict[str, Any]] = [] accepted_finding_indexes: set[int] = set() for risk in valid: # Image risks are only consumable when the explicit security profile actually builds # and scans images. The source-only full profile still validates their schema and # active dates, but cannot truthfully call them stale without image evidence. if ( self.request.profile != "security" and risk["scanner"] == "trivy" and risk["target"] in IMAGE_TARGETS ): continue matches = [ index for index, finding in enumerate(self.findings) if finding.identity == risk["_identity"] ] public_risk = {key: value for key, value in risk.items() if not key.startswith("_")} if not matches: item = {"index": risk["_index"], "entry": public_risk, "reason": "unused or stale"} invalid_or_unused.append(item) self.block( "accepted-risks", "GATE-RISK-UNUSED", f"accepted risk {risk['_index']} matches no blocking finding", ) continue if len(matches) != 1: item = {"index": risk["_index"], "entry": public_risk, "reason": "ambiguous"} invalid_or_unused.append(item) self.block( "accepted-risks", "GATE-RISK-AMBIGUOUS", f"accepted risk {risk['_index']} matches {len(matches)} findings", ) continue finding = self.findings[matches[0]] if finding.fix_versions: item = { "index": risk["_index"], "entry": public_risk, "reason": "fix available", } invalid_or_unused.append(item) self.block( "accepted-risks", "GATE-RISK-FIX-AVAILABLE", f"accepted risk {risk['_index']} has fix versions {list(finding.fix_versions)}", ) continue if finding.finding_type != "vulnerability": item = { "index": risk["_index"], "entry": public_risk, "reason": "finding type cannot be accepted", } invalid_or_unused.append(item) self.block( "accepted-risks", "GATE-RISK-TYPE", f"accepted risk {risk['_index']} is not a vulnerability", ) continue accepted_finding_indexes.add(matches[0]) applied.append({"entry": public_risk, "finding": finding.as_json(reason="accepted")}) self.findings = [ finding for index, finding in enumerate(self.findings) if index not in accepted_finding_indexes ] return applied, invalid_or_unused def report_hashes(self) -> dict[str, str]: if self.evidence_snapshot is None: self.block( "evidence", "GATE-EVIDENCE-PATH", self.evidence_tree_error or "secure evidence snapshot is absent", ) return {} try: return self.evidence_snapshot.hashes() except (GateInputError, OSError) as exc: self.block("evidence", "GATE-REPORT-HASH", f"cannot revalidate reports: {exc}") return {} def policy_hashes(self) -> dict[str, str | None]: result: dict[str, str | None] = {} for name, path in ( ("accepted_risks", self.request.accepted_risks), ("gitleaks_false_positives", self.request.gitleaks_false_positives), ): try: result[name] = _sha256(path) except (GateInputError, OSError) as exc: result[name] = None self.block(name, "GATE-POLICY-HASH", f"cannot hash policy: {exc}") return result def result( self, *, applied_false_positives: list[dict[str, Any]], unused_false_positives: list[dict[str, Any]], applied_risks: list[dict[str, Any]], unused_risks: list[dict[str, Any]], hashes: dict[str, str], policy_hashes: dict[str, str | None], ) -> dict[str, Any]: blocked_findings = self.policy_blocks + [ finding.as_json(reason="unaccepted finding") for finding in self.findings ] return { "schema_version": SCHEMA_VERSION, "commit": self.request.commit, "ref": self.request.ref, "run_id": self.request.run_id, "profile": self.request.profile, "evaluated_at": self.request.evaluation_time.isoformat(), "risk_policy_date": self.request.as_of.isoformat(), "scanner_versions": self.scanner_versions, "vulnerability_database_times": self.database_times, "exit_codes": self.exit_codes, "report_hashes": hashes, "policy_hashes": policy_hashes, "expected_targets": self.expected_targets, "actual_targets": { category: sorted(targets) for category, targets in self.actual_targets.items() }, "blocked_findings": blocked_findings, "applied_gitleaks_false_positives": applied_false_positives, "unused_or_invalid_gitleaks_false_positives": unused_false_positives, "applied_risks": applied_risks, "unused_or_invalid_risks": unused_risks, "resolved_transitive_packages": self.resolved_transitives, "image_identities": self.image_identities, "decision": "PASS" if not blocked_findings else "FAIL", } def run(self) -> dict[str, Any]: if self.evidence_tree_error is not None: self.block("evidence", "GATE-EVIDENCE-PATH", self.evidence_tree_error) return self.result( applied_false_positives=[], unused_false_positives=[], applied_risks=[], unused_risks=[], hashes={}, policy_hashes=self.policy_hashes(), ) self.evaluate_statuses() self.evaluate_history() source_packages = self.evaluate_source_inventory() self.evaluate_gitleaks() for target in SOURCE_COMPONENTS: self.evaluate_pip_report( self.request.reports_dir / f"pip-source-{target}.json", target, source_packages.get(target, {}), "python_source", allow_resolved_transitives=True, ) self.evaluate_npm("npm-audit-all.json", "frontend:all", "npm-audit-all") self.evaluate_npm( "npm-audit-production.json", "frontend:production", "npm-audit-production" ) self.evaluate_trivy("trivy-fs.json", "repository", "trivy_filesystem") if self.request.profile == "security": for target in IMAGE_TARGETS: self.evaluate_image_identity(target) self.evaluate_image_identity_set() self.evaluate_web_bundle_evidence() for target in PYTHON_IMAGE_TARGETS: inventory = self.evaluate_image_inventory(target) if inventory is not None: first_party_names = { project_name for _, project_name in SOURCE_COMPONENTS.values() } auditable_inventory = { name: version for name, version in inventory.items() if name not in first_party_names and name not in IMAGE_NON_PYPI_AUDIT_EXCEPTIONS.get(target, {}) } self.evaluate_pip_report( self.request.reports_dir / f"pip-image-{target}.json", target, auditable_inventory, "python_image_audit", allow_resolved_transitives=False, ) for target in IMAGE_TARGETS: self.evaluate_trivy(f"trivy-image-{target}.json", target, "image_scan") self.evaluate_versions() self.evaluate_target_coverage() applied_false_positives, unused_false_positives = ( self.evaluate_gitleaks_false_positives() ) applied_risks, unused_risks = self.evaluate_risks() hashes = self.report_hashes() policy_hashes = self.policy_hashes() return self.result( applied_false_positives=applied_false_positives, unused_false_positives=unused_false_positives, applied_risks=applied_risks, unused_risks=unused_risks, hashes=hashes, policy_hashes=policy_hashes, ) def evaluate_security_gate(request: EvaluationRequest) -> dict[str, Any]: """Evaluate reports. The caller is responsible for persisting the returned result.""" return Evaluator(request).run() def _emergency_failure_result(request: EvaluationRequest, error: BaseException) -> dict[str, Any]: return { "schema_version": SCHEMA_VERSION, "commit": request.commit, "ref": request.ref, "run_id": request.run_id, "profile": request.profile, "evaluated_at": request.evaluation_time.isoformat(), "risk_policy_date": request.as_of.isoformat(), "scanner_versions": {}, "vulnerability_database_times": {}, "exit_codes": {}, "report_hashes": {}, "policy_hashes": {}, "expected_targets": expected_targets(request.profile), "actual_targets": {category: [] for category in expected_targets(request.profile)}, "blocked_findings": [ { "scanner": "security-gate", "target": "evaluator", "advisory_id": "GATE-EVALUATOR-EXCEPTION", "package": "", "installed_version": "", "severity": "ERROR", "fix_versions": [], "finding_type": "policy", "reason": f"{type(error).__name__}: {error}", } ], "applied_gitleaks_false_positives": [], "unused_or_invalid_gitleaks_false_positives": [], "applied_risks": [], "unused_or_invalid_risks": [], "resolved_transitive_packages": {}, "image_identities": {}, "decision": "FAIL", } def _parser() -> argparse.ArgumentParser: parser = argparse.ArgumentParser(description=__doc__) subparsers = parser.add_subparsers(dest="command", required=True) source = subparsers.add_parser("source-inventory") source.add_argument("--repository", type=Path, required=True) source.add_argument("--output", type=Path, required=True) source.add_argument("--requirements-directory", type=Path, required=True) gitleaks = subparsers.add_parser("sanitize-gitleaks") gitleaks.add_argument("--input", type=Path, required=True) gitleaks.add_argument("--output", type=Path, required=True) image = subparsers.add_parser("image-inventory") image.add_argument("--target", choices=PYTHON_IMAGE_TARGETS, required=True) image.add_argument("--input", type=Path, required=True) image.add_argument("--output", type=Path, required=True) image.add_argument("--requirements-output", type=Path, required=True) identity = subparsers.add_parser("image-identity") identity.add_argument("--target", choices=IMAGE_TARGETS, required=True) identity.add_argument("--tag", required=True) identity.add_argument("--expected-revision", required=True) identity.add_argument("--expected-source", required=True) identity.add_argument("--expected-version", required=True) identity.add_argument("--output", type=Path, required=True) trivy_binding = subparsers.add_parser("bind-trivy-image") trivy_binding.add_argument("--input", type=Path, required=True) trivy_binding.add_argument("--output", type=Path, required=True) trivy_binding.add_argument("--image-id", required=True) trivy_binding.add_argument("--expected-tag", required=True) web = subparsers.add_parser("web-bundle-evidence") web.add_argument("--image-id", required=True) web.add_argument("--bundle-path", required=True) web.add_argument("--expected-origin", required=True) web.add_argument("--output", type=Path, required=True) history = subparsers.add_parser("history") history.add_argument("--repository", type=Path, required=True) history.add_argument("--expected-commit", required=True) history.add_argument("--output", type=Path, required=True) evaluate = subparsers.add_parser("evaluate") evaluate.add_argument("--reports-dir", type=Path, required=True) evaluate.add_argument("--accepted-risks", type=Path, required=True) evaluate.add_argument("--gitleaks-false-positives", type=Path, required=True) evaluate.add_argument("--repository", type=Path, required=True) evaluate.add_argument("--commit", required=True) evaluate.add_argument("--ref", required=True) evaluate.add_argument("--run-id", required=True) evaluate.add_argument("--source-repository", required=True) evaluate.add_argument("--profile", choices=SECURITY_PROFILES, required=True) evaluate.add_argument("--output", type=Path, required=True) evaluate.add_argument("--as-of", type=date.fromisoformat, default=None) evaluate.add_argument("--evaluation-time", type=_parse_cli_utc_timestamp, default=None) return parser def main(argv: list[str] | None = None) -> int: args = _parser().parse_args(argv) try: if args.command == "source-inventory": collect_source_inventory(args.repository, args.output, args.requirements_directory) return 0 if args.command == "sanitize-gitleaks": sanitize_gitleaks(args.input, args.output) return 0 if args.command == "image-inventory": normalize_image_inventory( args.target, args.input, args.output, args.requirements_output ) return 0 if args.command == "image-identity": record_image_identity( args.target, args.tag, args.expected_revision, args.expected_source, args.expected_version, args.output, ) return 0 if args.command == "bind-trivy-image": bind_trivy_image_report( args.input, args.output, args.image_id, args.expected_tag ) return 0 if args.command == "web-bundle-evidence": record_web_bundle_evidence( args.image_id, args.bundle_path, args.expected_origin, args.output ) return 0 if args.command == "history": record_history(args.repository, args.expected_commit, args.output) return 0 if args.command == "evaluate": evaluation_time = args.evaluation_time or datetime.now(UTC) request = EvaluationRequest( reports_dir=args.reports_dir, accepted_risks=args.accepted_risks, gitleaks_false_positives=args.gitleaks_false_positives, repository=args.repository, commit=args.commit, ref=args.ref, run_id=args.run_id, source_repository=args.source_repository, profile=args.profile, output=args.output, as_of=args.as_of or evaluation_time.date(), evaluation_time=evaluation_time, ) try: result = evaluate_security_gate(request) except BaseException as exc: # fail closed and still retain the evaluator error result = _emergency_failure_result(request, exc) _write_json(request.output, result) print( f"security gate: {result['decision']} " f"({len(result['blocked_findings'])} blocked finding(s))" ) return 0 if result["decision"] == "PASS" else 1 except (GateInputError, OSError, ValueError, TypeError) as exc: print(f"security gate input error: {exc}", file=sys.stderr) return 2 raise AssertionError(f"unhandled command: {args.command}") if __name__ == "__main__": raise SystemExit(main())