feat(security): guard VPA image rollout

scripts/security/tests/test_runtime_image_mirror_controller.py

@@ -46,13 +46,24 @@ class RuntimeImageMirrorControllerTest(unittest.TestCase):
         policy = controller.load_staging_policy(STAGING_POLICY_PATH)
         return replace(policy, images=(policy.images[0],))
 
+    def _image_with_availability_guard(self):
+        image = controller.load_policy(POLICY_PATH).images[0]
+        return replace(
+            image,
+            availability_guard=controller.AvailabilityGuard(
+                minimum_ready_replicas=1,
+                maximum_effective_unavailable=0,
+                minimum_effective_surge=1,
+            ),
+        )
+
     def test_policy_is_internal_digest_pinned_and_runtime_cache_only(self) -> None:
         policy = controller.load_policy(POLICY_PATH)
         raw = POLICY_PATH.read_text(encoding="utf-8")
 
         self.assertEqual(policy.work_item_id, "AIA-P0-006-02A")
         self.assertEqual(policy.risk_level, "high")
-        self.assertEqual(len(policy.images), 14)
+        self.assertEqual(len(policy.images), 15)
         self.assertNotIn("github.com", raw)
         self.assertNotIn("ghcr.io", raw)
         self.assertIn('"external_pull_allowed": false', raw)
@@ -77,9 +88,30 @@ class RuntimeImageMirrorControllerTest(unittest.TestCase):
             )
         )
         container_kinds = [image.workload.container_kind for image in policy.images]
-        self.assertEqual(container_kinds.count("container"), 11)
+        self.assertEqual(container_kinds.count("container"), 12)
         self.assertEqual(container_kinds.count("init_container"), 3)
 
+        vpa = next(
+            image
+            for image in policy.images
+            if image.asset_id == "runtime-image:vpa-recommender"
+        )
+        self.assertEqual(
+            vpa.target_registry_digest,
+            "sha256:6ae9b8c6ac256fe034f4e2741c2fef2c2ba2d64778c071d6d93c99cd86ba7c89",
+        )
+        self.assertEqual(vpa.workload.namespace, "kube-system")
+        self.assertEqual(vpa.workload.name, "vpa-recommender")
+        self.assertEqual(vpa.workload.container, "recommender")
+        self.assertEqual(
+            vpa.availability_guard,
+            controller.AvailabilityGuard(
+                minimum_ready_replicas=1,
+                maximum_effective_unavailable=0,
+                minimum_effective_surge=1,
+            ),
+        )
+
     def test_policy_loads_and_validates_init_container_kind(self) -> None:
         image = self._image_with_container_kind("init_container")
 
@@ -90,6 +122,25 @@ class RuntimeImageMirrorControllerTest(unittest.TestCase):
         ):
             self._image_with_container_kind("ephemeral_container")
 
+    def test_availability_guard_cannot_be_configured_as_noop(self) -> None:
+        payload = json.loads(POLICY_PATH.read_text(encoding="utf-8"))
+        vpa = next(
+            image
+            for image in payload["images"]
+            if image["asset_id"] == "runtime-image:vpa-recommender"
+        )
+        with tempfile.TemporaryDirectory() as temp:
+            path = Path(temp) / "policy.json"
+            for field in ("minimum_ready_replicas", "minimum_effective_surge"):
+                with self.subTest(field=field):
+                    vpa["availability_guard"][field] = 0
+                    path.write_text(json.dumps(payload), encoding="utf-8")
+                    with self.assertRaisesRegex(
+                        controller.ControllerError, f"invalid_{field}"
+                    ):
+                        controller.load_policy(path)
+                    vpa["availability_guard"][field] = 1
+
     def test_staging_policy_is_runtime_cache_only_and_bounded_canary_scoped(
         self,
     ) -> None:
@@ -176,6 +227,31 @@ class RuntimeImageMirrorControllerTest(unittest.TestCase):
             canary.runtime_ref.endswith("@" + canary.target_registry_digest)
         )
 
+    def test_vpa_stable_policy_reuses_verified_staging_provenance(self) -> None:
+        policy = controller.load_policy(POLICY_PATH)
+        staging = next(
+            image
+            for image in controller.load_staging_policy(STAGING_POLICY_PATH).images
+            if image.asset_id == "runtime-image:vpa-recommender-1.6.0-canary"
+        )
+        stable = next(
+            image
+            for image in policy.images
+            if image.asset_id == "runtime-image:vpa-recommender"
+        )
+
+        self.assertEqual(stable.source_index_digest, staging.source_index_digest)
+        self.assertEqual(stable.platform_digest, staging.platform_digest)
+        self.assertEqual(stable.push_ref, staging.push_ref)
+        self.assertEqual(stable.workload, staging.workload)
+        self.assertEqual(
+            stable.target_registry_digest,
+            "sha256:6ae9b8c6ac256fe034f4e2741c2fef2c2ba2d64778c071d6d93c99cd86ba7c89",
+        )
+        self.assertTrue(
+            stable.runtime_ref.endswith("@" + stable.target_registry_digest)
+        )
+
     def test_argocd_controller_batch_reuses_verified_staging_artifact(self) -> None:
         policy = controller.load_policy(POLICY_PATH)
         staging = controller.load_staging_policy(STAGING_POLICY_PATH).images[0]
@@ -459,6 +535,52 @@ class RuntimeImageMirrorControllerTest(unittest.TestCase):
         self.assertIn("--insecure", command)
         self.assertEqual(command[-1], controller._target_digest_ref(image))
 
+    def test_runtime_cache_verifies_shared_source_digest_once(self) -> None:
+        policy = controller.load_policy(POLICY_PATH)
+        shared = [
+            image
+            for image in policy.images
+            if image.source_index_digest
+            == "sha256:16b92ba472fbb9287459cc52e0ecff07288dff461209955098edb56ce866fe49"
+        ]
+        self.assertGreaterEqual(len(shared), 2)
+        source_ref = "quay.io/argoproj/argocd:v3.3.6"
+        source_list = (
+            f"{source_ref} application/vnd.oci.image.index.v1+json "
+            f"{shared[0].source_index_digest}\n"
+        )
+        source_index = json.dumps(
+            {
+                "manifests": [
+                    {
+                        "digest": shared[0].platform_digest,
+                        "platform": {"os": "linux", "architecture": "amd64"},
+                    }
+                ]
+            }
+        )
+        cache = controller.RuntimeCache(
+            policy,
+            Path("/unused/key"),
+            Path("/unused/known-hosts"),
+        )
+        completed = controller.subprocess.CompletedProcess
+        with patch.object(
+            controller,
+            "_run",
+            side_effect=[
+                completed([], 0, stdout=source_list),
+                completed([], 0, stdout="complete"),
+                completed([], 0, stdout=source_index),
+            ],
+        ) as run:
+            first = cache.verify_source(shared[0])
+            second = cache.verify_source(shared[1])
+
+        self.assertEqual(first, source_ref)
+        self.assertEqual(second, source_ref)
+        self.assertEqual(run.call_count, 3)
+
     def test_source_manifests_match_internal_policy(self) -> None:
         policy = controller.load_policy(POLICY_PATH)
         by_asset = {image.asset_id: image for image in policy.images}
@@ -784,6 +906,121 @@ class RuntimeImageMirrorControllerTest(unittest.TestCase):
                     run.return_value.stdout = json.dumps(pods)
                     kubernetes.verify_pods(image)
 
+    def test_single_replica_rolling_update_passes_availability_guard(self) -> None:
+        image = self._image_with_availability_guard()
+        workload = {
+            "metadata": {"generation": 4},
+            "spec": {
+                "replicas": 1,
+                "strategy": {
+                    "type": "RollingUpdate",
+                    "rollingUpdate": {
+                        "maxUnavailable": "25%",
+                        "maxSurge": "25%",
+                    },
+                },
+            },
+            "status": {
+                "observedGeneration": 4,
+                "readyReplicas": 1,
+                "availableReplicas": 1,
+                "unavailableReplicas": 0,
+            },
+        }
+        kubernetes = controller.Kubernetes(
+            use_sudo=False,
+            kubeconfig=None,
+            server=None,
+        )
+
+        with patch.object(kubernetes, "workload", return_value=workload):
+            kubernetes.verify_availability(image)
+
+        self.assertEqual(
+            kubernetes._effective_replicas(
+                "25%", 1, round_up=False, label="unavailable"
+            ),
+            0,
+        )
+        self.assertEqual(
+            kubernetes._effective_replicas(
+                "25%", 1, round_up=True, label="surge"
+            ),
+            1,
+        )
+
+    def test_availability_guard_fails_closed_on_unsafe_rollout_state(self) -> None:
+        image = self._image_with_availability_guard()
+        safe = {
+            "metadata": {"generation": 4},
+            "spec": {
+                "replicas": 1,
+                "strategy": {
+                    "type": "RollingUpdate",
+                    "rollingUpdate": {"maxUnavailable": 0, "maxSurge": 1},
+                },
+            },
+            "status": {
+                "observedGeneration": 4,
+                "readyReplicas": 1,
+                "availableReplicas": 1,
+                "unavailableReplicas": 0,
+            },
+        }
+        cases = (
+            (
+                {**safe, "metadata": {"generation": 5}},
+                "availability_generation_not_observed",
+            ),
+            (
+                {
+                    **safe,
+                    "status": {
+                        **safe["status"],
+                        "readyReplicas": 0,
+                        "availableReplicas": 0,
+                    },
+                },
+                "availability_ready_replicas_insufficient",
+            ),
+            (
+                {
+                    **safe,
+                    "spec": {
+                        **safe["spec"],
+                        "strategy": {
+                            "type": "RollingUpdate",
+                            "rollingUpdate": {
+                                "maxUnavailable": 1,
+                                "maxSurge": 0,
+                            },
+                        },
+                    },
+                },
+                "availability_effective_unavailable_exceeded",
+            ),
+            (
+                {
+                    **safe,
+                    "spec": {**safe["spec"], "strategy": {"type": "Recreate"}},
+                },
+                "availability_rolling_update_required",
+            ),
+        )
+        kubernetes = controller.Kubernetes(
+            use_sudo=False,
+            kubeconfig=None,
+            server=None,
+        )
+
+        for workload, error_code in cases:
+            with (
+                self.subTest(error_code=error_code),
+                patch.object(kubernetes, "workload", return_value=workload),
+                self.assertRaisesRegex(controller.ControllerError, error_code),
+            ):
+                kubernetes.verify_availability(image)
+
 
 if __name__ == "__main__":
     unittest.main()