feat(security): guard metrics server rollout

scripts/security/tests/test_runtime_image_mirror_controller.py

@@ -57,13 +57,31 @@ class RuntimeImageMirrorControllerTest(unittest.TestCase):
             ),
         )
 
+    @staticmethod
+    def _write_verified_mirror_receipt(
+        path: Path, policy, trace_id: str
+    ) -> None:
+        path.write_text(
+            json.dumps(
+                {
+                    "schema_version": controller.RECEIPT_SCHEMA_VERSION,
+                    "trace_id": trace_id,
+                    "work_item_id": policy.work_item_id,
+                    "policy_checksum": policy.checksum,
+                    "state": "verified",
+                    "verified_count": len(policy.images),
+                }
+            ),
+            encoding="utf-8",
+        )
+
     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), 15)
+        self.assertEqual(len(policy.images), 16)
         self.assertNotIn("github.com", raw)
         self.assertNotIn("ghcr.io", raw)
         self.assertIn('"external_pull_allowed": false', raw)
@@ -88,7 +106,7 @@ class RuntimeImageMirrorControllerTest(unittest.TestCase):
             )
         )
         container_kinds = [image.workload.container_kind for image in policy.images]
-        self.assertEqual(container_kinds.count("container"), 12)
+        self.assertEqual(container_kinds.count("container"), 13)
         self.assertEqual(container_kinds.count("init_container"), 3)
 
         vpa = next(
@@ -112,6 +130,29 @@ class RuntimeImageMirrorControllerTest(unittest.TestCase):
             ),
         )
 
+        metrics = next(
+            image
+            for image in policy.images
+            if image.asset_id == "runtime-image:metrics-server"
+        )
+        self.assertEqual(
+            metrics.target_registry_digest,
+            "sha256:4a4376379ab80d3e1fd5edbf8c3ddcc0b9be8ccb56167e4f8c9423e5e11b0cfe",
+        )
+        self.assertEqual(metrics.workload.namespace, "kube-system")
+        self.assertEqual(metrics.workload.name, "metrics-server")
+        self.assertEqual(metrics.workload.container, "metrics-server")
+        self.assertEqual(
+            metrics.availability_guard,
+            controller.AvailabilityGuard(
+                minimum_ready_replicas=1,
+                maximum_effective_unavailable=0,
+                minimum_effective_surge=1,
+                enforced_max_unavailable=0,
+                enforced_max_surge=1,
+            ),
+        )
+
     def test_policy_loads_and_validates_init_container_kind(self) -> None:
         image = self._image_with_container_kind("init_container")
 
@@ -129,6 +170,11 @@ class RuntimeImageMirrorControllerTest(unittest.TestCase):
             for image in payload["images"]
             if image["asset_id"] == "runtime-image:vpa-recommender"
         )
+        metrics = next(
+            image
+            for image in payload["images"]
+            if image["asset_id"] == "runtime-image:metrics-server"
+        )
         with tempfile.TemporaryDirectory() as temp:
             path = Path(temp) / "policy.json"
             for field in ("minimum_ready_replicas", "minimum_effective_surge"):
@@ -141,6 +187,26 @@ class RuntimeImageMirrorControllerTest(unittest.TestCase):
                         controller.load_policy(path)
                     vpa["availability_guard"][field] = 1
 
+            invalid_strategies = (
+                (
+                    "max_unavailable",
+                    1,
+                    "availability_enforced_unavailable_exceeds_guard",
+                ),
+                ("max_surge", 0, "invalid_enforced_max_surge"),
+            )
+            for field, value, error_code in invalid_strategies:
+                with self.subTest(field=field):
+                    strategy = metrics["availability_guard"]["enforced_strategy"]
+                    original = strategy[field]
+                    strategy[field] = value
+                    path.write_text(json.dumps(payload), encoding="utf-8")
+                    with self.assertRaisesRegex(
+                        controller.ControllerError, error_code
+                    ):
+                        controller.load_policy(path)
+                    strategy[field] = original
+
     def test_staging_policy_is_runtime_cache_only_and_bounded_canary_scoped(
         self,
     ) -> None:
@@ -273,6 +339,33 @@ class RuntimeImageMirrorControllerTest(unittest.TestCase):
             stable.runtime_ref.endswith("@" + stable.target_registry_digest)
         )
 
+    def test_metrics_server_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:metrics-server-0.8.1-canary"
+        )
+        stable = next(
+            image
+            for image in policy.images
+            if image.asset_id == "runtime-image:metrics-server"
+        )
+
+        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:4a4376379ab80d3e1fd5edbf8c3ddcc0b9be8ccb56167e4f8c9423e5e11b0cfe",
+        )
+        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]
@@ -848,6 +941,85 @@ class RuntimeImageMirrorControllerTest(unittest.TestCase):
         )
         self.assertIn("--dry-run=server", args)
 
+    def test_enforced_strategy_is_included_in_server_dry_run_patch(self) -> None:
+        image = next(
+            image
+            for image in controller.load_policy(POLICY_PATH).images
+            if image.asset_id == "runtime-image:metrics-server"
+        )
+        kubernetes = controller.Kubernetes(
+            use_sudo=False,
+            kubeconfig=None,
+            server=None,
+        )
+        strategy = kubernetes.enforced_strategy(image)
+        self.assertEqual(
+            strategy,
+            {
+                "type": "RollingUpdate",
+                "rollingUpdate": {"maxUnavailable": 0, "maxSurge": 1},
+            },
+        )
+
+        with patch.object(kubernetes, "run") as run:
+            kubernetes.patch_image(
+                image,
+                image.runtime_ref,
+                dry_run=True,
+                strategy=strategy,
+            )
+
+        args = run.call_args.args[0]
+        patch_payload = json.loads(args[args.index("--patch") + 1])
+        self.assertEqual(patch_payload["spec"]["strategy"], strategy)
+        self.assertEqual(
+            patch_payload["spec"]["template"]["spec"]["containers"],
+            [{"name": "metrics-server", "image": image.runtime_ref}],
+        )
+        self.assertIn("--dry-run=server", args)
+
+    def test_safe_strategy_override_repairs_unsafe_single_replica_plan(self) -> None:
+        image = next(
+            image
+            for image in controller.load_policy(POLICY_PATH).images
+            if image.asset_id == "runtime-image:metrics-server"
+        )
+        workload = {
+            "metadata": {"generation": 1},
+            "spec": {
+                "replicas": 1,
+                "strategy": {
+                    "type": "RollingUpdate",
+                    "rollingUpdate": {"maxUnavailable": 1, "maxSurge": "25%"},
+                },
+            },
+            "status": {
+                "observedGeneration": 1,
+                "readyReplicas": 1,
+                "availableReplicas": 1,
+                "unavailableReplicas": 0,
+            },
+        }
+        kubernetes = controller.Kubernetes(
+            use_sudo=False,
+            kubeconfig=None,
+            server=None,
+        )
+        with (
+            patch.object(kubernetes, "workload", return_value=workload),
+            self.assertRaisesRegex(
+                controller.ControllerError,
+                "availability_effective_unavailable_exceeded",
+            ),
+        ):
+            kubernetes.verify_availability(image)
+
+        with patch.object(kubernetes, "workload", return_value=workload):
+            kubernetes.verify_availability(
+                image,
+                strategy_override=kubernetes.enforced_strategy(image),
+            )
+
     def test_init_container_verifier_accepts_completed_digest_pinned_status(
         self,
     ) -> None:
@@ -1042,6 +1214,204 @@ class RuntimeImageMirrorControllerTest(unittest.TestCase):
             ):
                 kubernetes.verify_availability(image)
 
+    def test_strategy_only_diff_is_applied_and_recorded(self) -> None:
+        full_policy = controller.load_policy(POLICY_PATH)
+        image = next(
+            image
+            for image in full_policy.images
+            if image.asset_id == "runtime-image:metrics-server"
+        )
+        policy = replace(full_policy, images=(image,))
+        trace_id = "aia-p0-006-02a-strategy-only"
+        previous_strategy = {
+            "type": "RollingUpdate",
+            "rollingUpdate": {"maxUnavailable": 1, "maxSurge": "25%"},
+        }
+        kubernetes = controller.Kubernetes(
+            use_sudo=False,
+            kubeconfig=None,
+            server=None,
+        )
+        with tempfile.TemporaryDirectory() as temp:
+            receipt_path = Path(temp) / "mirror.json"
+            self._write_verified_mirror_receipt(receipt_path, policy, trace_id)
+            with (
+                patch.object(controller, "Kubernetes", return_value=kubernetes),
+                patch.object(
+                    kubernetes, "current_image", return_value=image.runtime_ref
+                ),
+                patch.object(
+                    kubernetes,
+                    "current_strategy",
+                    return_value=previous_strategy,
+                ),
+                patch.object(kubernetes, "verify_availability"),
+                patch.object(kubernetes, "patch_image") as patch_image,
+                patch.object(kubernetes, "rollout"),
+                patch.object(kubernetes, "verify_pods", return_value=1),
+                patch.object(kubernetes, "write_receipt") as write_receipt,
+            ):
+                receipt = controller.apply_policy(
+                    policy,
+                    trace_id=trace_id,
+                    mirror_receipt_path=receipt_path,
+                    apply=True,
+                    use_sudo=False,
+                    kubeconfig=None,
+                    server=None,
+                )
+
+        self.assertEqual(receipt["source_of_truth_diff_count"], 1)
+        self.assertEqual(receipt["execution_changed_count"], 1)
+        self.assertEqual(receipt["availability_strategy_diff_count"], 1)
+        self.assertEqual(receipt["availability_strategy_changed_count"], 1)
+        self.assertEqual(receipt["availability_strategy_verified_count"], 1)
+        self.assertTrue(receipt["independent_verifier_passed"])
+        self.assertEqual(patch_image.call_count, 2)
+        self.assertTrue(patch_image.call_args_list[0].kwargs["dry_run"])
+        self.assertFalse(patch_image.call_args_list[1].kwargs["dry_run"])
+        self.assertEqual(
+            patch_image.call_args_list[1].kwargs["strategy"],
+            kubernetes.enforced_strategy(image),
+        )
+        write_receipt.assert_called_once()
+
+    def test_failed_rollout_restores_previous_image_and_strategy(self) -> None:
+        full_policy = controller.load_policy(POLICY_PATH)
+        image = next(
+            image
+            for image in full_policy.images
+            if image.asset_id == "runtime-image:metrics-server"
+        )
+        policy = replace(full_policy, images=(image,))
+        trace_id = "aia-p0-006-02a-strategy-rollback"
+        previous_image = "rancher/mirrored-metrics-server:v0.8.1"
+        previous_strategy = {
+            "type": "RollingUpdate",
+            "rollingUpdate": {"maxUnavailable": 1, "maxSurge": "25%"},
+        }
+        kubernetes = controller.Kubernetes(
+            use_sudo=False,
+            kubeconfig=None,
+            server=None,
+        )
+        with tempfile.TemporaryDirectory() as temp:
+            receipt_path = Path(temp) / "mirror.json"
+            self._write_verified_mirror_receipt(receipt_path, policy, trace_id)
+            with (
+                patch.object(controller, "Kubernetes", return_value=kubernetes),
+                patch.object(
+                    kubernetes, "current_image", return_value=previous_image
+                ),
+                patch.object(
+                    kubernetes,
+                    "current_strategy",
+                    return_value=previous_strategy,
+                ),
+                patch.object(kubernetes, "verify_availability"),
+                patch.object(kubernetes, "patch_image") as patch_image,
+                patch.object(
+                    kubernetes,
+                    "rollout",
+                    side_effect=[controller.ControllerError("rollout_failed"), None],
+                ),
+                patch.object(kubernetes, "write_receipt") as write_receipt,
+            ):
+                receipt = controller.apply_policy(
+                    policy,
+                    trace_id=trace_id,
+                    mirror_receipt_path=receipt_path,
+                    apply=True,
+                    use_sudo=False,
+                    kubeconfig=None,
+                    server=None,
+                )
+
+        self.assertEqual(patch_image.call_count, 3)
+        rollback = patch_image.call_args_list[-1]
+        self.assertEqual(rollback.args[1], previous_image)
+        self.assertFalse(rollback.kwargs["dry_run"])
+        self.assertEqual(rollback.kwargs["strategy"], previous_strategy)
+        self.assertEqual(receipt["state"], "blocked_with_safe_next_action")
+        self.assertEqual(receipt["failure_reason_code"], "rollout_failed")
+        self.assertTrue(receipt["rollback_performed"])
+        self.assertTrue(receipt["rollback_verified"])
+        self.assertEqual(receipt["rollback_verified_workload_count"], 1)
+        self.assertFalse(receipt["no_write_terminal"])
+        self.assertFalse(receipt["independent_verifier_passed"])
+        write_receipt.assert_called_once_with(receipt)
+
+    def test_failed_independent_verifier_triggers_durable_rollback(self) -> None:
+        full_policy = controller.load_policy(POLICY_PATH)
+        image = next(
+            image
+            for image in full_policy.images
+            if image.asset_id == "runtime-image:metrics-server"
+        )
+        policy = replace(full_policy, images=(image,))
+        trace_id = "aia-p0-006-02a-verifier-rollback"
+        previous_image = "rancher/mirrored-metrics-server:v0.8.1"
+        previous_strategy = {
+            "type": "RollingUpdate",
+            "rollingUpdate": {"maxUnavailable": 1, "maxSurge": "25%"},
+        }
+        kubernetes = controller.Kubernetes(
+            use_sudo=False,
+            kubeconfig=None,
+            server=None,
+        )
+        with tempfile.TemporaryDirectory() as temp:
+            receipt_path = Path(temp) / "mirror.json"
+            self._write_verified_mirror_receipt(receipt_path, policy, trace_id)
+            with (
+                patch.object(controller, "Kubernetes", return_value=kubernetes),
+                patch.object(
+                    kubernetes,
+                    "current_image",
+                    side_effect=[
+                        previous_image,
+                        image.runtime_ref,
+                        previous_image,
+                        previous_image,
+                    ],
+                ),
+                patch.object(
+                    kubernetes,
+                    "current_strategy",
+                    return_value=previous_strategy,
+                ),
+                patch.object(kubernetes, "verify_availability"),
+                patch.object(kubernetes, "patch_image") as patch_image,
+                patch.object(kubernetes, "rollout"),
+                patch.object(
+                    kubernetes,
+                    "verify_pods",
+                    side_effect=controller.ControllerError(
+                        "pod_image_digest_mismatch"
+                    ),
+                ),
+                patch.object(kubernetes, "write_receipt") as write_receipt,
+            ):
+                receipt = controller.apply_policy(
+                    policy,
+                    trace_id=trace_id,
+                    mirror_receipt_path=receipt_path,
+                    apply=True,
+                    use_sudo=False,
+                    kubeconfig=None,
+                    server=None,
+                )
+
+        self.assertEqual(patch_image.call_count, 3)
+        self.assertEqual(receipt["state"], "blocked_with_safe_next_action")
+        self.assertEqual(
+            receipt["failure_reason_code"], "pod_image_digest_mismatch"
+        )
+        self.assertTrue(receipt["rollback_performed"])
+        self.assertTrue(receipt["rollback_verified"])
+        self.assertEqual(receipt["rollback_verified_workload_count"], 1)
+        write_receipt.assert_called_once_with(receipt)
+
 
 if __name__ == "__main__":
     unittest.main()