Jens/ModelForge-Public
1
0 Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
main
ModelForge-Public/scripts/security_gate.py
T

3722 lines
161 KiB
Python
Raw Permalink Blame History

#!/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<literal>"(?:\\.|[^"\\])*"|'[^']*')\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"(?<![0-9])([0-9]+\.[0-9]+\.[0-9]+)(?![0-9])"
match = re.search(pattern, text)
if match is None:
raise GateInputError(f"cannot parse version report: {path}")
return match.group(1)
parsed = self.guarded(scanner, "GATE-SCANNER-VERSION", read_version)
if parsed is not None:
self.scanner_versions[scanner] = parsed
trivy_path = versions / "trivy.json"
trivy_document = self.guarded(
"trivy", "GATE-SCANNER-VERSION", lambda: self.read_evidence_json(trivy_path)
)
if isinstance(trivy_document, dict):
version = trivy_document.get("Version", trivy_document.get("version"))
if isinstance(version, str):
self.scanner_versions["trivy"] = version.lstrip("v")
database = trivy_document.get(
"VulnerabilityDB", trivy_document.get("vulnerability_db")
)
if isinstance(database, dict):
updated = database.get("UpdatedAt", database.get("updated_at"))
downloaded = database.get("DownloadedAt", database.get("downloaded_at"))
database_version = database.get("Version", database.get("version"))
try:
updated_at = _utc_timestamp(updated, "Trivy DB UpdatedAt")
downloaded_at = _utc_timestamp(downloaded, "Trivy DB DownloadedAt")
evaluation_time = self.request.evaluation_time.astimezone(UTC)
if updated_at > 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())
Reference in new issue View Git Blame Copy Permalink