281 lines
13 KiB
Python
281 lines
13 KiB
Python
"""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()
|