Harden runtime distribution acceptance

tests/test_managed_ingress_drill.py

@@ -0,0 +1,65 @@
+from __future__ import annotations
+
+import importlib.util
+from pathlib import Path
+import subprocess
+import sys
+import unittest
+from unittest.mock import patch
+
+
+ROOT = Path(__file__).resolve().parents[1]
+
+
+def _load_module():
+    path = ROOT / "tools/checks/managed-ingress-drill.py"
+    spec = importlib.util.spec_from_file_location("managed_ingress_drill", path)
+    assert spec is not None and spec.loader is not None
+    module = importlib.util.module_from_spec(spec)
+    sys.modules[spec.name] = module
+    spec.loader.exec_module(module)
+    return module
+
+
+INGRESS = _load_module()
+
+
+class ManagedIngressDrillTests(unittest.TestCase):
+    def test_config_is_streamed_into_a_daemon_visible_volume(self) -> None:
+        completed = subprocess.CompletedProcess([], 0, "", "")
+        with patch.object(INGRESS, "_run", return_value=completed) as run:
+            INGRESS._write_volume_file(
+                image="registry.example/caddy@sha256:" + "1" * 64,
+                volume="config-volume",
+                filename="Caddyfile",
+                content=":8080 { respond /health 200 }\n",
+            )
+
+        argv = run.call_args.args[0]
+        self.assertIn("type=volume,src=config-volume,dst=/govoplan-config", argv)
+        self.assertNotIn("type=bind", " ".join(argv))
+        self.assertEqual(
+            ":8080 { respond /health 200 }\n",
+            run.call_args.kwargs["input_text"],
+        )
+
+    def test_config_filename_cannot_escape_the_volume(self) -> None:
+        with self.assertRaisesRegex(ValueError, "invalid config filename"):
+            INGRESS._write_volume_file(
+                image="registry.example/caddy@sha256:" + "1" * 64,
+                volume="config-volume",
+                filename="../Caddyfile",
+                content="",
+            )
+
+    def test_drill_has_no_runner_local_bind_mounts(self) -> None:
+        source = (ROOT / "tools/checks/managed-ingress-drill.py").read_text(
+            encoding="utf-8"
+        )
+
+        self.assertNotIn("type=bind", source)
+        self.assertIn('"--network-alias",\n                    "load-balancer"', source)
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_runtime_distribution_build.py

@@ -3,7 +3,9 @@ from __future__ import annotations
 import argparse
 import importlib.util
 import json
+import os
 from pathlib import Path
+import shutil
 import sys
 import tempfile
 import unittest
@@ -26,9 +28,32 @@ FINALIZE = _load(
     "finalize_runtime_distribution",
     ROOT / "tools/release/finalize-runtime-distribution.py",
 )
+DEPLOYER_BUILD = _load(
+    "build_deployer_zipapp",
+    ROOT / "tools/deployment/build-deployer-zipapp.py",
+)
 
 
 class RuntimeDistributionBuildTests(unittest.TestCase):
+    def test_deployment_zipapp_is_reproducible_across_source_mtimes(self) -> None:
+        with tempfile.TemporaryDirectory(prefix="govoplan-reproducible-zipapp-") as value:
+            root = Path(value)
+            source = root / "source"
+            shutil.copytree(ROOT / "tools/deployment", source)
+            first = root / "first.pyz"
+            second = root / "second.pyz"
+            original_root = DEPLOYER_BUILD.ROOT
+            try:
+                DEPLOYER_BUILD.ROOT = source
+                self.assertEqual(0, DEPLOYER_BUILD.main(["--output", str(first)]))
+                for path in source.rglob("*.py"):
+                    os.utime(path, (2_000_000_000, 2_000_000_000))
+                self.assertEqual(0, DEPLOYER_BUILD.main(["--output", str(second)]))
+            finally:
+                DEPLOYER_BUILD.ROOT = original_root
+
+            self.assertEqual(first.read_bytes(), second.read_bytes())
+
     def test_workflow_signs_with_the_release_environment(self) -> None:
         workflow = (ROOT / ".gitea/workflows/runtime-distribution.yml").read_text(
             encoding="utf-8"
@@ -43,6 +68,35 @@ class RuntimeDistributionBuildTests(unittest.TestCase):
             workflow,
         )
 
+    def test_workflow_verifies_portable_bootstrap_artifacts_before_execution(
+        self,
+    ) -> None:
+        workflow = (ROOT / ".gitea/workflows/runtime-distribution.yml").read_text(
+            encoding="utf-8"
+        )
+
+        self.assertIn(
+            "(cd runtime-output && sha256sum govoplan-deploy.pyz > "
+            "govoplan-deploy.pyz.sha256)",
+            workflow,
+        )
+        self.assertIn("openssl pkeyutl -verify -pubin", workflow)
+        self.assertIn("govoplan-deploy.tampered.pyz", workflow)
+        self.assertLess(
+            workflow.index("openssl pkeyutl -verify -pubin"),
+            workflow.index("python runtime-output/govoplan-deploy.pyz init"),
+        )
+
+    def test_workflow_retains_both_platform_runtime_smoke_receipts(self) -> None:
+        workflow = (ROOT / ".gitea/workflows/runtime-distribution.yml").read_text(
+            encoding="utf-8"
+        )
+
+        self.assertIn('for ARCH in amd64 arm64; do', workflow)
+        self.assertIn("tools/checks/runtime-image-smoke.py", workflow)
+        self.assertIn("runtime-smoke-amd64.json", workflow)
+        self.assertIn("runtime-smoke-arm64.json", workflow)
+
     def test_resolves_platforms_and_builds_evidence_descriptor(self) -> None:
         index = {
             "schemaVersion": 2,

tests/test_runtime_image_smoke.py

@@ -0,0 +1,57 @@
+from __future__ import annotations
+
+import importlib.util
+import json
+from pathlib import Path
+import sys
+import tempfile
+import unittest
+
+
+ROOT = Path(__file__).resolve().parents[1]
+SCRIPT = ROOT / "tools/checks/runtime-image-smoke.py"
+SPEC = importlib.util.spec_from_file_location("runtime_image_smoke", SCRIPT)
+assert SPEC is not None and SPEC.loader is not None
+MODULE = importlib.util.module_from_spec(SPEC)
+sys.modules[SPEC.name] = MODULE
+SPEC.loader.exec_module(MODULE)
+
+
+class RuntimeImageSmokeTests(unittest.TestCase):
+    def test_selects_the_exact_platform_digest(self) -> None:
+        with tempfile.TemporaryDirectory(prefix="govoplan-runtime-smoke-") as value:
+            path = Path(value) / "metadata.json"
+            path.write_text(
+                json.dumps(
+                    {
+                        "index": "registry.example/api@sha256:" + "a" * 64,
+                        "platforms": {
+                            "linux/amd64": "registry.example/api@sha256:" + "1" * 64,
+                            "linux/arm64": "registry.example/api@sha256:" + "2" * 64,
+                        },
+                    }
+                ),
+                encoding="utf-8",
+            )
+
+            self.assertEqual(
+                "registry.example/api@sha256:" + "2" * 64,
+                MODULE.platform_image(path, "linux/arm64", "API"),
+            )
+
+    def test_rejects_mutable_or_missing_platform_images(self) -> None:
+        with tempfile.TemporaryDirectory(prefix="govoplan-runtime-smoke-") as value:
+            path = Path(value) / "metadata.json"
+            path.write_text(
+                json.dumps({"platforms": {"linux/amd64": "registry.example/api:latest"}}),
+                encoding="utf-8",
+            )
+
+            with self.assertRaisesRegex(MODULE.SmokeError, "exact sha256"):
+                MODULE.platform_image(path, "linux/amd64", "API")
+            with self.assertRaisesRegex(MODULE.SmokeError, "exact sha256"):
+                MODULE.platform_image(path, "linux/arm64", "API")
+
+
+if __name__ == "__main__":
+    unittest.main()