"""Structural safety checks for public-candidate server acceptance.""" from __future__ import annotations import subprocess import importlib.util import json import tempfile import unittest from pathlib import Path ROOT = Path(__file__).resolve().parents[2] WORKFLOW = ROOT / ".gitea" / "workflows" / "public-candidate-acceptance.yml" MANAGED_WORKFLOW = ROOT / ".gitea" / "workflows" / "managed-validation.yml" ACCEPTANCE = ROOT / "scripts" / "rc_server_acceptance.py" TRIVY_VALIDATOR = ROOT / "scripts" / "validate_trivy_report.py" RUNTIME_DOCKERFILE = ROOT / "runtime-worker" / "Dockerfile" RUNTIME_PYPROJECT = ROOT / "runtime-worker" / "pyproject.toml" class RcAcceptancePolicyTests(unittest.TestCase): @classmethod def setUpClass(cls) -> None: cls.workflow = WORKFLOW.read_text(encoding="utf-8") cls.managed = MANAGED_WORKFLOW.read_text(encoding="utf-8") cls.acceptance = ACCEPTANCE.read_text(encoding="utf-8") cls.trivy = TRIVY_VALIDATOR.read_text(encoding="utf-8") cls.runtime_dockerfile = RUNTIME_DOCKERFILE.read_text(encoding="utf-8") cls.runtime_pyproject = RUNTIME_PYPROJECT.read_text(encoding="utf-8") def test_workflow_is_manual_and_never_calls_production_deployment(self) -> None: self.assertIn("workflow_dispatch:", self.workflow) self.assertNotIn("\n push:", self.workflow) self.assertNotIn("pull_request:", self.workflow) self.assertNotIn("deploy.py", self.workflow) self.assertNotIn("DEPLOY_STABLE_TO_PRODUCTION", self.workflow) def test_exact_source_and_docker_are_required(self) -> None: self.assertIn(r"^[0-9a-f]{40}$", self.workflow) self.assertIn("ref: ${{ inputs.source_commit }}", self.workflow) self.assertIn("docker version", self.workflow) self.assertNotIn("SKIPPED: no docker", self.workflow) def test_security_tools_are_downloaded_with_exact_checksums(self) -> None: self.assertIn("GITLEAKS_VERSION: 8.30.1", self.workflow) self.assertIn( "GITLEAKS_SHA256: 551f6fc83ea457d62a0d98237cbad105af8d557003051f41f3e7ca7b3f2470eb", self.workflow, ) self.assertIn('printf \'GITLEAKS=%s\\n\'', self.workflow) self.assertIn('\"${GITLEAKS}\" dir \"${public_source}\"', self.workflow) self.assertIn("sha256sum --check --strict", self.workflow) self.assertNotIn("command -v gitleaks", self.workflow) def test_unmerged_branch_can_use_the_existing_managed_dispatch_entry(self) -> None: self.assertIn("workflow_call:", self.workflow) self.assertIn("uses: ./.gitea/workflows/public-candidate-acceptance.yml", self.managed) self.assertIn("inputs.profile == 'build'", self.managed) self.assertIn("source_commit: ${{ gitea.sha }}", self.managed) def test_all_four_images_are_built_and_scanned(self) -> None: for image in ( "modelforge-api", "modelforge-web", "modelforge-node-agent", "modelforge-runtime-worker", ): self.assertIn(image, self.acceptance) self.assertIn('"Metadata"', self.trivy) self.assertIn('metadata.get("ImageID") != expected_image_id', self.trivy) self.assertIn("reject_duplicate_keys", self.trivy) self.assertIn("fixable_high_critical + unreviewed_unfixed_high_critical", self.trivy) self.assertIn("public-candidate-unfixed-vulnerabilities.json", self.acceptance) def test_runtime_image_applies_security_updates_and_exact_inference_pins(self) -> None: self.assertIn("apt-get upgrade --yes", self.runtime_dockerfile) self.assertIn("apt-get purge --yes linux-libc-dev", self.runtime_dockerfile) self.assertNotIn("apt-get autoremove", self.runtime_dockerfile) self.assertIn("pip check", self.runtime_dockerfile) for dependency in ( "Pillow==12.3.0", "protobuf==5.29.6", "sentencepiece==0.2.2", "sentence-transformers==6.0.1", "transformers==5.16.1", "urllib3==2.7.0", ): self.assertIn(dependency, self.runtime_pyproject) def test_compose_project_is_isolated_and_always_removed(self) -> None: self.assertIn('project = f"modelforge-rc-', self.acceptance) self.assertIn('"MODELFORGE_API_BIND": "127.0.0.1"', self.acceptance) self.assertIn('"MODELFORGE_API_PUBLISHED_PORT": "0"', self.acceptance) self.assertIn('"production_changed": False', self.acceptance) self.assertIn('"compute_identity_created": False', self.acceptance) self.assertIn('"down", "--volumes", "--remove-orphans"', self.acceptance) self.assertIn("atexit.register(cleanup_candidate_images", self.acceptance) self.assertIn('"docker", "image", "rm", "--force", tag', self.acceptance) self.assertNotIn("docker-compose.runtime-worker.yml", self.acceptance) self.assertNotIn("docker-compose.node-agent.yml", self.acceptance) def test_clean_install_bootstraps_roles_across_a_remote_docker_daemon(self) -> None: self.assertIn("def provision_database_roles(", self.acceptance) self.assertIn('run("docker", "cp", str(bootstrap)', self.acceptance) self.assertIn('"psql",', self.acceptance) self.assertIn('"--file",', self.acceptance) self.assertIn( '"--wait", "postgres", "redis", env=env', self.acceptance ) self.assertLess( self.acceptance.index("provision_database_roles(compose, env, database)"), self.acceptance.index( '"--wait", "api", "web", env=env' ), ) def test_clean_install_seeds_config_across_a_remote_docker_daemon(self) -> None: self.assertIn("def provision_config_volume(project: str)", self.acceptance) self.assertIn('f"{project}_acceptance-config"', self.acceptance) self.assertIn('f"com.docker.compose.project={project}"', self.acceptance) self.assertIn('run("docker", "cp", f"{ROOT / \'config\'}/."', self.acceptance) self.assertIn( 'run("docker", "container", "rm", "--force", seed, check=False)', self.acceptance, ) self.assertIn('" read_only: true\\n"', self.acceptance) self.assertLess( self.acceptance.index("provision_config_volume(project)"), self.acceptance.index('"--wait", "api", "web", env=env'), ) def test_http_contract_runs_inside_the_isolated_compose_network(self) -> None: self.assertIn("def container_http_status(", self.acceptance) self.assertIn("def wait_for_container_status(", self.acceptance) self.assertIn('headers["X-ModelForge-Admin-Token"]', self.acceptance) self.assertIn('os.environ["MODELFORGE_OPERATOR_API_KEY"]', self.acceptance) self.assertIn('wait_for_container_status(api_container, "http://web:3000/", 200)', self.acceptance) self.assertNotIn("urllib.request.urlopen(request, timeout=timeout)", self.acceptance) def test_workflow_has_an_always_run_exact_project_cleanup_fallback(self) -> None: self.assertIn("- name: Always remove acceptance Docker resources", self.workflow) cleanup = self.workflow.split( " - name: Always remove acceptance Docker resources", 1 )[1].split(" - name: Upload acceptance evidence", 1)[0] self.assertIn("if: always()", cleanup) self.assertIn('label=com.docker.compose.project=${project}', cleanup) self.assertIn("docker container rm --force", cleanup) self.assertIn("docker volume rm --force", cleanup) self.assertIn("docker network rm", cleanup) self.assertIn("docker image rm --force", cleanup) self.assertNotIn("docker system prune", cleanup) def test_python_entrypoints_compile(self) -> None: for path in (ACCEPTANCE, TRIVY_VALIDATOR): result = subprocess.run( ["python", "-m", "py_compile", str(path)], capture_output=True, check=False, ) with self.subTest(path=path.name): self.assertEqual(result.returncode, 0, result.stderr.decode("utf-8", "replace")) def test_trivy_validator_binds_image_and_rejects_blockers(self) -> None: spec = importlib.util.spec_from_file_location("trivy_report", TRIVY_VALIDATOR) assert spec and spec.loader module = importlib.util.module_from_spec(spec) spec.loader.exec_module(module) image_id = "sha256:" + "a" * 64 report = { "ArtifactName": image_id, "ArtifactType": "container_image", "Metadata": {"ImageID": image_id}, "Results": [ { "Target": "debian", "Class": "os-pkgs", "Type": "debian", "Vulnerabilities": [ { "VulnerabilityID": "CVE-2099-9999", "PkgName": "blocked-package", "InstalledVersion": "1.0.0", "FixedVersion": "1.0.1", "Severity": "HIGH", } ], } ], } with tempfile.TemporaryDirectory() as directory: path = Path(directory) / "report.json" path.write_text(json.dumps(report), encoding="utf-8") result = module.validate(path, image_id) self.assertEqual(result["release_blockers"], 1) with self.assertRaisesRegex(ValueError, "does not match"): module.validate(path, "sha256:" + "b" * 64) def test_trivy_validator_rejects_duplicate_keys_and_empty_coverage(self) -> None: spec = importlib.util.spec_from_file_location("trivy_report_negative", TRIVY_VALIDATOR) assert spec and spec.loader module = importlib.util.module_from_spec(spec) spec.loader.exec_module(module) image_id = "sha256:" + "a" * 64 with tempfile.TemporaryDirectory() as directory: path = Path(directory) / "report.json" path.write_text('{"ArtifactType":"container_image","ArtifactType":"container_image"}', encoding="utf-8") with self.assertRaisesRegex(ValueError, "duplicate JSON key"): module.validate(path, image_id) path.write_text( json.dumps( { "ArtifactType": "container_image", "Metadata": {"ImageID": image_id}, "Results": [], } ), encoding="utf-8", ) with self.assertRaisesRegex(ValueError, "no package result coverage"): module.validate(path, image_id) def test_trivy_validator_allows_only_exact_reviewed_upstream_unfixed_findings(self) -> None: spec = importlib.util.spec_from_file_location("trivy_report_reviewed", TRIVY_VALIDATOR) assert spec and spec.loader module = importlib.util.module_from_spec(spec) spec.loader.exec_module(module) image_id = "sha256:" + "a" * 64 finding = { "VulnerabilityID": "CVE-2099-0001", "PkgName": "example-package", "InstalledVersion": "1.0.0", "FixedVersion": "", "Severity": "HIGH", } report = { "ArtifactType": "container_image", "Metadata": {"ImageID": image_id}, "Results": [ { "Target": "debian", "Class": "os-pkgs", "Vulnerabilities": [finding], } ], } baseline = { "schema_version": 1, "vulnerabilities": [ { "vulnerability_id": "CVE-2099-0001", "package": "example-package", "installed_version": "1.0.0", "severity": "HIGH", } ], } with tempfile.TemporaryDirectory() as directory: report_path = Path(directory) / "report.json" baseline_path = Path(directory) / "baseline.json" report_path.write_text(json.dumps(report), encoding="utf-8") baseline_path.write_text(json.dumps(baseline), encoding="utf-8") summary = module.validate(report_path, image_id, baseline_path) self.assertEqual(summary["reviewed_unfixed_high_critical"], 1) self.assertEqual(summary["release_blockers"], 0) finding["InstalledVersion"] = "1.0.1" report_path.write_text(json.dumps(report), encoding="utf-8") self.assertEqual(module.validate(report_path, image_id, baseline_path)["release_blockers"], 1) finding["InstalledVersion"] = "1.0.0" finding["FixedVersion"] = "1.0.2" report_path.write_text(json.dumps(report), encoding="utf-8") summary = module.validate(report_path, image_id, baseline_path) self.assertEqual(summary["fixable_high_critical"], 1) self.assertEqual(summary["release_blockers"], 1) if __name__ == "__main__": unittest.main()