diff --git a/config/security/runtime-image-mirror-policy.json b/config/security/runtime-image-mirror-policy.json index 26219f176..9741597d6 100644 --- a/config/security/runtime-image-mirror-policy.json +++ b/config/security/runtime-image-mirror-policy.json @@ -232,6 +232,29 @@ "maximum_effective_unavailable": 0, "minimum_effective_surge": 1 } + }, + { + "asset_id": "runtime-image:metrics-server", + "source_index_digest": "sha256:b2d2efaf5ac3b366ed0f839d2412a2c4279d4fc2a2a733f12c52133faed36c41", + "platform_digest": "sha256:6231fb0a1ffab76c92ab880f51a0d11b290f688373647bcedff85af025dfd8a9", + "target_registry_digest": "sha256:4a4376379ab80d3e1fd5edbf8c3ddcc0b9be8ccb56167e4f8c9423e5e11b0cfe", + "push_ref": "registry.wooo.work/awoooi/runtime-mirror/metrics-server:0.8.1-linux-amd64", + "runtime_ref": "192.168.0.110:5000/awoooi/runtime-mirror/metrics-server@sha256:4a4376379ab80d3e1fd5edbf8c3ddcc0b9be8ccb56167e4f8c9423e5e11b0cfe", + "workload": { + "kind": "deployment", + "namespace": "kube-system", + "name": "metrics-server", + "container": "metrics-server" + }, + "availability_guard": { + "minimum_ready_replicas": 1, + "maximum_effective_unavailable": 0, + "minimum_effective_surge": 1, + "enforced_strategy": { + "max_unavailable": 0, + "max_surge": 1 + } + } } ] } diff --git a/scripts/security/runtime_image_mirror_controller.py b/scripts/security/runtime_image_mirror_controller.py index 011c5ea77..0050f514a 100644 --- a/scripts/security/runtime_image_mirror_controller.py +++ b/scripts/security/runtime_image_mirror_controller.py @@ -47,6 +47,8 @@ class AvailabilityGuard: minimum_ready_replicas: int maximum_effective_unavailable: int minimum_effective_surge: int + enforced_max_unavailable: int | None = None + enforced_max_surge: int | None = None @dataclass(frozen=True) @@ -230,19 +232,46 @@ def load_policy(path: Path) -> Policy: raise ControllerError("availability_guard_invalid") if workload.kind != "deployment": raise ControllerError("availability_guard_requires_deployment") + minimum_ready_replicas = _require_positive_int( + guard_row.get("minimum_ready_replicas"), + "minimum_ready_replicas", + ) + maximum_effective_unavailable = _require_nonnegative_int( + guard_row.get("maximum_effective_unavailable"), + "maximum_effective_unavailable", + ) + minimum_effective_surge = _require_positive_int( + guard_row.get("minimum_effective_surge"), + "minimum_effective_surge", + ) + enforced_max_unavailable = None + enforced_max_surge = None + strategy_row = guard_row.get("enforced_strategy") + if strategy_row is not None: + if not isinstance(strategy_row, dict): + raise ControllerError("availability_enforced_strategy_invalid") + enforced_max_unavailable = _require_nonnegative_int( + strategy_row.get("max_unavailable"), + "enforced_max_unavailable", + ) + enforced_max_surge = _require_positive_int( + strategy_row.get("max_surge"), + "enforced_max_surge", + ) + if enforced_max_unavailable > maximum_effective_unavailable: + raise ControllerError( + "availability_enforced_unavailable_exceeds_guard" + ) + if enforced_max_surge < minimum_effective_surge: + raise ControllerError( + "availability_enforced_surge_below_guard" + ) availability_guard = AvailabilityGuard( - minimum_ready_replicas=_require_positive_int( - guard_row.get("minimum_ready_replicas"), - "minimum_ready_replicas", - ), - maximum_effective_unavailable=_require_nonnegative_int( - guard_row.get("maximum_effective_unavailable"), - "maximum_effective_unavailable", - ), - minimum_effective_surge=_require_positive_int( - guard_row.get("minimum_effective_surge"), - "minimum_effective_surge", - ), + minimum_ready_replicas=minimum_ready_replicas, + maximum_effective_unavailable=maximum_effective_unavailable, + minimum_effective_surge=minimum_effective_surge, + enforced_max_unavailable=enforced_max_unavailable, + enforced_max_surge=enforced_max_surge, ) images.append( ImagePolicy( @@ -890,6 +919,29 @@ class Kubernetes: raise ControllerError("workload_container_not_found") return values[0] + def current_strategy(self, item: ImagePolicy) -> dict[str, Any]: + strategy = self.workload(item).get("spec", {}).get("strategy") + if not isinstance(strategy, dict): + raise ControllerError("workload_strategy_missing") + return strategy + + @staticmethod + def enforced_strategy(item: ImagePolicy) -> dict[str, Any] | None: + guard = item.availability_guard + if ( + guard is None + or guard.enforced_max_unavailable is None + or guard.enforced_max_surge is None + ): + return None + return { + "type": "RollingUpdate", + "rollingUpdate": { + "maxUnavailable": guard.enforced_max_unavailable, + "maxSurge": guard.enforced_max_surge, + }, + } + @staticmethod def _effective_replicas( value: Any, replicas: int, *, round_up: bool, label: str @@ -909,7 +961,12 @@ class Kubernetes: return math.ceil(scaled) if round_up else math.floor(scaled) raise ControllerError(f"{label}_invalid") - def verify_availability(self, item: ImagePolicy) -> None: + def verify_availability( + self, + item: ImagePolicy, + *, + strategy_override: dict[str, Any] | None = None, + ) -> None: guard = item.availability_guard if guard is None: return @@ -953,7 +1010,9 @@ class Kubernetes: ): raise ControllerError("availability_unavailable_replicas_exceeded") - strategy = spec.get("strategy", {}) + strategy = strategy_override if strategy_override is not None else spec.get( + "strategy", {} + ) if not isinstance(strategy, dict) or strategy.get("type") != "RollingUpdate": raise ControllerError("availability_rolling_update_required") rolling = strategy.get("rollingUpdate", {}) @@ -976,20 +1035,28 @@ class Kubernetes: if effective_surge < guard.minimum_effective_surge: raise ControllerError("availability_effective_surge_insufficient") - def patch_image(self, item: ImagePolicy, image_ref: str, *, dry_run: bool) -> None: + def patch_image( + self, + item: ImagePolicy, + image_ref: str, + *, + dry_run: bool, + strategy: dict[str, Any] | None = None, + ) -> None: container_field = self._spec_container_field(item) - patch = json.dumps( - { + spec_patch: dict[str, Any] = { + "template": { "spec": { - "template": { - "spec": { - container_field: [ - {"name": item.workload.container, "image": image_ref} - ] - } - } + container_field: [ + {"name": item.workload.container, "image": image_ref} + ] } - }, + } + } + if strategy is not None: + spec_patch["strategy"] = strategy + patch = json.dumps( + {"spec": spec_patch}, separators=(",", ":"), ) args = [ @@ -1127,39 +1194,102 @@ def apply_policy( mirror_receipt = _read_mirror_receipt(mirror_receipt_path, policy, trace_id) kubernetes = Kubernetes(use_sudo=use_sudo, kubeconfig=kubeconfig, server=server) previous: dict[str, str] = {} + previous_strategies: dict[str, dict[str, Any] | None] = {} + desired_strategies: dict[str, dict[str, Any] | None] = {} + strategy_diffs: dict[str, bool] = {} + requires_changes: dict[str, bool] = {} changed: list[ImagePolicy] = [] + failure_reason_code: str | None = None + rollback_performed = False + rollback_verified = False + rollback_verified_workload_count = 0 for item in policy.images: previous[item.asset_id] = kubernetes.current_image(item) - kubernetes.verify_availability(item) - kubernetes.patch_image(item, item.runtime_ref, dry_run=True) + desired_strategy = kubernetes.enforced_strategy(item) + desired_strategies[item.asset_id] = desired_strategy + previous_strategy = ( + kubernetes.current_strategy(item) if desired_strategy is not None else None + ) + previous_strategies[item.asset_id] = previous_strategy + strategy_diffs[item.asset_id] = ( + desired_strategy is not None and previous_strategy != desired_strategy + ) + requires_changes[item.asset_id] = ( + previous[item.asset_id] != item.runtime_ref + or strategy_diffs[item.asset_id] + ) + kubernetes.verify_availability( + item, + strategy_override=desired_strategy, + ) + kubernetes.patch_image( + item, + item.runtime_ref, + dry_run=True, + strategy=desired_strategy, + ) if apply: try: for item in policy.images: - if previous[item.asset_id] == item.runtime_ref: + if not requires_changes[item.asset_id]: continue - kubernetes.patch_image(item, item.runtime_ref, dry_run=False) + kubernetes.patch_image( + item, + item.runtime_ref, + dry_run=False, + strategy=desired_strategies[item.asset_id], + ) changed.append(item) for item in policy.images: kubernetes.rollout(item) kubernetes.verify_availability(item) - except ControllerError: + for item in policy.images: + if kubernetes.current_image(item) != item.runtime_ref: + raise ControllerError("workload_image_not_applied") + kubernetes.verify_pods(item) + except ControllerError as exc: + failure_reason_code = str(exc) + rollback_performed = bool(changed) for item in reversed(changed): try: - kubernetes.patch_image(item, previous[item.asset_id], dry_run=False) + kubernetes.patch_image( + item, + previous[item.asset_id], + dry_run=False, + strategy=previous_strategies[item.asset_id], + ) kubernetes.rollout(item) + image_restored = ( + kubernetes.current_image(item) == previous[item.asset_id] + ) + strategy_restored = ( + previous_strategies[item.asset_id] is None + or kubernetes.current_strategy(item) + == previous_strategies[item.asset_id] + ) except ControllerError: - pass - raise + continue + if image_restored and strategy_restored: + rollback_verified_workload_count += 1 + rollback_verified = ( + not changed or rollback_verified_workload_count == len(changed) + ) pod_count = 0 compliant_count = 0 for item in policy.images: - if kubernetes.current_image(item) == item.runtime_ref: - compliant_count += 1 - pod_count += kubernetes.verify_pods(item) + try: + if kubernetes.current_image(item) == item.runtime_ref: + compliant_count += 1 + pod_count += kubernetes.verify_pods(item) + except ControllerError as exc: + if failure_reason_code is None: + failure_reason_code = str(exc) - verified = compliant_count == len(policy.images) + verified = ( + failure_reason_code is None and compliant_count == len(policy.images) + ) receipt = { "schema_version": RECEIPT_SCHEMA_VERSION, "trace_id": trace_id, @@ -1171,7 +1301,7 @@ def apply_policy( "sensor_receipt_present": True, "normalized_asset_count": len(policy.images), "source_of_truth_diff_count": sum( - 1 for item in policy.images if previous[item.asset_id] != item.runtime_ref + 1 for item in policy.images if requires_changes[item.asset_id] ), "ai_candidate_action": "replace_forbidden_registry_with_internal_digest", "policy_decision": "controlled_apply_allowed", @@ -1182,17 +1312,30 @@ def apply_policy( "availability_guard_verified_count": sum( 1 for item in policy.images if item.availability_guard is not None ), + "availability_strategy_diff_count": sum(strategy_diffs.values()), + "availability_strategy_changed_count": sum( + 1 for item in changed if strategy_diffs[item.asset_id] + ), + "availability_strategy_verified_count": ( + sum(1 for value in desired_strategies.values() if value is not None) + if apply and verified + else 0 + ), "execution_changed_count": len(changed), "independent_verifier_passed": verified, "verified_workload_count": compliant_count, "verified_pod_count": pod_count, - "rollback_performed": False, + "rollback_performed": rollback_performed, + "rollback_verified": rollback_verified, + "rollback_verified_workload_count": rollback_verified_workload_count, + "no_write_terminal": failure_reason_code is not None and not changed, + "failure_reason_code": failure_reason_code, "external_pull_allowed": False, "mirror_stage_verified": mirror_receipt.get("state") == "verified", "durable_writeback": "kubernetes_configmap", "state": "verified" if verified else "blocked_with_safe_next_action", } - if apply and verified: + if apply: kubernetes.write_receipt(receipt) return receipt diff --git a/scripts/security/tests/test_runtime_image_mirror_controller.py b/scripts/security/tests/test_runtime_image_mirror_controller.py index 966a63f41..8a245a6f4 100644 --- a/scripts/security/tests/test_runtime_image_mirror_controller.py +++ b/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()