diff --git a/config/security/runtime-image-mirror-policy.json b/config/security/runtime-image-mirror-policy.json index b3a3cb5b7..26219f176 100644 --- a/config/security/runtime-image-mirror-policy.json +++ b/config/security/runtime-image-mirror-policy.json @@ -213,6 +213,25 @@ "name": "node-problem-detector", "container": "node-problem-detector" } + }, + { + "asset_id": "runtime-image:vpa-recommender", + "source_index_digest": "sha256:4af365370c187e4eb60302e937b18a188774698b049a1bfdd77d43e51b30abec", + "platform_digest": "sha256:f7da718d281d5e880e966d0182ffb2160b82517c7bc3a38f8002b49e7a4bd0a1", + "target_registry_digest": "sha256:6ae9b8c6ac256fe034f4e2741c2fef2c2ba2d64778c071d6d93c99cd86ba7c89", + "push_ref": "registry.wooo.work/awoooi/runtime-mirror/vpa-recommender:1.6.0-linux-amd64", + "runtime_ref": "192.168.0.110:5000/awoooi/runtime-mirror/vpa-recommender@sha256:6ae9b8c6ac256fe034f4e2741c2fef2c2ba2d64778c071d6d93c99cd86ba7c89", + "workload": { + "kind": "deployment", + "namespace": "kube-system", + "name": "vpa-recommender", + "container": "recommender" + }, + "availability_guard": { + "minimum_ready_replicas": 1, + "maximum_effective_unavailable": 0, + "minimum_effective_surge": 1 + } } ] } diff --git a/scripts/security/runtime_image_mirror_controller.py b/scripts/security/runtime_image_mirror_controller.py index 6438bf040..011c5ea77 100644 --- a/scripts/security/runtime_image_mirror_controller.py +++ b/scripts/security/runtime_image_mirror_controller.py @@ -6,6 +6,7 @@ from __future__ import annotations import argparse import hashlib import json +import math import re import shlex import subprocess @@ -41,6 +42,13 @@ class Workload: container_kind: str +@dataclass(frozen=True) +class AvailabilityGuard: + minimum_ready_replicas: int + maximum_effective_unavailable: int + minimum_effective_surge: int + + @dataclass(frozen=True) class ImagePolicy: asset_id: str @@ -50,6 +58,7 @@ class ImagePolicy: push_ref: str runtime_ref: str workload: Workload + availability_guard: AvailabilityGuard | None @dataclass(frozen=True) @@ -112,6 +121,19 @@ def _require_name(value: Any, label: str) -> str: return name +def _require_nonnegative_int(value: Any, label: str) -> int: + if isinstance(value, bool) or not isinstance(value, int) or value < 0: + raise ControllerError(f"invalid_{label}") + return value + + +def _require_positive_int(value: Any, label: str) -> int: + number = _require_nonnegative_int(value, label) + if number == 0: + raise ControllerError(f"invalid_{label}") + return number + + def _policy_checksum(payload: dict[str, Any]) -> str: canonical = json.dumps(payload, sort_keys=True, separators=(",", ":")).encode() return "sha256:" + hashlib.sha256(canonical).hexdigest() @@ -201,6 +223,27 @@ def load_policy(path: Path) -> Policy: if workload_key in workloads: raise ControllerError("duplicate_workload_target") workloads.add(workload_key) + availability_guard = None + guard_row = row.get("availability_guard") + if guard_row is not None: + if not isinstance(guard_row, dict): + raise ControllerError("availability_guard_invalid") + if workload.kind != "deployment": + raise ControllerError("availability_guard_requires_deployment") + 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", + ), + ) images.append( ImagePolicy( asset_id=asset_id, @@ -210,6 +253,7 @@ def load_policy(path: Path) -> Policy: push_ref=push_ref, runtime_ref=runtime_ref, workload=workload, + availability_guard=availability_guard, ) ) @@ -372,6 +416,7 @@ class RuntimeCache: self, policy: Policy | StagingPolicy, ssh_key: Path, known_hosts: Path ) -> None: self.policy = policy + self._verified_sources: dict[tuple[str, str], str] = {} self.ssh = [ "ssh", "-i", @@ -403,6 +448,10 @@ class RuntimeCache: return sorted(set(matches))[0] def verify_source(self, image: ImagePolicy | StagingImage) -> str: + cache_key = (image.source_index_digest, image.platform_digest) + cached = self._verified_sources.get(cache_key) + if cached is not None: + return cached source_ref = self.source_ref(image.source_index_digest) check = _run(self._command("images", "check", f"name=={source_ref}")) if "complete" not in (check.stdout or ""): @@ -423,6 +472,7 @@ class RuntimeCache: ] if selected != [image.platform_digest]: raise ControllerError("runtime_cache_platform_digest_mismatch") + self._verified_sources[cache_key] = source_ref return source_ref def source_config_digest(self, platform_digest: str) -> str: @@ -840,6 +890,92 @@ class Kubernetes: raise ControllerError("workload_container_not_found") return values[0] + @staticmethod + def _effective_replicas( + value: Any, replicas: int, *, round_up: bool, label: str + ) -> int: + if isinstance(value, bool): + raise ControllerError(f"{label}_invalid") + if isinstance(value, int) and value >= 0: + return value + if isinstance(value, str) and value.endswith("%"): + try: + percent = int(value[:-1]) + except ValueError as exc: + raise ControllerError(f"{label}_invalid") from exc + if percent < 0 or percent > 100: + raise ControllerError(f"{label}_invalid") + scaled = replicas * percent / 100 + return math.ceil(scaled) if round_up else math.floor(scaled) + raise ControllerError(f"{label}_invalid") + + def verify_availability(self, item: ImagePolicy) -> None: + guard = item.availability_guard + if guard is None: + return + workload = self.workload(item) + metadata = workload.get("metadata", {}) + spec = workload.get("spec", {}) + status = workload.get("status", {}) + generation = metadata.get("generation") + observed_generation = status.get("observedGeneration") + replicas = spec.get("replicas", 1) + if ( + isinstance(replicas, bool) + or not isinstance(replicas, int) + or replicas < guard.minimum_ready_replicas + ): + raise ControllerError("availability_replicas_insufficient") + if ( + isinstance(generation, bool) + or not isinstance(generation, int) + or isinstance(observed_generation, bool) + or not isinstance(observed_generation, int) + or observed_generation < generation + ): + raise ControllerError("availability_generation_not_observed") + ready = status.get("readyReplicas", 0) + available = status.get("availableReplicas", 0) + unavailable = status.get("unavailableReplicas", 0) + if ( + isinstance(ready, bool) + or not isinstance(ready, int) + or isinstance(available, bool) + or not isinstance(available, int) + or ready < guard.minimum_ready_replicas + or available < guard.minimum_ready_replicas + ): + raise ControllerError("availability_ready_replicas_insufficient") + if ( + isinstance(unavailable, bool) + or not isinstance(unavailable, int) + or unavailable > guard.maximum_effective_unavailable + ): + raise ControllerError("availability_unavailable_replicas_exceeded") + + strategy = spec.get("strategy", {}) + if not isinstance(strategy, dict) or strategy.get("type") != "RollingUpdate": + raise ControllerError("availability_rolling_update_required") + rolling = strategy.get("rollingUpdate", {}) + if not isinstance(rolling, dict): + raise ControllerError("availability_rolling_update_invalid") + effective_unavailable = self._effective_replicas( + rolling.get("maxUnavailable", "25%"), + replicas, + round_up=False, + label="availability_max_unavailable", + ) + effective_surge = self._effective_replicas( + rolling.get("maxSurge", "25%"), + replicas, + round_up=True, + label="availability_max_surge", + ) + if effective_unavailable > guard.maximum_effective_unavailable: + raise ControllerError("availability_effective_unavailable_exceeded") + 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: container_field = self._spec_container_field(item) patch = json.dumps( @@ -994,6 +1130,7 @@ def apply_policy( changed: list[ImagePolicy] = [] 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) if apply: @@ -1005,6 +1142,7 @@ def apply_policy( changed.append(item) for item in policy.images: kubernetes.rollout(item) + kubernetes.verify_availability(item) except ControllerError: for item in reversed(changed): try: @@ -1038,6 +1176,12 @@ def apply_policy( "ai_candidate_action": "replace_forbidden_registry_with_internal_digest", "policy_decision": "controlled_apply_allowed", "server_dry_run_passed": True, + "availability_guard_count": sum( + 1 for item in policy.images if item.availability_guard is not None + ), + "availability_guard_verified_count": sum( + 1 for item in policy.images if item.availability_guard is not None + ), "execution_changed_count": len(changed), "independent_verifier_passed": verified, "verified_workload_count": compliant_count, diff --git a/scripts/security/tests/test_runtime_image_mirror_controller.py b/scripts/security/tests/test_runtime_image_mirror_controller.py index fac8a7b40..50bed6043 100644 --- a/scripts/security/tests/test_runtime_image_mirror_controller.py +++ b/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()