"""Policy and synthetic evaluator tests for the fail-closed managed security gate.""" from __future__ import annotations import copy import hashlib import json import os import re import shutil import stat import subprocess import tempfile import unittest from datetime import UTC, date, datetime from pathlib import Path from types import SimpleNamespace from typing import Any, ClassVar from unittest import mock import tomllib from scripts import security_gate as gate REPOSITORY_ROOT = Path(__file__).resolve().parents[2] WORKFLOW_PATH = REPOSITORY_ROOT / ".gitea" / "workflows" / "managed-validation.yml" GITLEAKS_CONFIG = REPOSITORY_ROOT / ".gitleaks.toml" PIP_AUDIT_LOCK = REPOSITORY_ROOT / "config" / "security" / "pip-audit-toolchain.lock" ACCEPTED_RISKS = REPOSITORY_ROOT / "config" / "security" / "accepted-risks.json" GITLEAKS_FALSE_POSITIVES = ( REPOSITORY_ROOT / "config" / "security" / "gitleaks-false-positives.json" ) PRODUCT_VERSION = (REPOSITORY_ROOT / "VERSION").read_text(encoding="utf-8").strip() def _indented_block(text: str, header: str) -> str: lines = text.splitlines() try: start = lines.index(header) except ValueError as exc: raise AssertionError(f"Missing YAML header: {header!r}") from exc indent = len(header) - len(header.lstrip()) end = len(lines) for index in range(start + 1, len(lines)): line = lines[index] if not line.strip() or line.lstrip().startswith("#"): continue current_indent = len(line) - len(line.lstrip()) if current_indent <= indent: end = index break return "\n".join(lines[start:end]) def _named_step(text: str, name: str) -> str: return _indented_block(text, f" - name: {name}") def _run_script(step: str) -> str: lines = step.splitlines() if " run: |" not in lines: inline = next((line for line in lines if line.startswith(" run: ")), None) if inline is None: raise AssertionError("Named step does not contain a run command") return inline.removeprefix(" run: ") + "\n" start = lines.index(" run: |") + 1 script_lines = lines[start:] if any(line and not line.startswith(" ") for line in script_lines): raise AssertionError("Unexpected indentation in literal run block") return "\n".join(line[10:] if line else "" for line in script_lines) + "\n" def _timeout_minutes(block: str) -> int: match = re.search(r"(?m)^\s+timeout-minutes: ([0-9]+)$", block) if match is None: raise AssertionError("YAML block has no explicit timeout-minutes") return int(match.group(1)) def _write_json(path: Path, value: Any) -> None: path.parent.mkdir(parents=True, exist_ok=True) path.write_text(json.dumps(value, indent=2) + "\n", encoding="utf-8") def _git(repository: Path, *arguments: str) -> subprocess.CompletedProcess[str]: return subprocess.run( ["git", "-C", str(repository), *arguments], capture_output=True, check=True, text=True, timeout=30, ) def _copy_source_gate_inputs(destination: Path) -> None: relative_paths = [ *(manifest for manifest, _ in gate.SOURCE_COMPONENTS.values()), gate.BUILD_TOOLCHAIN_LOCK, *gate.IMAGE_DOCKERFILES.values(), ] for relative_path in relative_paths: source = REPOSITORY_ROOT / relative_path target = destination / relative_path target.parent.mkdir(parents=True, exist_ok=True) shutil.copyfile(source, target) class SecurityWorkflowPolicyTests(unittest.TestCase): workflow: ClassVar[str] security_job: ClassVar[str] @classmethod def setUpClass(cls) -> None: cls.workflow = WORKFLOW_PATH.read_text(encoding="utf-8") cls.security_job = _indented_block(cls.workflow, " security:") def test_all_checkouts_are_full_history_without_persisted_credentials(self) -> None: checkouts = [ _named_step(self.workflow, "Check out validation source with full history"), _named_step(self.workflow, "Check out security source with full history"), ] self.assertEqual(self.workflow.count("uses: actions/checkout@"), 2) for checkout in checkouts: with self.subTest(checkout=checkout.splitlines()[0]): self.assertRegex(checkout, r"actions/checkout@[0-9a-f]{40}") self.assertIn("fetch-depth: 0", checkout) self.assertIn("persist-credentials: false", checkout) security_checkout = _named_step( self.workflow, "Check out security source with full history" ) self.assertIn("fetch-tags: true", security_checkout) history = _run_script(_named_step(self.workflow, "Prove complete non-shallow history")) self.assertIn("+refs/heads/*:refs/remotes/origin/*", history) self.assertIn("+refs/tags/*:refs/tags/*", history) self.assertIn("git fetch --force --prune", history) self.assertIn("git-history.json", history) self.assertIn('git remote set-url origin "${SOURCE_REPOSITORY}"', history) self.assertIn("x-access-token:%s", history) self.assertIn("git config --local http.extraheader", history) self.assertIn("git config --local --unset-all http.extraheader", history) history_step = _named_step(self.workflow, "Prove complete non-shallow history") self.assertIn("GITHUB_TOKEN: ${{ github.token }}", history_step) def test_profiles_are_source_mandatory_and_only_explicit_security_builds_images(self) -> None: self.assertIn("inputs.profile == 'security'", self.security_job) self.assertIn("inputs.profile == 'full'", self.security_job) self.assertIn("!inputs.profile", self.security_job) self.assertIn("SECURITY_PROFILE:", self.security_job) images = _named_step(self.workflow, "Build and scan exact shipping images") self.assertIn("gitea.event_name == 'workflow_dispatch'", images) self.assertIn("inputs.profile == 'security'", images) source = _named_step(self.workflow, "Collect source security reports") self.assertNotIn("if:", source) def test_time_budget_arithmetic_reserves_evaluator_and_upload(self) -> None: names = ( "Prepare fail-closed security evidence", "Check out security source with full history", "Prove complete non-shallow history", "Install checksum-pinned security scanners", "Collect source security reports", "Build and scan exact shipping images", "Remove exact security images", "Evaluate the complete security gate", "Upload immutable security evidence", ) bounds = {name: _timeout_minutes(_named_step(self.workflow, name)) for name in names} job_timeout = _timeout_minutes(self.security_job.split(" steps:", maxsplit=1)[0]) self.assertGreaterEqual(bounds["Evaluate the complete security gate"], 5) self.assertGreaterEqual(bounds["Upload immutable security evidence"], 5) self.assertGreaterEqual(job_timeout - sum(bounds.values()), 10) def test_scanner_artifacts_are_version_and_checksum_pinned(self) -> None: expected = { "PIP_AUDIT_VERSION: 2.10.1", "99ef3f600a317c1945f1e89e227ef26e1c2d618429b8bd3fa6f4f7c440c4611a", "GITLEAKS_VERSION: 8.30.1", "551f6fc83ea457d62a0d98237cbad105af8d557003051f41f3e7ca7b3f2470eb", "TRIVY_VERSION: 0.74.0", "2ae6fe3ee734b7fdf11335663e18c75ea12dccc76062f09f164a3b0f8be4371a", } for fragment in expected: with self.subTest(fragment=fragment): self.assertIn(fragment, self.security_job) self.assertNotRegex(self.security_job, r"curl[^\n]*\|\s*(?:ba)?sh") self.assertNotIn("trivy-action", self.security_job) def test_release_version_gate_accepts_only_stable_or_numbered_rc_semver(self) -> None: history = _run_script(_named_step(self.workflow, "Prove complete non-shallow history")) self.assertIn( '^(0|[1-9][0-9]*)\\.(0|[1-9][0-9]*)\\.(0|[1-9][0-9]*)' '(-rc\\.(0|[1-9][0-9]*))?$', history, ) self.assertNotIn('^[0-9]+\\.[0-9]+\\.[0-9]+$', history) self.assertNotIn("gitleaks-action", self.security_job) def test_pip_audit_toolchain_is_fully_resolved_and_hash_locked(self) -> None: entries = [ line.strip() for line in PIP_AUDIT_LOCK.read_text(encoding="utf-8").splitlines() if line.strip() and not line.lstrip().startswith("#") ] self.assertGreaterEqual(len(entries), 20) for entry in entries: with self.subTest(entry=entry): self.assertRegex( entry, r"^[A-Za-z0-9._-]+==[^ ]+ --hash=sha256:[0-9a-f]{64}$", ) self.assertEqual(sum(entry.startswith("pip-audit==2.10.1 ") for entry in entries), 1) install = _named_step(self.workflow, "Install checksum-pinned security scanners") self.assertTrue(_run_script(install).startswith("set -euo pipefail\n")) self.assertIn("--index-url https://pypi.org/simple", install) self.assertGreaterEqual(install.count("--require-hashes"), 2) self.assertIn("--no-index", install) self.assertIn("config/security/pip-audit-toolchain.lock", install) def test_source_scans_are_complete_and_do_not_suppress_findings(self) -> None: source = _named_step(self.workflow, "Collect source security reports") self.assertTrue(_run_script(source).startswith("set -euo pipefail\n")) required = ( "source-inventory", "--log-opts=--all", "--redact=100", "sanitize-gitleaks", "pip-source-backend.json", "pip-source-node-agent.json", "pip-source-runtime-worker.json", '"${PIP_AUDIT}" --strict', "npm audit --json", "npm audit --omit=dev --json", "trivy-fs.json", "--scanners vuln,misconfig", "--severity UNKNOWN,LOW,MEDIUM,HIGH,CRITICAL", '"${TRIVY}" version --format json', ) for fragment in required: with self.subTest(fragment=fragment): self.assertIn(fragment, source) for forbidden in ("--ignore-vuln", "--ignore-unfixed", "--exit-code", "|| true"): self.assertNotIn(forbidden, source) self.assertNotIn("--no-deps", source) self.assertNotIn("--disable-pip", source) self.assertNotIn("${GITHUB_WORKSPACE}/${REPORTS}", source) def test_exactly_four_commit_bound_non_push_images_are_built_and_scanned(self) -> None: images = _named_step(self.workflow, "Build and scan exact shipping images") self.assertTrue(_run_script(images).startswith("set -euo pipefail\n")) self.assertEqual(images.count(" docker build --pull"), 4) self.assertEqual(images.count("docker build --pull --no-cache"), 4) self.assertIn("modelforge_runtime_worker.main", images) self.assertIn("sentence_transformers", images) self.assertIn('{"pip", "setuptools", "wheel"} & packages', images) self.assertEqual(images.count("scan_image "), 4) self.assertIn("bind-trivy-image", images) self.assertIn('--expected-tag "modelforge-security-${target}:${GITHUB_SHA}"', images) self.assertIn('${SECURITY_TOOLS}/trivy-image-${target}.raw.json', images) # Four image tags plus the exact expected tag passed to the Trivy binder. self.assertEqual(images.count(":${GITHUB_SHA}"), 5) for target in ("api", "web", "node-agent", "runtime-worker"): self.assertIn(f"image-build-{target}", images) self.assertIn(f"scan_image {target} ", images) self.assertNotIn("docker push", images) self.assertNotIn("--push", images) self.assertEqual(images.count("image-identity --target"), 1) self.assertIn('"${target}"', images) self.assertIn('"${API_ID}"', images) self.assertIn('"${WEB_ID}"', images) self.assertIn("org.opencontainers.image.revision=${GITHUB_SHA}", images) self.assertEqual(images.count("org.opencontainers.image.source=${SOURCE_REPOSITORY}"), 4) self.assertEqual(images.count("org.opencontainers.image.version=${MODELFORGE_VERSION}"), 4) self.assertIn("io.modelforge.security.bundle-path=/usr/share/nginx/html", images) self.assertIn('--expected-source "${SOURCE_REPOSITORY}"', images) self.assertIn('--expected-version "${MODELFORGE_VERSION}"', images) self.assertIn('--bundle-path "${WEB_BUNDLE_PATH}"', images) self.assertIn("VITE_API_BASE_URL=https://security-scan.invalid", images) self.assertIn("web-bundle-evidence", images) cleanup = _named_step(self.workflow, "Remove exact security images") self.assertIn("always()", cleanup) self.assertIn("docker image rm", cleanup) self.assertIn("docker image inspect", cleanup) self.assertIn("status/image-cleanup.exit", cleanup) self.assertNotIn("docker image prune", cleanup) def test_image_source_uses_the_canonical_https_repository(self) -> None: source = _run_script(_named_step(self.workflow, "Prepare fail-closed security evidence")) self.assertIn( '[[ "${GITHUB_REPOSITORY}" == "Jens/ITWorx-ModelForge" ]]', source ) self.assertIn( 'SOURCE_REPOSITORY}" == "https://git.example.com/example/modelforge"', source, ) for relative_path in gate.IMAGE_DOCKERFILES.values(): dockerfile = (REPOSITORY_ROOT / relative_path).read_text(encoding="utf-8") self.assertEqual( dockerfile.count( f'LABEL org.opencontainers.image.source="{gate.CANONICAL_SOURCE_REPOSITORY}"' ), 1, relative_path, ) def test_api_release_builds_include_the_root_toolchain_lock(self) -> None: image_build = _run_script( _named_step(self.workflow, "Build and scan exact shipping images") ) self.assertIn("--file backend/Dockerfile .", image_build) release_builder = (REPOSITORY_ROOT / "scripts" / "release_build.py").read_text( encoding="utf-8" ) self.assertIn('"modelforge-api": (".", "backend/Dockerfile")', release_builder) for relative_path, services in ( ("docker-compose.yml", ("api", "migrate")), ("docker-compose.production.yml", ("api",)), ("docker-compose.dr.yml", ("dr-api",)), ): compose = (REPOSITORY_ROOT / relative_path).read_text(encoding="utf-8") for service in services: block = re.search( rf"(?ms)^ {re.escape(service)}:\n(.*?)(?=^ [A-Za-z0-9_-]+:|\Z)", compose, ) self.assertIsNotNone(block, f"{relative_path} is missing {service}") self.assertRegex( block.group(1), r"(?m)^ build:\n context: \.\n dockerfile: backend/Dockerfile$", ) def test_repository_root_image_context_excludes_local_secrets_and_model_data(self) -> None: dockerignore = set( (REPOSITORY_ROOT / ".dockerignore").read_text(encoding="utf-8").splitlines() ) self.assertTrue({".env", "**/.env", "**/.env.*"}.issubset(dockerignore)) self.assertTrue( { "**/*.key", "**/*.pem", "**/*.crt", "**/*.p12", "**/*.pfx", "**/id_rsa*", "**/id_ed25519*", "**/private_key*", "**/secrets/**", "**/credentials/**", "**/.npmrc", "**/.pypirc", "**/.netrc", "**/.git-credentials", "**/.htpasswd", "**/pip.conf", "**/.state/**", "**/.ssh/**", "**/.aws/**", }.issubset(dockerignore) ) self.assertTrue( { "/models/", "/model-artifacts/", "/artifacts/", "/data/", "/.data/", "/.state/", "/state/", "/hf-cache/", "/huggingface/", "**/*.safetensors", "**/*.gguf", "**/*.ckpt", "**/*.onnx", "**/*.bin", "**/*.h5", "**/*.pb", "**/*.model", }.issubset(dockerignore) ) self.assertNotIn("models/", dockerignore) self.assertTrue((REPOSITORY_ROOT / "config/models/initial-candidates.yaml").is_file()) self.assertIn( ".dockerignore", (REPOSITORY_ROOT / "public-source.allowlist").read_text(encoding="utf-8").splitlines(), ) def test_console_network_policy_admits_only_the_build_bound_api_origin(self) -> None: headers = (REPOSITORY_ROOT / "frontend" / "security-headers.inc.template").read_text( encoding="utf-8" ) dockerfile = (REPOSITORY_ROOT / "frontend" / "Dockerfile").read_text(encoding="utf-8") self.assertIn("connect-src __API_ORIGIN__;", headers) self.assertNotIn("connect-src 'self'", headers) self.assertIn('grep -q "connect-src ${API_ORIGIN};"', dockerfile) def test_shipping_python_inventory_is_extracted_and_strictly_audited(self) -> None: images = _named_step(self.workflow, "Build and scan exact shipping images") self.assertEqual(images.count("collect_inventory "), 3) self.assertEqual(images.count("pip-image-") // 2, 3) for target in ("api", "node-agent", "runtime-worker"): self.assertIn(f'collect_inventory {target} "', images) self.assertIn(f"pip-image-{target}.json", images) self.assertIn("importlib.metadata", images) self.assertIn("--strict --no-deps", images) self.assertIn("--disable-pip", images) def test_python_validation_environments_are_not_shared(self) -> None: self.assertIn("BACKEND_VENV:", self.workflow) self.assertIn("NODE_AGENT_VENV:", self.workflow) self.assertIn("RUNTIME_WORKER_VENV:", self.workflow) self.assertNotIn("VENV: /tmp/modelforge-validation-venv", self.workflow) def test_artifact_upload_is_immutable_retained_and_fail_closed(self) -> None: upload = _named_step(self.workflow, "Upload immutable security evidence") self.assertIn("if: always()", upload) self.assertIn( "https://gitea.com/actions/gitea-upload-artifact@" "81f940d004763f986ba3582c007fd842dd5cb0d7", upload, ) self.assertIn( "name: security-gate-${{ gitea.run_id }}-${{ gitea.run_attempt }}-" "${{ gitea.sha }}", upload, ) self.assertIn("if-no-files-found: error", upload) self.assertIn("retention-days: 90", upload) self.assertIn("path: |", upload) self.assertNotIn("path: ${{ env.REPORTS }}\n", upload) uploaded_paths = set( re.findall(r"\$\{\{ env\.REPORTS \}\}/([^\r\n]+)", upload) ) self.assertEqual( uploaded_paths, {"security-gate-result.json", *gate.expected_evidence_paths("security")}, ) self.assertNotIn("continue-on-error", upload) def test_gitleaks_scanner_config_has_narrow_current_github_rules_and_no_allowlist(self) -> None: with GITLEAKS_CONFIG.open("rb") as handle: config = tomllib.load(handle) self.assertNotIn("extend", config) rules = {rule["id"]: rule for rule in config["rules"]} self.assertEqual( set(rules), { "modelforge-github-opaque-token", "modelforge-github-installation-token", "modelforge-github-fine-grained-pat", }, ) serialized = json.dumps(config).lower() self.assertNotIn("allowlist", serialized) self.assertTrue(all(rule.get("secretGroup") == 1 for rule in rules.values())) source = _run_script(_named_step(self.workflow, "Collect source security reports")) self.assertIn( 'DEFAULT_GITLEAKS_REPOSITORY="${GITLEAKS_TEMP}/default-worktree"', source ) self.assertIn('rm -f -- "${DEFAULT_GITLEAKS_REPOSITORY}/.gitleaks.toml"', source) self.assertIn("--config .gitleaks.toml", source) self.assertIn("RAW_GITLEAKS_DEFAULT", source) self.assertIn("RAW_GITLEAKS_CUSTOM", source) self.assertIn("gitleaks-merge.exit", source) classic = "".join(("gh", "p_", "aBcDeFgHiJkLmNoPqRsTuVwXyZ", "0123456789")) self.assertIsNotNone(re.search(rules["modelforge-github-opaque-token"]["regex"], classic)) opaque_body = classic.removeprefix("ghp_") for prefix in ("ghp_", "gho_", "ghu_", "ghr_"): with self.subTest(prefix=prefix): self.assertIsNotNone( re.search( rules["modelforge-github-opaque-token"]["regex"], prefix + opaque_body, ) ) installation_rule = rules["modelforge-github-installation-token"]["regex"] self.assertIsNotNone(re.search(installation_rule, "ghs_" + opaque_body)) self.assertIsNotNone( re.search(installation_rule, "ghs_" + "A" * 40 + "." + "B" * 40 + "." + "C" * 40) ) self.assertIsNotNone( re.search( rules["modelforge-github-fine-grained-pat"]["regex"], "github_pat_" + "A1" * 41, ) ) self.assertIsNone( re.search( rules["modelforge-github-opaque-token"]["regex"], "example-not-a-credential", ) ) install = _run_script( _named_step(self.workflow, "Install checksum-pinned security scanners") ) for fragment in ( "custom_formats=(ghp gho ghu ghs ghr github-pat)", "custom-${format}", "default-trigger", "custom-control", "default-control", '"AK", "IA", "A1B2C3D4E5F6G7H8"', "custom_trigger_code", "default_trigger_code", "trap cleanup_self_test EXIT", 'rm -r -- "${test_root}"', ): self.assertIn(fragment, install) self.assertNotIn("|| true", install) def test_pinned_gitleaks_binary_detects_temporary_history_and_benign_control(self) -> None: binary = os.environ.get("GITLEAKS_BIN") or shutil.which("gitleaks") if binary is None: self.skipTest("a Gitleaks binary is not available locally") version = subprocess.run( [binary, "version"], capture_output=True, check=False, text=True ) if "8.30.1" not in version.stdout + version.stderr: self.skipTest("available Gitleaks binary is not the pinned 8.30.1 build") with tempfile.TemporaryDirectory() as temporary: root = Path(temporary) custom_names = ("ghp", "gho", "ghu", "ghs", "ghr", "github-pat") repositories = { name: root / name for name in (*custom_names, "default-trigger", "control") } for repository in repositories.values(): repository.mkdir() subprocess.run(["git", "-C", str(repository), "init", "-q"], check=True) subprocess.run( ["git", "-C", str(repository), "config", "user.name", "security-gate"], check=True, ) subprocess.run( [ "git", "-C", str(repository), "config", "user.email", "security-gate@invalid.example", ], check=True, ) body = "".join(("aBcDeFgHiJkLmNoPqRsTuVwXyZ", "0123456789")) aws_key = "".join(("AK", "IA", "A1B2C3D4E5F6G7H8")) custom_tokens = { name: "".join(("gh", f"{name[-1]}_", body)) for name in custom_names if name != "github-pat" } custom_tokens["github-pat"] = "".join(("github", "_pat_", "A" * 82)) trigger_values = { **{ name: f'token = "{token}"\n' for name, token in custom_tokens.items() }, "default-trigger": f'key = "{aws_key}"\n', } for name, value in trigger_values.items(): trigger_path = repositories[name] / "credential.txt" trigger_path.write_text(value, encoding="utf-8") _git(repositories[name], "add", ".") _git(repositories[name], "commit", "-qm", "trigger") trigger_path.write_text("credential removed\n", encoding="utf-8") _git(repositories[name], "add", ".") _git(repositories[name], "commit", "-qm", "remove") (repositories["control"] / "control.txt").write_text( 'token = "example-not-a-credential"\n', encoding="utf-8" ) subprocess.run(["git", "-C", str(repositories["control"]), "add", "."], check=True) subprocess.run( ["git", "-C", str(repositories["control"]), "commit", "-qm", "control"], check=True, ) def detect(name: str, *, custom: bool) -> subprocess.CompletedProcess[str]: command = [binary, "git"] if custom: command.extend(("--config", str(GITLEAKS_CONFIG))) command.extend( ( "--log-opts=--all", "--redact=100", "--report-format", "json", "--report-path", str(root / f"{name}-{'custom' if custom else 'default'}.json"), "--no-banner", str(repositories[name]), ) ) return subprocess.run( command, capture_output=True, check=False, text=True, timeout=60, ) trigger_custom = [detect(name, custom=True) for name in custom_names] trigger_default = detect("default-trigger", custom=False) control_default = detect("control", custom=False) control_custom = detect("control", custom=True) self.assertEqual([result.returncode for result in trigger_custom], [1] * 6) self.assertEqual(trigger_default.returncode, 1) self.assertEqual(control_default.returncode, 0) self.assertEqual(control_custom.returncode, 0) def test_dual_pass_gitleaks_matches_exact_historical_false_positive_hashes(self) -> None: binary = os.environ.get("GITLEAKS_BIN") or shutil.which("gitleaks") if binary is None: self.skipTest("a Gitleaks binary is not available locally") version = subprocess.run( [binary, "version"], capture_output=True, check=False, text=True ) if "8.30.1" not in version.stdout + version.stderr: self.skipTest("available Gitleaks binary is not the pinned 8.30.1 build") with tempfile.TemporaryDirectory() as temporary: root = Path(temporary) repository = root / "config-free-history" subprocess.run( ["git", "clone", "--quiet", "--no-local", str(REPOSITORY_ROOT), str(repository)], check=True, timeout=60, ) (repository / ".gitleaks.toml").unlink(missing_ok=True) def scan(*, custom: bool) -> tuple[subprocess.CompletedProcess[str], Path]: output = root / f"repository-{'custom' if custom else 'default'}.json" command = [binary, "git"] if custom: command.extend(("--config", str(GITLEAKS_CONFIG))) command.extend( ( "--log-opts=--all", "--redact=100", "--report-format", "json", "--report-path", str(output), "--no-banner", str(repository), ) ) result = subprocess.run( command, capture_output=True, check=False, text=True, timeout=120, ) return result, output default_result, default_report = scan(custom=False) custom_result, custom_report = scan(custom=True) self.assertEqual(default_result.returncode, 1) self.assertEqual(custom_result.returncode, 0) combined = root / "combined.json" _write_json( combined, [ *json.loads(default_report.read_text(encoding="utf-8")), *json.loads(custom_report.read_text(encoding="utf-8")), ], ) sanitized = root / "sanitized.json" gate.sanitize_gitleaks(combined, sanitized) findings = json.loads(sanitized.read_text(encoding="utf-8"))["findings"] actual = {(item["rule_id"], item["fingerprint_sha256"]) for item in findings} registered = json.loads(GITLEAKS_FALSE_POSITIVES.read_text(encoding="utf-8")) expected = { (item["rule_id"], item["fingerprint_sha256"]) for item in registered } self.assertEqual(actual, expected) def test_risk_registry_is_exact_bounded_and_tracked(self) -> None: registry = json.loads(ACCEPTED_RISKS.read_text(encoding="utf-8")) self.assertEqual(registry["schema_version"], 1) risks = registry["accepted_risks"] self.assertEqual(risks, []) identities = { ( entry["scanner"], entry["target"], entry["advisory_id"], entry["package"], entry["installed_version"], ) for entry in risks } self.assertEqual(len(identities), len(risks)) for entry in risks: self.assertEqual(entry["scanner"], "trivy") self.assertEqual(entry["target"], "node-agent") self.assertEqual(entry["classification"], "ACCEPTED_TEMPORARY") self.assertEqual(entry["owner"], "ModelForge security owner") self.assertEqual(entry["accepted_on"], "2026-09-02") self.assertEqual(entry["expires_on"], "2026-10-02") self.assertEqual( entry["tracking"], "https://git.example.com/example/modelforge/issues/24", ) self.assertNotIn("*", "|".join(str(value) for value in entry.values())) def test_false_positives_use_only_exact_hashes(self) -> None: false_positives = json.loads(GITLEAKS_FALSE_POSITIVES.read_text(encoding="utf-8")) self.assertEqual(len(false_positives), 2) for entry in false_positives: self.assertEqual(entry["classification"], "FALSE_POSITIVE") self.assertRegex(entry["fingerprint_sha256"], r"^[0-9a-f]{64}$") self.assertNotIn("path", entry) self.assertNotIn("commit", entry) self.assertNotIn("secret", entry) def test_precheckout_fail_evidence_survives_workspace_cleanup(self) -> None: script = _run_script(_named_step(self.workflow, "Prepare fail-closed security evidence")) for fragment in ( "GITHUB_RUN_ID", "GITHUB_RUN_ATTEMPT", "GITHUB_SHA", "mktemp -d", "XXXXXXXX", "umask 077", "stat -c '%u'", "stat -c '%a'", ): self.assertIn(fragment, script) self.assertNotIn("mkdir -p", script) sha = "a" * 40 execution_script = "\n".join( ( "set -euo pipefail", "test_base=$(mktemp -d /tmp/modelforge-root-policy.XXXXXXXX)", "cleanup_test() { code=$?; trap - EXIT; rm -r -- \"${test_base}\"; exit \"${code}\"; }", "trap cleanup_test EXIT", "export RUNNER_TEMP=${test_base}", "export GITHUB_ENV=${test_base}/github-env", f"export GITHUB_SHA={sha}", "export GITHUB_REF=refs/pull/1/head", "export GITHUB_RUN_ID=synthetic-run", "export GITHUB_RUN_ATTEMPT=7", "export GITHUB_SERVER_URL=http://192.0.2.10:3000", "export GITHUB_REPOSITORY=Jens/ITWorx-ModelForge", "export SECURITY_PROFILE=full", "poison_target=${test_base}/poison-target", "mkdir \"${poison_target}\"", "ln -s \"${poison_target}\" \"${test_base}/modelforge-security-evidence-synthetic-run-7-" + sha + "\"", script, "[[ ${SOURCE_REPOSITORY} == " "https://git.example.com/example/modelforge ]]", "first_reports=${REPORTS}", "first_tools=${SECURITY_TOOLS}", "[[ ${first_reports} == ${test_base}/modelforge-security-evidence-synthetic-run-7-" + sha + ".* ]]", "[[ -d ${first_reports} && ! -L ${first_reports} ]]", "[[ $(stat -c '%a' \"${first_reports}\") == 700 ]]", "printf '99\\n' > \"${first_reports}/status/stale.exit\"", "workspace=${test_base}/workspace", "mkdir \"${workspace}\"", "printf 'ephemeral\\n' > \"${workspace}/checkout-marker\"", "rm \"${workspace}/checkout-marker\"", "python3 - \"${first_reports}/security-gate-result.json\" <<'PY'", "import json, sys", "from pathlib import Path", "value = json.loads(Path(sys.argv[1]).read_text(encoding='utf-8'))", "assert value['decision'] == 'FAIL' and value['profile'] == 'full'", "PY", script, "second_reports=${REPORTS}", "second_tools=${SECURITY_TOOLS}", "[[ ${second_reports} != ${first_reports} && ${second_tools} != ${first_tools} ]]", "[[ ! -e ${second_reports}/status/stale.exit ]]", "[[ -f ${first_reports}/security-gate-result.json ]]", ) ) result = subprocess.run( ["bash"], input=execution_script.encode(), cwd=REPOSITORY_ROOT, capture_output=True, check=False, ) self.assertEqual(result.returncode, 0, result.stderr.decode(errors="replace")) def test_precheckout_rejects_unknown_insecure_source_server(self) -> None: script = _run_script(_named_step(self.workflow, "Prepare fail-closed security evidence")) execution_script = "\n".join( ( "set -euo pipefail", "test_base=$(mktemp -d /tmp/modelforge-source-policy.XXXXXXXX)", "cleanup_test() { code=$?; trap - EXIT; rm -r -- \"${test_base}\"; " "exit \"${code}\"; }", "trap cleanup_test EXIT", "export RUNNER_TEMP=${test_base}", "export GITHUB_ENV=${test_base}/github-env", f"export GITHUB_SHA={'a' * 40}", "export GITHUB_REF=refs/pull/1/head", "export GITHUB_RUN_ID=synthetic-run", "export GITHUB_RUN_ATTEMPT=1", "export GITHUB_SERVER_URL=http://untrusted.invalid", "export GITHUB_REPOSITORY=Jens/ITWorx-ModelForge", "export SECURITY_PROFILE=full", script, ) ) result = subprocess.run( ["bash"], input=execution_script.encode(), cwd=REPOSITORY_ROOT, capture_output=True, check=False, ) self.assertNotEqual(result.returncode, 0) self.assertIn("Untrusted non-HTTPS Gitea server URL", result.stderr.decode()) def test_source_inventory_reads_every_optional_group_and_first_party_package(self) -> None: inventory = gate.build_source_inventory(REPOSITORY_ROOT) components = {component["target"]: component for component in inventory["components"]} self.assertEqual(set(components), {"backend", "node-agent", "runtime-worker"}) self.assertEqual(components["backend"]["optional_groups"], ["dev"]) self.assertEqual(components["node-agent"]["optional_groups"], ["dev"]) self.assertEqual(components["runtime-worker"]["optional_groups"], ["dev", "inference"]) own = { item["package"] for item in inventory["first_party_coverage"] if item["consumer_target"] == item["provider_target"] } self.assertEqual( own, {"modelforge-api", "modelforge-node-agent", "modelforge-runtime-worker"} ) def test_pep517_requirements_and_build_toolchain_are_exact_and_hash_locked(self) -> None: inventory = gate.build_source_inventory(REPOSITORY_ROOT) components = {component["target"]: component for component in inventory["components"]} for target in gate.SOURCE_COMPONENTS: by_name = { requirement["name"]: requirement for requirement in components[target]["requirements"] } with self.subTest(target=target): backend_requirement = next(iter(gate.PEP517_BUILD_SYSTEMS[target][1])) self.assertIn("build-system", by_name[backend_requirement]["groups"]) for name, version in gate.BUILD_TOOLCHAIN_PINS.items(): self.assertEqual(by_name[name]["version"], version) self.assertIn("build-toolchain", by_name[name]["groups"]) lock_lines = [ line.strip() for line in (REPOSITORY_ROOT / gate.BUILD_TOOLCHAIN_LOCK) .read_text(encoding="utf-8") .splitlines() if line.strip() and not line.lstrip().startswith("#") ] self.assertEqual(len(lock_lines), len(gate.BUILD_TOOLCHAIN_PINS)) self.assertTrue(all("--hash=sha256:" in line for line in lock_lines)) msgpack_line = next(line for line in lock_lines if line.startswith("msgpack==")) self.assertEqual(msgpack_line.count("--hash=sha256:"), 14) with tempfile.TemporaryDirectory() as temporary: repository = Path(temporary) _copy_source_gate_inputs(repository) manifest = repository / "backend/pyproject.toml" manifest.write_text( manifest.read_text(encoding="utf-8").replace( ' "hatchling==1.32.4",', ' "hatchling",' ), encoding="utf-8", ) with self.assertRaisesRegex(gate.GateInputError, "not pinned"): gate.build_source_inventory(repository) with tempfile.TemporaryDirectory() as temporary: repository = Path(temporary) _copy_source_gate_inputs(repository) dockerfile = repository / "node-agent/Dockerfile" dockerfile.write_text( dockerfile.read_text(encoding="utf-8").replace( "--require-hashes", "", 1 ), encoding="utf-8", ) with self.assertRaisesRegex(gate.GateInputError, "hash-locked bootstrap"): gate.build_source_inventory(repository) with tempfile.TemporaryDirectory() as temporary: repository = Path(temporary) _copy_source_gate_inputs(repository) dockerfile = repository / "backend/Dockerfile" dockerfile.write_text( dockerfile.read_text(encoding="utf-8").replace( "--no-build-isolation", "", 1 ), encoding="utf-8", ) with self.assertRaisesRegex(gate.GateInputError, "disable build isolation"): gate.build_source_inventory(repository) with tempfile.TemporaryDirectory() as temporary: repository = Path(temporary) _copy_source_gate_inputs(repository) dockerfile = repository / "node-agent/Dockerfile" dockerfile.write_text( dockerfile.read_text(encoding="utf-8").replace( "--ignore-installed", "", 1 ), encoding="utf-8", ) with self.assertRaisesRegex(gate.GateInputError, "runtime dependencies"): gate.build_source_inventory(repository) with tempfile.TemporaryDirectory() as temporary: repository = Path(temporary) _copy_source_gate_inputs(repository) dockerfile = repository / "frontend/Dockerfile" dockerfile.write_text( dockerfile.read_text(encoding="utf-8").replace( gate.CANONICAL_SOURCE_REPOSITORY, "https://git.example.com/example/modelforge.git", ), encoding="utf-8", ) with self.assertRaisesRegex(gate.GateInputError, "source label bound"): gate.build_source_inventory(repository) with tempfile.TemporaryDirectory() as temporary: repository = Path(temporary) _copy_source_gate_inputs(repository) manifest_files = [] for relative_path in gate.IMAGE_DOCKERFILES.values(): dockerfile = repository / relative_path rendered = dockerfile.read_text(encoding="utf-8").replace( gate.CANONICAL_SOURCE_REPOSITORY, gate.PUBLIC_SOURCE_LABEL_PLACEHOLDER, ) dockerfile.write_text(rendered, encoding="utf-8") rendered_bytes = rendered.encode("utf-8") manifest_files.append( { "path": relative_path, "bytes": len(rendered_bytes), "sha256": hashlib.sha256(rendered_bytes).hexdigest(), } ) _write_json( repository / gate.PUBLIC_SOURCE_MANIFEST, { "schemaVersion": 1, "sourceRevision": "a" * 40, "sourceRepository": gate.PUBLIC_SOURCE_LABEL_PLACEHOLDER, "files": manifest_files, }, ) exporter = (REPOSITORY_ROOT / "scripts" / "export-public-source.mjs").read_text( encoding="utf-8" ) self.assertIn(f'"{gate.PUBLIC_SOURCE_LABEL_PLACEHOLDER}"', exporter) gate.build_source_inventory(repository) manifest_path = repository / gate.PUBLIC_SOURCE_MANIFEST manifest = json.loads(manifest_path.read_text(encoding="utf-8")) manifest["sourceRepository"] = "https://attacker.invalid/repository" _write_json(manifest_path, manifest) with self.assertRaisesRegex(gate.GateInputError, "approved sanitized identity"): gate.build_source_inventory(repository) manifest["sourceRepository"] = gate.PUBLIC_SOURCE_LABEL_PLACEHOLDER _write_json(manifest_path, manifest) dockerfile = repository / "frontend/Dockerfile" dockerfile.write_text( dockerfile.read_text(encoding="utf-8").replace( gate.PUBLIC_SOURCE_LABEL_PLACEHOLDER, "https://attacker.invalid/repository" ), encoding="utf-8", ) with self.assertRaisesRegex(gate.GateInputError, "does not bind the current"): gate.build_source_inventory(repository) def test_evaluator_rejects_a_non_https_repository_identity(self) -> None: request = gate.EvaluationRequest( reports_dir=Path("reports"), accepted_risks=Path("accepted-risks.json"), gitleaks_false_positives=Path("false-positives.json"), repository=REPOSITORY_ROOT, commit="a" * 40, ref="refs/heads/master", run_id="test", source_repository="http://git.example.com/Jens/ITWorx-ModelForge", profile="full", output=Path("reports/result.json"), as_of=date(2026, 8, 30), evaluation_time=datetime(2026, 8, 30, 0, 10, tzinfo=UTC), ) with self.assertRaisesRegex(gate.GateInputError, "canonical HTTPS"): gate.Evaluator(request) def test_source_inventory_emits_exact_pep508_extras(self) -> None: with tempfile.TemporaryDirectory() as temporary: root = Path(temporary) gate.collect_source_inventory( REPOSITORY_ROOT, root / "source-inventory.json", root / "source-requirements", ) backend = (root / "source-requirements" / "backend.txt").read_text( encoding="utf-8" ) self.assertIn("uvicorn[standard]==0.52.4\n", backend) self.assertIn("psycopg[binary]==3.3.4\n", backend) self.assertIn("hatchling==1.32.4\n", backend) self.assertIn("editables==0.5\n", backend) self.assertIn("msgpack==1.2.2\n", backend) self.assertIn("pip==26.2.1\n", backend) self.assertNotIn("\nuvicorn==0.52.4\n", f"\n{backend}") self.assertNotIn("\npsycopg==3.3.4\n", f"\n{backend}") def test_source_inventory_rejects_identical_extra_declarations_before_dedup(self) -> None: declarations = ("uvicorn[standard]==0.52.4", "psycopg[binary]==3.3.4") with tempfile.TemporaryDirectory() as temporary: temporary_root = Path(temporary) for attack_index, declaration in enumerate(declarations): with self.subTest(declaration=declaration): repository = temporary_root / f"attack-{attack_index}" _copy_source_gate_inputs(repository) manifest = repository / "backend" / "pyproject.toml" text = manifest.read_text(encoding="utf-8") declaration_line = f' "{declaration}",' first_line = text.splitlines().index(declaration_line) + 1 manifest.write_text( text.replace( declaration_line, f"{declaration_line}\n{declaration_line}", 1, ), encoding="utf-8", ) with self.assertRaises(gate.GateInputError) as raised: gate.build_source_inventory(repository) message = str(raised.exception) self.assertIn("identical manifest declaration", message) self.assertIn(f"backend/pyproject.toml:{first_line}", message) self.assertIn(f"backend/pyproject.toml:{first_line + 1}", message) self.assertIn("project.dependencies", message) def test_all_security_shell_blocks_parse(self) -> None: names = ( "Prepare fail-closed security evidence", "Prove complete non-shallow history", "Install checksum-pinned security scanners", "Collect source security reports", "Build and scan exact shipping images", "Evaluate the complete security gate", ) for name in names: result = subprocess.run( ["bash", "-n"], input=_run_script(_named_step(self.workflow, name)).encode(), capture_output=True, check=False, ) with self.subTest(name=name): self.assertEqual(result.returncode, 0, result.stderr.decode(errors="replace")) class GateFixture: commit = "a" * 40 as_of = date(2026, 8, 30) evaluation_time = datetime(2026, 8, 30, 0, 10, tzinfo=UTC) source_repository = gate.CANONICAL_SOURCE_REPOSITORY def __init__(self, root: Path, *, profile: str = "full") -> None: self.root = root self.profile = profile self.reports = root / "reports" self.risks = root / "accepted-risks.json" self.false_positives = root / "gitleaks-false-positives.json" self.output = self.reports / "security-gate-result.json" (self.reports / "status").mkdir(parents=True) self.reports.chmod(0o700) (self.reports / "versions").mkdir() for status in gate.expected_status_codes(profile): (self.reports / "status" / f"{status}.exit").write_text("0\n", encoding="ascii") _write_json( self.reports / "git-history.json", { "schema_version": 1, "target": "full-history", "head": self.commit, "expected_commit": self.commit, "is_shallow": False, "head_commit_count": 151, "all_commit_count": 153, "root_commits": ["b" * 40], "remote_name": "origin", "remote_url": self.source_repository, "advertised_remote_heads": [ {"ref": "refs/heads/master", "object": "c" * 40} ], "local_remote_heads": [ {"ref": "refs/heads/master", "object": "c" * 40} ], "advertised_tags": [], "local_tags": [], "all_ref_count": 1, "all_objects_reachable": True, }, ) gate.collect_source_inventory( REPOSITORY_ROOT, self.reports / "source-inventory.json", self.reports / "source-requirements", ) source_inventory = json.loads( (self.reports / "source-inventory.json").read_text(encoding="utf-8") ) for component in source_inventory["components"]: packages = { item["name"]: item["version"] for item in component["requirements"] if not item["first_party"] } _write_json( self.reports / f"pip-source-{component['target']}.json", self.pip_report(packages), ) _write_json( self.reports / "gitleaks.json", { "schema_version": 1, "scanner": "gitleaks", "target": "full-history", "redaction_percent": 100, "findings": [], }, ) clean_npm = { "auditReportVersion": 2, "vulnerabilities": {}, "metadata": { "vulnerabilities": { "info": 0, "low": 0, "moderate": 0, "high": 0, "critical": 0, "total": 0, } }, } _write_json(self.reports / "npm-audit-all.json", clean_npm) _write_json(self.reports / "npm-audit-production.json", clean_npm) _write_json( self.reports / "trivy-fs.json", { "SchemaVersion": 2, "ArtifactName": str(REPOSITORY_ROOT.resolve()), "ArtifactType": "repository", "Results": [], }, ) image_packages = { "api": {"modelforge-api": PRODUCT_VERSION, "httpx": "0.28.1"}, "node-agent": { "modelforge-api": PRODUCT_VERSION, "modelforge-node-agent": PRODUCT_VERSION, "httpx": "0.28.1", }, "runtime-worker": { "modelforge-runtime-worker": PRODUCT_VERSION, **gate.IMAGE_NON_PYPI_AUDIT_EXCEPTIONS["runtime-worker"], "httpx": "0.28.1", }, } if profile == "security": identity_characters = { "api": "1", "web": "2", "node-agent": "3", "runtime-worker": "4", } for target in gate.IMAGE_TARGETS: image_id = "sha256:" + identity_characters[target] * 64 tag = f"modelforge-security-{target}:{self.commit}" identity = { "schema_version": 1, "target": target, "tag": tag, "image_id": image_id, "repo_tags": [tag], "repo_digests": [], "revision": self.commit, "source": self.source_repository, "version": PRODUCT_VERSION, "bundle_path": gate.WEB_BUNDLE_PATH if target == "web" else None, } _write_json(self.reports / f"image-identity-{target}.json", identity) _write_json( self.reports / f"trivy-image-{target}.json", { "SchemaVersion": 2, "ArtifactName": image_id, "ArtifactID": "sha256:" + "a" * 64, "ArtifactType": "container_image", "ScannedImageID": image_id, "Metadata": { "ImageID": image_id, "Reference": tag, "RepoTags": [tag], "RepoDigests": [], }, "Results": [], }, ) _write_json( self.reports / "web-bundle-origin.json", { "schema_version": 1, "target": "web", "image_id": "sha256:" + "2" * 64, "bundle_path": gate.WEB_BUNDLE_PATH, "bundle_sha256": "f" * 64, "bundle_file_count": 3, "provenance_path": gate.WEB_PROVENANCE_PATH, "provenance_sha256": "e" * 64, "served_provenance_sha256": "e" * 64, "provenance_binding_sha256": gate._web_provenance_binding( gate.WEB_SECURITY_ORIGIN ), "runtime_chunk_path": "assets/index.js", "runtime_chunk_sha256": "d" * 64, "api_origins": [gate.WEB_SECURITY_ORIGIN], "non_api_origins": list(gate.WEB_EXPECTED_NON_API_ORIGINS), }, ) for target, packages in image_packages.items(): _write_json( self.reports / f"image-inventory-{target}.json", { "schema_version": 1, "target": target, "python_version": "3.12.0", "packages": [ {"name": name, "version": version} for name, version in packages.items() ], }, ) _write_json( self.reports / f"pip-image-{target}.json", self.pip_report( { name: version for name, version in packages.items() if not name.startswith("modelforge-") and name not in gate.IMAGE_NON_PYPI_AUDIT_EXCEPTIONS.get(target, {}) } ), ) (self.reports / "versions" / "pip-audit.txt").write_text( "pip-audit 2.10.1\n", encoding="utf-8" ) (self.reports / "versions" / "gitleaks.txt").write_text( "8.30.1\n", encoding="utf-8" ) (self.reports / "versions" / "npm.txt").write_text("11.0.0\n", encoding="utf-8") _write_json( self.reports / "versions" / "trivy.json", { "Version": "0.74.0", "VulnerabilityDB": { "Version": 2, "UpdatedAt": "2026-08-30T00:00:00Z", "DownloadedAt": "2026-08-30T00:05:00Z", }, }, ) self.write_risks([]) _write_json(self.false_positives, []) @staticmethod def pip_report(packages: dict[str, str]) -> dict[str, Any]: return { "dependencies": [ {"name": name, "version": version, "vulns": []} for name, version in sorted(packages.items()) ], "fixes": [], } def evaluate(self) -> dict[str, Any]: return gate.evaluate_security_gate( gate.EvaluationRequest( reports_dir=self.reports, accepted_risks=self.risks, gitleaks_false_positives=self.false_positives, repository=REPOSITORY_ROOT, commit=self.commit, ref="refs/pull/1/head", run_id="synthetic-1", source_repository=self.source_repository, profile=self.profile, output=self.output, as_of=self.as_of, evaluation_time=self.evaluation_time, ) ) def write_risks(self, entries: list[dict[str, str]]) -> None: _write_json( self.risks, {"schema_version": gate.SCHEMA_VERSION, "accepted_risks": entries}, ) def add_pip_vulnerability( self, *, fix_versions: list[str] | None = None, duplicate: bool = False ) -> dict[str, str]: path = self.reports / "pip-source-backend.json" report = json.loads(path.read_text(encoding="utf-8")) dependency = report["dependencies"][0] vulnerability = {"id": "PYSEC-TEST-1", "fix_versions": fix_versions or []} dependency["vulns"] = [vulnerability, copy.deepcopy(vulnerability)] if duplicate else [vulnerability] _write_json(path, report) (self.reports / "status" / "pip-source-backend.exit").write_text("1\n") return { "scanner": "pip-audit", "target": "backend", "advisory_id": "PYSEC-TEST-1", "package": dependency["name"], "installed_version": dependency["version"], } def risk(self, identity: dict[str, str]) -> dict[str, str]: return { **identity, "classification": "ACCEPTED_TEMPORARY", "owner": "security-owner", "accepted_on": "2026-08-30", "expires_on": "2026-09-29", "tracking": "https://tracker.invalid/RC-DEL-02", "reason": "No upstream fix exists; bounded review is required before release.", } class SecurityGateEvaluatorTests(unittest.TestCase): def setUp(self) -> None: self.temporary = tempfile.TemporaryDirectory() self.fixture = GateFixture(Path(self.temporary.name)) def tearDown(self) -> None: self.temporary.cleanup() def assert_failed_with(self, result: dict[str, Any], advisory_id: str) -> None: self.assertEqual(result["decision"], "FAIL") self.assertIn(advisory_id, {item["advisory_id"] for item in result["blocked_findings"]}) def test_clean_complete_evidence_passes(self) -> None: result = self.fixture.evaluate() self.assertEqual(result["decision"], "PASS") self.assertEqual(result["blocked_findings"], []) self.assertEqual(result["expected_targets"], result["actual_targets"]) self.assertEqual(result["scanner_versions"]["trivy"], "0.74.0") self.assertEqual( result["vulnerability_database_times"]["trivy"]["updated_at"], "2026-08-30T00:00:00Z", ) self.assertEqual(result["profile"], "full") self.assertNotIn("image_scan", result["expected_targets"]) def test_security_profile_requires_and_accepts_source_plus_exact_images(self) -> None: fixture = GateFixture(Path(self.temporary.name) / "security", profile="security") result = fixture.evaluate() self.assertEqual(result["decision"], "PASS") self.assertEqual(result["expected_targets"], result["actual_targets"]) self.assertEqual(set(result["image_identities"]), set(gate.IMAGE_TARGETS)) (fixture.reports / "source-inventory.json").unlink() self.assert_failed_with(fixture.evaluate(), "GATE-SOURCE-INVENTORY") def test_runtime_worker_inventory_rejects_installers(self) -> None: fixture = GateFixture(Path(self.temporary.name) / "installers", profile="security") inventory = fixture.reports / "image-inventory-runtime-worker.json" report = json.loads(inventory.read_text(encoding="utf-8")) report["packages"].append({"name": "pip", "version": "26.0"}) _write_json(inventory, report) self.assert_failed_with(fixture.evaluate(), "GATE-IMAGE-FORBIDDEN-PACKAGE") def test_source_only_profile_defers_but_validates_image_risks(self) -> None: identity = { "scanner": "trivy", "target": "node-agent", "advisory_id": "CVE-2099-0001", "package": "base-package", "installed_version": "1.0", } risk = self.fixture.risk(identity) self.fixture.write_risks([risk]) result = self.fixture.evaluate() self.assertEqual(result["decision"], "PASS") self.assertEqual(result["applied_risks"], []) self.assertEqual(result["unused_or_invalid_risks"], []) risk["expires_on"] = "2026-08-30" self.fixture.write_risks([risk]) self.assert_failed_with(self.fixture.evaluate(), "GATE-RISK-INVALID") security = GateFixture(Path(self.temporary.name) / "unused-image", profile="security") security.write_risks([security.risk(identity)]) self.assert_failed_with(security.evaluate(), "GATE-RISK-UNUSED") def test_scanner_timeout_fails(self) -> None: (self.fixture.reports / "status" / "trivy-fs.exit").write_text("124\n") self.assert_failed_with(self.fixture.evaluate(), "GATE-STATUS-FAILED") def test_missing_empty_and_malformed_reports_fail(self) -> None: cases = (None, "", "{") path = self.fixture.reports / "npm-audit-all.json" original = path.read_text(encoding="utf-8") for replacement in cases: with self.subTest(replacement=replacement): if replacement is None: path.unlink() else: path.write_text(replacement, encoding="utf-8") self.assert_failed_with(self.fixture.evaluate(), "GATE-NPM-REPORT") path.write_text(original, encoding="utf-8") def test_component_gap_fails(self) -> None: path = self.fixture.reports / "pip-source-node-agent.json" report = json.loads(path.read_text(encoding="utf-8")) report["dependencies"].pop(0) _write_json(path, report) self.assert_failed_with(self.fixture.evaluate(), "GATE-PIP-COMPONENT-GAP") def test_source_direct_packages_are_exact_and_resolved_transitives_are_recorded(self) -> None: path = self.fixture.reports / "pip-source-backend.json" report = json.loads(path.read_text(encoding="utf-8")) report["dependencies"].extend( [ {"name": "starlette", "version": "0.47.3", "vulns": []}, {"name": "uvloop", "version": "0.21.0", "vulns": []}, {"name": "psycopg-binary", "version": "3.3.4", "vulns": []}, ] ) _write_json(path, report) result = self.fixture.evaluate() self.assertEqual(result["decision"], "PASS") for transitive in ( {"name": "starlette", "version": "0.47.3"}, {"name": "uvloop", "version": "0.21.0"}, {"name": "psycopg-binary", "version": "3.3.4"}, ): self.assertIn(transitive, result["resolved_transitive_packages"]["backend"]) report["dependencies"] = [ dependency for dependency in report["dependencies"] if gate._normalize_name(dependency["name"]) != "fastapi" ] _write_json(path, report) self.assert_failed_with(self.fixture.evaluate(), "GATE-PIP-COMPONENT-GAP") def test_only_pip_audit_packaging_tool_omissions_are_accepted(self) -> None: source = self.fixture.reports / "pip-source-runtime-worker.json" report = json.loads(source.read_text(encoding="utf-8")) report["dependencies"] = [ dependency for dependency in report["dependencies"] if gate._normalize_name(dependency["name"]) not in gate.PIP_AUDIT_SOURCE_PACKAGING_TOOL_OMISSIONS ] _write_json(source, report) self.assertEqual(self.fixture.evaluate()["decision"], "PASS") report["dependencies"] = [ dependency for dependency in report["dependencies"] if gate._normalize_name(dependency["name"]) != "sentencepiece" ] _write_json(source, report) self.assert_failed_with(self.fixture.evaluate(), "GATE-PIP-COMPONENT-GAP") image_fixture = GateFixture( Path(self.temporary.name) / "packaging-tool-image", profile="security" ) image = image_fixture.reports / "pip-image-runtime-worker.json" image_report = json.loads(image.read_text(encoding="utf-8")) image_report["dependencies"] = [ dependency for dependency in image_report["dependencies"] if gate._normalize_name(dependency["name"]) not in gate.PIP_AUDIT_PACKAGING_TOOL_OMISSIONS ] _write_json(image, image_report) self.assertEqual(image_fixture.evaluate()["decision"], "PASS") def test_source_pep508_extra_omission_attack_fails(self) -> None: requirements = self.fixture.reports / "source-requirements" / "backend.txt" retained = requirements.read_text(encoding="utf-8") self.assertIn("uvicorn[standard]==0.52.4", retained) self.assertIn("psycopg[binary]==3.3.4", retained) poisoned = retained.replace("uvicorn[standard]", "uvicorn").replace( "psycopg[binary]", "psycopg" ) requirements.write_text(poisoned, encoding="utf-8") self.assert_failed_with(self.fixture.evaluate(), "GATE-SOURCE-REQUIREMENTS") def test_identical_duplicate_pip_row_is_malformed(self) -> None: path = self.fixture.reports / "pip-source-backend.json" report = json.loads(path.read_text(encoding="utf-8")) report["dependencies"].append(copy.deepcopy(report["dependencies"][0])) _write_json(path, report) self.assert_failed_with(self.fixture.evaluate(), "GATE-PIP-DUPLICATE") def test_evidence_symlink_redirection_fails_before_hashing(self) -> None: outside = self.fixture.root / "outside-evidence" outside.mkdir() redirect = self.fixture.reports / "redirected-evidence" if os.name == "nt": subprocess.run( ["cmd.exe", "/d", "/c", "mklink", "/J", str(redirect), str(outside)], capture_output=True, check=True, text=True, timeout=30, ) else: redirect.symlink_to(outside, target_is_directory=True) result = self.fixture.evaluate() self.assert_failed_with(result, "GATE-EVIDENCE-PATH") self.assertEqual(result["report_hashes"], {}) def test_evidence_metadata_rejects_foreign_owner_and_writable_modes(self) -> None: def metadata(mode: int, uid: int) -> Any: return SimpleNamespace( st_dev=1, st_ino=2, st_mode=mode, st_uid=uid, st_size=1, st_mtime_ns=1, st_ctime_ns=1, st_file_attributes=0, ) with mock.patch.object(os, "geteuid", return_value=1000, create=True): with self.assertRaisesRegex(gate.GateInputError, "not owned"): gate._validate_evidence_metadata( metadata(stat.S_IFREG | 0o600, 65534), path="foreign.json", expected_kind="file", enforce_posix_permissions=True, ) for mode, kind, root in ( (stat.S_IFDIR | 0o777, "directory", True), (stat.S_IFREG | 0o666, "file", False), ): with self.subTest(mode=oct(mode)): with self.assertRaisesRegex(gate.GateInputError, "permissions|writable"): gate._validate_evidence_metadata( metadata(mode, 1000), path="unsafe", expected_kind=kind, root=root, enforce_posix_permissions=True, ) def test_evidence_lstat_open_race_fails_before_replacement_is_read(self) -> None: root = self.fixture.root / "race-reports" root.mkdir(mode=0o700) root.chmod(0o700) victim = root / "report.json" victim.write_bytes(b"inside-safe") outside = self.fixture.root / "outside-secret.json" outside.write_bytes(b"outside-must-not-be-consumed") replaced = False def replace_after_lstat(event: str, path: Path) -> None: nonlocal replaced if event == "after_lstat" and path == victim and not replaced: replaced = True os.replace(outside, victim) with mock.patch.object( gate.SecureEvidenceSnapshot, "_read_fd", wraps=gate.SecureEvidenceSnapshot._read_fd, ) as reader: with self.assertRaisesRegex(gate.GateInputError, "changed before open"): gate.SecureEvidenceSnapshot( root, root / "result.json", race_hook=replace_after_lstat ) self.assertTrue(replaced) reader.assert_not_called() @unittest.skipUnless(os.name == "posix", "open-handle replacement semantics are Linux-only") def test_evidence_after_open_swap_reads_only_original_descriptor_then_fails(self) -> None: root = self.fixture.root / "open-race-reports" root.mkdir(mode=0o700) root.chmod(0o700) victim = root / "report.json" victim.write_bytes(b"inside-safe") outside = self.fixture.root / "outside-open-race.json" outside.write_bytes(b"outside-must-not-be-consumed") consumed: list[bytes] = [] original_reader = gate.SecureEvidenceSnapshot._read_fd def swap_after_open(event: str, path: Path) -> None: if event == "after_open" and path == victim: os.replace(outside, victim) def read_descriptor(file_descriptor: int) -> bytes: value = original_reader(file_descriptor) consumed.append(value) return value with mock.patch.object( gate.SecureEvidenceSnapshot, "_read_fd", side_effect=read_descriptor ): with self.assertRaisesRegex(gate.GateInputError, "changed during read"): gate.SecureEvidenceSnapshot( root, root / "result.json", race_hook=swap_after_open ) self.assertEqual(consumed, [b"inside-safe"]) def test_evidence_directory_substitution_after_snapshot_is_rejected(self) -> None: root = self.fixture.root / "snapshot-reports" nested = root / "nested" nested.mkdir(parents=True) root.chmod(0o700) (nested / "report.json").write_bytes(b"inside-safe") outside = self.fixture.root / "outside-directory" outside.mkdir() (outside / "report.json").write_bytes(b"outside-must-not-be-consumed") snapshot = gate.SecureEvidenceSnapshot(root, root / "result.json") (nested / "report.json").unlink() nested.rmdir() if os.name == "nt": subprocess.run( ["cmd.exe", "/d", "/c", "mklink", "/J", str(nested), str(outside)], capture_output=True, check=True, text=True, timeout=30, ) else: nested.symlink_to(outside, target_is_directory=True) with self.assertRaisesRegex( gate.GateInputError, "identity changed|not a real|changed after secure capture" ): snapshot.read_bytes("nested/report.json") def test_evidence_in_place_mutation_after_snapshot_is_rejected(self) -> None: root = self.fixture.root / "mutation-reports" root.mkdir(mode=0o700) root.chmod(0o700) report = root / "report.json" report.write_bytes(b"inside-safe") snapshot = gate.SecureEvidenceSnapshot(root, root / "result.json") report.write_bytes(b"outside-poison-is-longer") with self.assertRaisesRegex(gate.GateInputError, "changed after secure capture"): snapshot.read_bytes("report.json") def test_record_history_rejects_nonshallow_single_branch_clone(self) -> None: with tempfile.TemporaryDirectory() as temporary: root = Path(temporary) origin = root / "origin.git" seed = root / "seed" partial = root / "partial" complete = root / "complete" origin.mkdir() seed.mkdir() _git(origin, "init", "--bare", "--initial-branch=main") _git(seed, "init", "--initial-branch=main") _git(seed, "config", "user.name", "security-gate") _git(seed, "config", "user.email", "security-gate@invalid.example") (seed / "tracked.txt").write_text("history proof\n", encoding="utf-8") _git(seed, "add", "tracked.txt") _git(seed, "commit", "-m", "initial") _git(seed, "branch", "retained-remote-branch") _git(seed, "tag", "retained-tag") _git(seed, "remote", "add", "origin", str(origin)) _git(seed, "push", "origin", "--all") _git(seed, "push", "origin", "--tags") subprocess.run( [ "git", "clone", "--single-branch", "--branch", "main", "--no-tags", str(origin), str(partial), ], capture_output=True, check=True, text=True, timeout=30, ) head = _git(partial, "rev-parse", "HEAD").stdout.strip() with self.assertRaisesRegex(gate.GateInputError, "advertised remote heads"): gate.record_history(partial, head, root / "partial-history.json") subprocess.run( ["git", "clone", str(origin), str(complete)], capture_output=True, check=True, text=True, timeout=30, ) complete_head = _git(complete, "rev-parse", "HEAD").stdout.strip() output = root / "complete-history.json" gate.record_history(complete, complete_head, output) report = json.loads(output.read_text(encoding="utf-8")) self.assertEqual(len(report["advertised_remote_heads"]), 2) self.assertEqual(report["advertised_tags"][0]["ref"], "refs/tags/retained-tag") def test_shallow_history_fails(self) -> None: path = self.fixture.reports / "git-history.json" report = json.loads(path.read_text(encoding="utf-8")) report["is_shallow"] = True _write_json(path, report) self.assert_failed_with(self.fixture.evaluate(), "GATE-HISTORY-INCOMPLETE") def test_history_origin_and_ref_namespace_are_exact(self) -> None: attacks: tuple[tuple[str, Any], ...] = ( ("remote_url", "https://attacker.invalid/repository"), ( "advertised_remote_heads", [{"ref": "refs/pull/1/head", "object": "c" * 40}], ), ("advertised_remote_heads", []), ) for index, (field, value) in enumerate(attacks): with self.subTest(field=field, value=value): fixture = GateFixture(self.fixture.root / f"history-attack-{index}") path = fixture.reports / "git-history.json" report = json.loads(path.read_text(encoding="utf-8")) report[field] = value if field == "advertised_remote_heads" and value: report["local_remote_heads"] = copy.deepcopy(value) _write_json(path, report) self.assert_failed_with(fixture.evaluate(), "GATE-HISTORY-INCOMPLETE") def test_unclassified_gitleaks_finding_fails_without_secret_material(self) -> None: fingerprint = "c" * 64 _write_json( self.fixture.reports / "gitleaks.json", { "schema_version": 1, "scanner": "gitleaks", "target": "full-history", "redaction_percent": 100, "findings": [ {"rule_id": "generic-api-key", "fingerprint_sha256": fingerprint} ], }, ) (self.fixture.reports / "status" / "gitleaks.exit").write_text("1\n") result = self.fixture.evaluate() self.assert_failed_with(result, "generic-api-key") serialized = json.dumps(result) for forbidden in ('"Secret"', '"Match"', '"Line"', '"author"', '"email"'): self.assertNotIn(forbidden, serialized) def test_gitleaks_sanitizer_and_evaluator_reject_metadata_fields(self) -> None: raw = self.fixture.root / "raw-gitleaks.json" sanitized = self.fixture.root / "sanitized-gitleaks.json" _write_json( raw, [ { "RuleID": "synthetic-rule", "Fingerprint": "synthetic-fingerprint", "Secret": "must-not-survive", "Match": "must-not-survive", "File": "private/path.txt", "Commit": "b" * 40, "Author": "private-author", } ], ) gate.sanitize_gitleaks(raw, sanitized) report = json.loads(sanitized.read_text(encoding="utf-8")) self.assertEqual(set(report), gate.GITLEAKS_REPORT_FIELDS) self.assertEqual(set(report["findings"][0]), gate.GITLEAKS_FINDING_FIELDS) serialized = json.dumps(report) for forbidden in ( "must-not-survive", "private/path.txt", "private-author", "synthetic-fingerprint", ): self.assertNotIn(forbidden, serialized) unsafe = json.loads(raw.read_text(encoding="utf-8")) unsafe[0]["RuleID"] = "../private/path" _write_json(raw, unsafe) with self.assertRaisesRegex(gate.GateInputError, "safe rule metadata"): gate.sanitize_gitleaks(raw, sanitized) for field, value in ( ("path", "private/path.txt"), ("commit", "b" * 40), ("secret", "must-not-survive"), ("unknown", "unexpected"), ): with self.subTest(field=field): fixture = GateFixture(self.fixture.root / f"gitleaks-{field}") path = fixture.reports / "gitleaks.json" poisoned = json.loads(path.read_text(encoding="utf-8")) poisoned["findings"] = [ { "rule_id": "synthetic-rule", "fingerprint_sha256": "c" * 64, field: value, } ] _write_json(path, poisoned) fixture_status = fixture.reports / "status" / "gitleaks.exit" fixture_status.write_text("1\n", encoding="ascii") self.assert_failed_with(fixture.evaluate(), "GATE-GITLEAKS-SCHEMA") def test_every_pip_and_npm_vulnerability_fails(self) -> None: self.fixture.add_pip_vulnerability() self.assert_failed_with(self.fixture.evaluate(), "PYSEC-TEST-1") self.fixture = GateFixture(Path(self.temporary.name) / "npm") npm = { "auditReportVersion": 2, "vulnerabilities": { "react": { "name": "react", "severity": "high", "via": [ {"source": 1, "url": "https://github.com/advisories/GHSA-test-0001"} ], "fixAvailable": False, } }, "metadata": {"vulnerabilities": {"high": 1, "total": 1}}, } _write_json(self.fixture.reports / "npm-audit-all.json", npm) (self.fixture.reports / "status" / "npm-audit-all.exit").write_text("1\n") self.assert_failed_with(self.fixture.evaluate(), "GHSA-test-0001") def test_trivy_unknown_high_and_critical_fail_but_medium_does_not(self) -> None: path = self.fixture.reports / "trivy-fs.json" for severity, expected in ( ("UNKNOWN", "CVE-UNKNOWN"), ("HIGH", "CVE-HIGH"), ("CRITICAL", "CVE-CRITICAL"), ): with self.subTest(severity=severity): _write_json( path, { "SchemaVersion": 2, "ArtifactName": str(REPOSITORY_ROOT.resolve()), "ArtifactType": "repository", "Results": [ { "Target": ".", "Vulnerabilities": [ { "VulnerabilityID": expected, "PkgName": "package", "InstalledVersion": "1.0", "FixedVersion": "", "Severity": severity, } ], } ], }, ) self.assert_failed_with(self.fixture.evaluate(), expected) _write_json( path, { "SchemaVersion": 2, "ArtifactName": str(REPOSITORY_ROOT.resolve()), "ArtifactType": "repository", "Results": [ { "Target": ".", "Vulnerabilities": [ { "VulnerabilityID": "CVE-MEDIUM", "PkgName": "package", "InstalledVersion": "1.0", "Severity": "MEDIUM", } ] } ], }, ) self.assertEqual(self.fixture.evaluate()["decision"], "PASS") def test_trivy_filesystem_and_image_reports_are_bound_to_exact_targets(self) -> None: filesystem = self.fixture.reports / "trivy-fs.json" report = json.loads(filesystem.read_text(encoding="utf-8")) report["ArtifactType"] = "filesystem" _write_json(filesystem, report) self.assert_failed_with(self.fixture.evaluate(), "GATE-TRIVY-TARGET") report["ArtifactType"] = "repository" report["ArtifactName"] = "swapped-target" _write_json(filesystem, report) self.assert_failed_with(self.fixture.evaluate(), "GATE-TRIVY-TARGET") fixture = GateFixture(Path(self.temporary.name) / "image-target", profile="security") api_report = fixture.reports / "trivy-image-api.json" swapped = json.loads( (fixture.reports / "trivy-image-web.json").read_text(encoding="utf-8") ) _write_json(api_report, swapped) self.assert_failed_with(fixture.evaluate(), "GATE-TRIVY-TARGET") fixture = GateFixture(Path(self.temporary.name) / "nested-target", profile="security") api_report = fixture.reports / "trivy-image-api.json" nested = json.loads(api_report.read_text(encoding="utf-8")) nested["Results"] = [ { "Target": "sha256:" + "9" * 64, "ComponentTarget": "Python", "Vulnerabilities": [], } ] _write_json(api_report, nested) self.assert_failed_with(fixture.evaluate(), "GATE-TRIVY-TARGET") fixture = GateFixture( Path(self.temporary.name) / "duplicate-trivy-alias", profile="security" ) api_report = fixture.reports / "trivy-image-api.json" duplicate_alias = json.loads(api_report.read_text(encoding="utf-8")) duplicate_alias["Metadata"]["RepoTags"] *= 2 _write_json(api_report, duplicate_alias) self.assert_failed_with(fixture.evaluate(), "GATE-TRIVY-TARGET") fixture = GateFixture( Path(self.temporary.name) / "missing-scan-binding", profile="security" ) api_report = fixture.reports / "trivy-image-api.json" missing_binding = json.loads(api_report.read_text(encoding="utf-8")) del missing_binding["ScannedImageID"] _write_json(api_report, missing_binding) self.assert_failed_with(fixture.evaluate(), "GATE-TRIVY-TARGET") fixture = GateFixture( Path(self.temporary.name) / "malformed-artifact-id", profile="security" ) api_report = fixture.reports / "trivy-image-api.json" malformed_artifact = json.loads(api_report.read_text(encoding="utf-8")) malformed_artifact["ArtifactID"] = "mutable-artifact" _write_json(api_report, malformed_artifact) self.assert_failed_with(fixture.evaluate(), "GATE-TRIVY-SCHEMA") fixture = GateFixture( Path(self.temporary.name) / "missing-component-target", profile="security" ) api_report = fixture.reports / "trivy-image-api.json" nested = json.loads(api_report.read_text(encoding="utf-8")) nested["Results"] = [ { "Target": "sha256:" + "1" * 64, "Vulnerabilities": [], } ] _write_json(api_report, nested) self.assert_failed_with(fixture.evaluate(), "GATE-TRIVY-TARGET") def test_trivy_binding_preserves_component_and_rejects_swapped_artifact(self) -> None: image_id = "sha256:" + "1" * 64 expected_tag = "modelforge-security-api:" + "a" * 40 raw = self.fixture.root / "raw-trivy-image.json" bound = self.fixture.root / "bound-trivy-image.json" report = { "SchemaVersion": 2, "ArtifactName": image_id, "ArtifactID": "sha256:" + "2" * 64, "ArtifactType": "container_image", "Metadata": {"ImageID": image_id, "Reference": image_id}, "Results": [{"Target": "Python", "Vulnerabilities": []}], } _write_json(raw, report) gate.bind_trivy_image_report(raw, bound, image_id, expected_tag) normalized = json.loads(bound.read_text(encoding="utf-8")) self.assertEqual(normalized["Results"][0]["Target"], image_id) self.assertEqual(normalized["Results"][0]["ComponentTarget"], "Python") self.assertEqual(normalized["ScannedImageID"], image_id) report["Metadata"]["Reference"] = expected_tag _write_json(raw, report) gate.bind_trivy_image_report(raw, bound, image_id, expected_tag) report["Metadata"]["Reference"] = "attacker:latest" _write_json(raw, report) with self.assertRaisesRegex(gate.GateInputError, "metadata"): gate.bind_trivy_image_report(raw, bound, image_id, expected_tag) report["Metadata"]["Reference"] = image_id report["ArtifactName"] = "sha256:" + "9" * 64 _write_json(raw, report) with self.assertRaisesRegex(gate.GateInputError, "exact scanned image"): gate.bind_trivy_image_report(raw, bound, image_id, expected_tag) report["ArtifactName"] = image_id report["ArtifactID"] = "mutable-artifact" _write_json(raw, report) with self.assertRaisesRegex(gate.GateInputError, "ArtifactID"): gate.bind_trivy_image_report(raw, bound, image_id, expected_tag) def test_runtime_non_pypi_exception_is_exact_and_still_inventory_bound(self) -> None: fixture = GateFixture( Path(self.temporary.name) / "non-pypi-runtime", profile="security" ) self.assertEqual(fixture.evaluate()["decision"], "PASS") inventory_path = fixture.reports / "image-inventory-runtime-worker.json" inventory = json.loads(inventory_path.read_text(encoding="utf-8")) next(item for item in inventory["packages"] if item["name"] == "conda-build")[ "version" ] = "25.5.1" _write_json(inventory_path, inventory) self.assert_failed_with(fixture.evaluate(), "GATE-IMAGE-NON-PYPI-VERSION") def test_trivy_malformed_identity_fields_and_result_target_cannot_be_waived(self) -> None: path = self.fixture.reports / "trivy-fs.json" base: dict[str, Any] = { "SchemaVersion": 2, "ArtifactName": str(REPOSITORY_ROOT.resolve()), "ArtifactType": "repository", "Results": [ { "Target": ".", "Vulnerabilities": [ { "VulnerabilityID": "CVE-SYNTHETIC", "PkgName": "package", "InstalledVersion": "1.0", "Severity": "HIGH", "FixedVersion": "", } ], } ], } for missing in ( "VulnerabilityID", "PkgName", "InstalledVersion", "Severity", ): with self.subTest(missing=missing): malformed = copy.deepcopy(base) del malformed["Results"][0]["Vulnerabilities"][0][missing] _write_json(path, malformed) self.assert_failed_with(self.fixture.evaluate(), "GATE-TRIVY-SCHEMA") malformed_target = copy.deepcopy(base) del malformed_target["Results"][0]["Target"] _write_json(path, malformed_target) self.fixture.write_risks( [ self.fixture.risk( { "scanner": "trivy", "target": "repository", "advisory_id": "TRIVY-UNKNOWN", "package": "unknown", "installed_version": "unknown", } ) ] ) result = self.fixture.evaluate() self.assert_failed_with(result, "GATE-TRIVY-SCHEMA") self.assertEqual(result["applied_risks"], []) def test_trivy_filesystem_result_must_remain_within_repository(self) -> None: path = self.fixture.reports / "trivy-fs.json" _write_json( path, { "SchemaVersion": 2, "ArtifactName": str(REPOSITORY_ROOT.resolve()), "ArtifactType": "repository", "Results": [{"Target": "../outside", "Vulnerabilities": []}], }, ) self.assert_failed_with(self.fixture.evaluate(), "GATE-TRIVY-TARGET") def test_trivy_database_timestamps_are_required_utc_current_and_nonfuture(self) -> None: path = self.fixture.reports / "versions" / "trivy.json" original = json.loads(path.read_text(encoding="utf-8")) cases: list[tuple[str, Any]] = [ ("missing", None), ("malformed", "not-a-time"), ("non-utc", "2026-08-30T01:00:00+01:00"), ("future", "2026-08-30T00:11:00Z"), ("stale", "2026-08-27T00:00:00Z"), ] for name, updated in cases: with self.subTest(name=name): document = copy.deepcopy(original) if updated is None: document["VulnerabilityDB"].pop("UpdatedAt") else: document["VulnerabilityDB"]["UpdatedAt"] = updated _write_json(path, document) self.assert_failed_with(self.fixture.evaluate(), "GATE-TRIVY-DB-TIME") document = copy.deepcopy(original) document["VulnerabilityDB"]["DownloadedAt"] = "2026-08-29T00:00:00Z" _write_json(path, document) self.assert_failed_with(self.fixture.evaluate(), "GATE-TRIVY-DB-TIME") def test_exact_active_no_fix_risk_can_be_applied(self) -> None: identity = self.fixture.add_pip_vulnerability() self.fixture.write_risks([self.fixture.risk(identity)]) result = self.fixture.evaluate() self.assertEqual(result["decision"], "PASS") self.assertEqual(len(result["applied_risks"]), 1) def test_risk_expiration_is_exclusive_after_exactly_thirty_calendar_dates(self) -> None: identity = self.fixture.add_pip_vulnerability() risk = self.fixture.risk(identity) risk["accepted_on"] = "2026-08-01" risk["expires_on"] = "2026-08-31" self.fixture.write_risks([risk]) self.fixture.as_of = date(2026, 8, 30) self.assertEqual(self.fixture.evaluate()["decision"], "PASS") self.fixture.as_of = date(2026, 8, 31) self.assert_failed_with(self.fixture.evaluate(), "GATE-RISK-INVALID") def test_risk_with_fix_available_fails(self) -> None: identity = self.fixture.add_pip_vulnerability(fix_versions=["9.9.9"]) self.fixture.write_risks([self.fixture.risk(identity)]) self.assert_failed_with(self.fixture.evaluate(), "GATE-RISK-FIX-AVAILABLE") def test_risk_registry_version_and_classification_are_exact(self) -> None: _write_json(self.fixture.risks, []) self.assert_failed_with(self.fixture.evaluate(), "GATE-RISK-SCHEMA") identity = self.fixture.add_pip_vulnerability() risk = self.fixture.risk(identity) risk["classification"] = "ACCEPTED" self.fixture.write_risks([risk]) self.assert_failed_with(self.fixture.evaluate(), "GATE-RISK-INVALID") def test_image_inventory_requires_exact_first_party_versions_and_web_origin(self) -> None: fixture = GateFixture(Path(self.temporary.name) / "image-policy", profile="security") inventory_path = fixture.reports / "image-inventory-api.json" inventory = json.loads(inventory_path.read_text(encoding="utf-8")) next( package for package in inventory["packages"] if package["name"] == "modelforge-api" )["version"] = "0.0.0" _write_json(inventory_path, inventory) self.assert_failed_with(fixture.evaluate(), "GATE-IMAGE-FIRST-PARTY-VERSION") fixture = GateFixture(Path(self.temporary.name) / "web-policy", profile="security") evidence_path = fixture.reports / "web-bundle-origin.json" evidence = json.loads(evidence_path.read_text(encoding="utf-8")) evidence["api_origins"] = [gate.WEB_SECURITY_ORIGIN, "https://attacker.invalid"] _write_json(evidence_path, evidence) self.assert_failed_with(fixture.evaluate(), "GATE-WEB-BUNDLE-SCHEMA") def test_web_bundle_evidence_schema_binding_and_origin_controls(self) -> None: origins = gate._extract_http_origins( b'const api="https:\\/\\/security-scan.invalid/v1";' ) self.assertEqual(origins, {gate.WEB_SECURITY_ORIGIN}) poisoned = gate._extract_http_origins( b"https://security-scan.invalid/v1 https://attacker.invalid/v1" ) self.assertEqual(poisoned, {gate.WEB_SECURITY_ORIGIN, "https://attacker.invalid"}) split_poison = gate._extract_http_origins( b'const origin = "https" + "://attacker.invalid";' ) self.assertEqual(split_poison, {"https://attacker.invalid"}) for wrong_context in ( b'fetch("https://react.dev/collect")', b'fetch("http://www.w3.org/attacker")', ): with self.subTest(wrong_context=wrong_context): with self.assertRaisesRegex(gate.GateInputError, "pinned context"): gate._validate_non_api_origin_contexts(wrong_context) for origin in ( b"http://127.42.0.1:8080", b"http://[::1]:8080", b"http://LOCALHOST.:8080", b"https://api.localhost/v1", ): with self.subTest(origin=origin): with self.assertRaisesRegex(gate.GateInputError, "loopback"): gate._extract_http_origins(origin) attacks: tuple[tuple[str, Any], ...] = ( ("image_id", "sha256:" + "9" * 64), ("bundle_path", "/tmp/dead-marker"), ("bundle_sha256", "not-a-digest"), ("bundle_file_count", 0), ("provenance_path", "assets/dead-marker.json"), ("served_provenance_sha256", "0" * 64), ("provenance_binding_sha256", "0" * 64), ("runtime_chunk_path", "../outside.js"), ("runtime_chunk_sha256", "not-a-digest"), ("api_origins", [gate.WEB_SECURITY_ORIGIN, "https://attacker.invalid"]), ("non_api_origins", ["https://attacker.invalid"]), ("unexpected", "field"), ) for field, value in attacks: with self.subTest(field=field): fixture = GateFixture(self.fixture.root / f"web-{field}", profile="security") path = fixture.reports / "web-bundle-origin.json" evidence = json.loads(path.read_text(encoding="utf-8")) evidence[field] = value _write_json(path, evidence) self.assert_failed_with(fixture.evaluate(), "GATE-WEB-BUNDLE-SCHEMA") def test_web_bundle_analyzer_binds_provenance_and_rejects_dead_marker_split_poison(self) -> None: bundle = self.fixture.root / "synthetic-bundle" provenance_path = bundle / gate.WEB_PROVENANCE_PATH runtime_path = bundle / "assets" / "index.js" provenance_path.parent.mkdir(parents=True) runtime_path.parent.mkdir(parents=True) binding = gate._web_provenance_binding(gate.WEB_SECURITY_ORIGIN) provenance = { "schema_version": 1, "target": "modelforge-web", "api_base_url": gate.WEB_SECURITY_ORIGIN, "binding_sha256": binding, } _write_json(provenance_path, provenance) framework_constants = ( 'const svg="http://www.w3.org/2000/svg";' 'const docs="https://react.dev/errors/";' 'const runbook="https://git.example.com/example/modelforge/' 'src/branch/master/docs/operations/RUNBOOK_NODE_OFFLINE.md";' ) runtime_path.write_text( f'globalThis.__MODELFORGE_BUILD_PROVENANCE__={{api_base_url:' f'"{gate.WEB_SECURITY_ORIGIN}",binding_sha256:"{binding}"}};' f"{framework_constants}\n", encoding="utf-8", ) served_provenance = provenance_path.read_bytes() evidence = gate._analyze_web_bundle( bundle, image_id="sha256:" + "2" * 64, bundle_path=gate.WEB_BUNDLE_PATH, expected_origin=gate.WEB_SECURITY_ORIGIN, served_provenance=served_provenance, ) self.assertEqual(evidence["api_origins"], [gate.WEB_SECURITY_ORIGIN]) self.assertEqual(evidence["runtime_chunk_path"], "assets/index.js") self.assertEqual(evidence["provenance_path"], gate.WEB_PROVENANCE_PATH) self.assertEqual( evidence["non_api_origins"], list(gate.WEB_EXPECTED_NON_API_ORIGINS) ) with self.assertRaisesRegex(gate.GateInputError, "served Web provenance differs"): gate._analyze_web_bundle( bundle, image_id="sha256:" + "2" * 64, bundle_path=gate.WEB_BUNDLE_PATH, expected_origin=gate.WEB_SECURITY_ORIGIN, served_provenance=b"{}\n", ) runtime_path.write_text( f'const dead="{gate.WEB_SECURITY_ORIGIN}";const binding="{binding}";' f'{framework_constants}const effective="https" + "://attacker.invalid";\n', encoding="utf-8", ) with self.assertRaisesRegex(gate.GateInputError, "network origins are not exact"): gate._analyze_web_bundle( bundle, image_id="sha256:" + "2" * 64, bundle_path=gate.WEB_BUNDLE_PATH, expected_origin=gate.WEB_SECURITY_ORIGIN, served_provenance=served_provenance, ) runtime_path.write_text( f'const api="{gate.WEB_SECURITY_ORIGIN}";const binding="{binding}";' f"{framework_constants}\n", encoding="utf-8", ) (bundle / "assets" / "index.js.map").write_text( '{"sourcesContent":["const evil = `wss://${\"attacker.invalid\"}`;"]}\n', encoding="utf-8", ) with self.assertRaisesRegex(gate.GateInputError, "network origins are not exact"): gate._analyze_web_bundle( bundle, image_id="sha256:" + "2" * 64, bundle_path=gate.WEB_BUNDLE_PATH, expected_origin=gate.WEB_SECURITY_ORIGIN, served_provenance=served_provenance, ) def test_oci_identity_requires_exact_source_revision_version_and_bundle_path(self) -> None: attacks: tuple[tuple[str, str, Any], ...] = ( ("api", "source", "https://attacker.invalid/repository"), ("api", "revision", "b" * 40), ("api", "version", "0.0.0"), ( "api", "repo_tags", [ f"modelforge-security-api:{GateFixture.commit}", "attacker:latest", ], ), ( "api", "repo_tags", [ f"modelforge-security-api:{GateFixture.commit}", f"modelforge-security-api:{GateFixture.commit}", ], ), ("api", "repo_digests", ["attacker.invalid/api@sha256:" + "9" * 64]), ("web", "bundle_path", "/tmp/dead-marker"), ("runtime-worker", "unexpected", "field"), ) for attack_index, (target, field, value) in enumerate(attacks): with self.subTest(target=target, field=field): fixture = GateFixture( self.fixture.root / f"identity-{attack_index}-{target}-{field}", profile="security", ) path = fixture.reports / f"image-identity-{target}.json" identity = json.loads(path.read_text(encoding="utf-8")) identity[field] = value _write_json(path, identity) self.assert_failed_with( fixture.evaluate(), "GATE-IMAGE-IDENTITY-SCHEMA" ) fixture = GateFixture(self.fixture.root / "duplicate-image-id", profile="security") web = fixture.reports / "image-identity-web.json" identity = json.loads(web.read_text(encoding="utf-8")) identity["image_id"] = "sha256:" + "1" * 64 _write_json(web, identity) self.assert_failed_with(fixture.evaluate(), "GATE-IMAGE-SET") fixture = GateFixture(self.fixture.root / "extra-image", profile="security") _write_json( fixture.reports / "image-identity-attacker.json", {"schema_version": 1, "target": "attacker", "tag": "attacker:latest"}, ) self.assert_failed_with(fixture.evaluate(), "GATE-EVIDENCE-PATH") def test_evidence_manifest_rejects_unknown_bytes_before_hashing(self) -> None: sentinel = b"unexpected-evidence-must-not-be-consumed" (self.fixture.reports / "unexpected-secret.txt").write_bytes(sentinel) original_reader = gate.SecureEvidenceSnapshot._read_fd consumed: list[bytes] = [] def record_reads(file_descriptor: int) -> bytes: value = original_reader(file_descriptor) consumed.append(value) return value with mock.patch.object(gate.SecureEvidenceSnapshot, "_read_fd", side_effect=record_reads): result = self.fixture.evaluate() self.assert_failed_with(result, "GATE-EVIDENCE-PATH") self.assertEqual(result["report_hashes"], {}) self.assertNotIn(sentinel, consumed) def test_evidence_hardlink_is_rejected_before_hashing(self) -> None: root = self.fixture.root / "hardlink-evidence" root.mkdir(mode=0o700) root.chmod(0o700) outside = self.fixture.root / "outside-evidence.json" outside.write_bytes(b"outside-must-not-be-consumed") linked = root / "report.json" os.link(outside, linked) with mock.patch.object( gate.SecureEvidenceSnapshot, "_read_fd", wraps=gate.SecureEvidenceSnapshot._read_fd, ) as reader: with self.assertRaisesRegex(gate.GateInputError, "hard links"): gate.SecureEvidenceSnapshot(root, root / "result.json") reader.assert_not_called() def test_evaluator_output_hardlink_is_rejected_and_never_overwritten(self) -> None: outside = self.fixture.root / "outside-result.json" original = b"external-result-must-survive" outside.write_bytes(original) os.link(outside, self.fixture.output) result = self.fixture.evaluate() self.assert_failed_with(result, "GATE-EVIDENCE-PATH") self.assertEqual(outside.read_bytes(), original) # Result persistence replaces a supplied directory entry instead of following it. gate._write_json(self.fixture.output, {"replacement": "safe"}) self.assertEqual(outside.read_bytes(), original) self.assertEqual(json.loads(self.fixture.output.read_text(encoding="utf-8")), {"replacement": "safe"}) def test_duplicate_json_keys_and_normalized_inventory_rows_fail_closed(self) -> None: trivy = self.fixture.reports / "trivy-fs.json" trivy.write_text( '{"SchemaVersion":2,"ArtifactName":"' + str(REPOSITORY_ROOT.resolve()) + '","ArtifactType":"repository","Results":[{"Target":"' + str(REPOSITORY_ROOT.resolve()) + '","Vulnerabilities":[{"VulnerabilityID":"CVE-1","PkgName":"x",' '"InstalledVersion":"1","Severity":"CRITICAL"}]}],"Results":[]}', encoding="utf-8", ) self.assert_failed_with(self.fixture.evaluate(), "GATE-TRIVY-REPORT") raw_inventory = self.fixture.root / "raw-inventory.json" output = self.fixture.root / "inventory.json" requirements = self.fixture.root / "requirements.txt" for packages in ( [{"name": "HTTPX", "version": "0.28.1"}, {"name": "httpx", "version": "0.28.1"}], [{"name": "httpx", "version": "0.28.1"}, {"name": "httpx", "version": "0.28.2"}], ): _write_json(raw_inventory, {"python_version": "3.13", "packages": packages}) with self.assertRaisesRegex(gate.GateInputError, "conflicting|duplicate"): gate.normalize_image_inventory("api", raw_inventory, output, requirements) runtime_packages = [ {"name": name, "version": version} for name, version in gate.IMAGE_NON_PYPI_AUDIT_EXCEPTIONS[ "runtime-worker" ].items() ] + [ {"name": "httpx", "version": "0.28.1"}, ] _write_json( raw_inventory, {"python_version": "3.11", "packages": runtime_packages}, ) gate.normalize_image_inventory( "runtime-worker", raw_inventory, output, requirements ) self.assertNotIn("conda", requirements.read_text(encoding="utf-8")) next(item for item in runtime_packages if item["name"] == "conda")[ "version" ] = "25.5.1" _write_json( raw_inventory, {"python_version": "3.11", "packages": runtime_packages}, ) with self.assertRaisesRegex(gate.GateInputError, "non-PyPI"): gate.normalize_image_inventory( "runtime-worker", raw_inventory, output, requirements ) def test_wildcard_expired_duplicate_unused_and_ambiguous_risks_fail(self) -> None: identity = self.fixture.add_pip_vulnerability() base = self.fixture.risk(identity) cases: list[tuple[str, list[dict[str, str]], str]] = [] wildcard = copy.deepcopy(base) wildcard["target"] = "*" cases.append(("wildcard", [wildcard], "GATE-RISK-INVALID")) expired = copy.deepcopy(base) expired["accepted_on"] = "2026-07-01" expired["expires_on"] = "2026-07-31" cases.append(("expired", [expired], "GATE-RISK-INVALID")) cases.append(("duplicate", [base, copy.deepcopy(base)], "GATE-RISK-INVALID")) unused = copy.deepcopy(base) unused["advisory_id"] = "PYSEC-STALE" cases.append(("unused", [unused], "GATE-RISK-UNUSED")) for name, risks, advisory in cases: with self.subTest(name=name): self.fixture.write_risks(risks) self.assert_failed_with(self.fixture.evaluate(), advisory) self.fixture = GateFixture(Path(self.temporary.name) / "ambiguous") identity = self.fixture.add_pip_vulnerability(duplicate=True) self.fixture.write_risks([self.fixture.risk(identity)]) self.assert_failed_with(self.fixture.evaluate(), "GATE-RISK-AMBIGUOUS") def test_exact_gitleaks_false_positive_applies_and_drift_or_duplicate_fails(self) -> None: fingerprint = "d" * 64 _write_json( self.fixture.reports / "gitleaks.json", { "schema_version": 1, "scanner": "gitleaks", "target": "full-history", "redaction_percent": 100, "findings": [ {"rule_id": "generic-api-key", "fingerprint_sha256": fingerprint} ], }, ) (self.fixture.reports / "status" / "gitleaks.exit").write_text("1\n") entry = { "scanner": "gitleaks", "rule_id": "generic-api-key", "fingerprint_sha256": fingerprint, "classification": "FALSE_POSITIVE", "owner": "security-owner", "classified_on": "2026-08-30", "review_on": "2026-09-29", "tracking": "RC-DEL-02", "reason": "Verified synthetic test fixture, never credential material.", } _write_json(self.fixture.false_positives, [entry]) result = self.fixture.evaluate() self.assertEqual(result["decision"], "PASS") self.assertEqual(len(result["applied_gitleaks_false_positives"]), 1) drifted = copy.deepcopy(entry) drifted["fingerprint_sha256"] = "e" * 64 _write_json(self.fixture.false_positives, [drifted]) self.assert_failed_with(self.fixture.evaluate(), "GATE-GITLEAKS-FP-UNUSED") _write_json(self.fixture.false_positives, [entry, copy.deepcopy(entry)]) self.assert_failed_with(self.fixture.evaluate(), "GATE-GITLEAKS-FP-INVALID") if __name__ == "__main__": unittest.main()