from __future__ import annotations import importlib.util import json import sys import tarfile import tempfile import unittest from dataclasses import replace from io import BytesIO from pathlib import Path from unittest.mock import patch ROOT = Path(__file__).resolve().parents[3] CONTROLLER_PATH = ROOT / "scripts/security/runtime_image_mirror_controller.py" POLICY_PATH = ROOT / "config/security/runtime-image-mirror-policy.json" STAGING_POLICY_PATH = ROOT / "config/security/runtime-image-mirror-staging-policy.json" def _load_controller(): spec = importlib.util.spec_from_file_location( "runtime_image_mirror_controller", CONTROLLER_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 controller = _load_controller() class RuntimeImageMirrorControllerTest(unittest.TestCase): def _image_with_container_kind(self, container_kind: str): payload = json.loads(POLICY_PATH.read_text(encoding="utf-8")) payload["images"] = [payload["images"][0]] payload["images"][0]["workload"]["container_kind"] = container_kind with tempfile.TemporaryDirectory() as temp: path = Path(temp) / "policy.json" path.write_text(json.dumps(payload), encoding="utf-8") return controller.load_policy(path).images[0] def _single_staging_image_policy(self): 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, ), ) @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), 16) self.assertNotIn("github.com", raw) self.assertNotIn("ghcr.io", raw) self.assertIn('"external_pull_allowed": false', raw) self.assertTrue( all( image.runtime_ref.startswith( "192.168.0.110:5000/awoooi/runtime-mirror/" ) for image in policy.images ) ) self.assertTrue( all( image.runtime_ref.endswith("@" + image.target_registry_digest) for image in policy.images ) ) self.assertTrue( all( image.platform_digest != image.target_registry_digest for image in policy.images ) ) container_kinds = [image.workload.container_kind for image in policy.images] self.assertEqual(container_kinds.count("container"), 13) 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, ), ) 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") self.assertEqual(image.workload.container_kind, "init_container") with self.assertRaisesRegex( controller.ControllerError, "container_kind_not_allowed" ): 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" ) 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"): 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 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: policy = controller.load_staging_policy(STAGING_POLICY_PATH) raw = STAGING_POLICY_PATH.read_text(encoding="utf-8") self.assertEqual(policy.work_item_id, "AIA-P0-006-02B") self.assertEqual(policy.risk_level, "high") self.assertEqual(len(policy.images), 5) by_asset = {image.asset_id: image for image in policy.images} self.assertEqual( set(by_asset), { "runtime-image:argocd-v3.3.6-canary", "runtime-image:local-path-provisioner-0.0.34-canary", "runtime-image:metrics-server-0.8.1-canary", "runtime-image:redis-8.2.3-canary", "runtime-image:vpa-recommender-1.6.0-canary", }, ) argocd = by_asset["runtime-image:argocd-v3.3.6-canary"] self.assertEqual(argocd.runtime_repository, "argocd") self.assertEqual(argocd.workload.name, "argocd-dex-server") self.assertEqual(argocd.workload.container, "copyutil") self.assertEqual(argocd.workload.container_kind, "init_container") redis = by_asset["runtime-image:redis-8.2.3-canary"] self.assertEqual( redis.source_index_digest, "sha256:08ad0b1d280850169a790dba1393ff7a90aef951fc19632cf4d3ce4f78e679ba", ) self.assertEqual( redis.platform_digest, "sha256:e499175dfb27569cd40010c2eee346113db95fdd0efc88ab9fd70a9e807f4542", ) self.assertEqual( redis.source_config_digest, "sha256:f508abec2c47b3b5099daf9e7b11ead2a9396c17ca46654f2f9723c14892e3da", ) self.assertEqual(redis.runtime_repository, "redis") self.assertEqual(redis.workload.name, "argocd-redis") self.assertEqual(redis.workload.container, "redis") self.assertEqual(redis.workload.container_kind, "container") vpa = by_asset["runtime-image:vpa-recommender-1.6.0-canary"] self.assertEqual( vpa.source_index_digest, "sha256:4af365370c187e4eb60302e937b18a188774698b049a1bfdd77d43e51b30abec", ) self.assertEqual( vpa.platform_digest, "sha256:f7da718d281d5e880e966d0182ffb2160b82517c7bc3a38f8002b49e7a4bd0a1", ) self.assertEqual( vpa.source_config_digest, "sha256:05526683ba80f70887ef5622c4ee7d78e1e0482d2a9e2f0e29ad92c32cb73819", ) self.assertEqual(vpa.runtime_repository, "vpa-recommender") self.assertEqual(vpa.workload.kind, "deployment") self.assertEqual(vpa.workload.namespace, "kube-system") self.assertEqual(vpa.workload.name, "vpa-recommender") self.assertEqual(vpa.workload.container, "recommender") self.assertEqual(vpa.workload.container_kind, "container") metrics = by_asset["runtime-image:metrics-server-0.8.1-canary"] self.assertEqual( metrics.source_index_digest, "sha256:b2d2efaf5ac3b366ed0f839d2412a2c4279d4fc2a2a733f12c52133faed36c41", ) self.assertEqual( metrics.platform_digest, "sha256:6231fb0a1ffab76c92ab880f51a0d11b290f688373647bcedff85af025dfd8a9", ) self.assertEqual( metrics.source_config_digest, "sha256:e76b3f3568b7f440dfd477c1d6de638d7769ba34c93eef999dee418eb72bc0e3", ) self.assertEqual(metrics.runtime_repository, "metrics-server") self.assertEqual(metrics.workload.kind, "deployment") self.assertEqual(metrics.workload.namespace, "kube-system") self.assertEqual(metrics.workload.name, "metrics-server") self.assertEqual(metrics.workload.container, "metrics-server") self.assertEqual(metrics.workload.container_kind, "container") local_path = by_asset[ "runtime-image:local-path-provisioner-0.0.34-canary" ] self.assertEqual( local_path.source_index_digest, "sha256:6ff68ebe98bc623b45ad22c28be84f8a08214982710f3247d5862e9bccce73ef", ) self.assertEqual( local_path.platform_digest, "sha256:90745b5457d61b84362ea2aeeb06caef4d576c99861eaff4c7bfe9b36e1cf8d7", ) self.assertEqual( local_path.source_config_digest, "sha256:acccaf97bcb578daff51c6246e47babbca6e998e4225aa230544684cf147d6c1", ) self.assertEqual(local_path.runtime_repository, "local-path-provisioner") self.assertEqual(local_path.workload.kind, "deployment") self.assertEqual(local_path.workload.namespace, "kube-system") self.assertEqual(local_path.workload.name, "local-path-provisioner") self.assertEqual(local_path.workload.container, "local-path-provisioner") self.assertEqual(local_path.workload.container_kind, "container") self.assertNotIn("github.com", raw) self.assertNotIn("ghcr.io", raw) self.assertIn('"external_pull_allowed": false', raw) def test_argocd_copyutil_canary_matches_verified_staging_contract(self) -> None: policy = controller.load_policy(POLICY_PATH) staging = controller.load_staging_policy(STAGING_POLICY_PATH).images[0] canary = next( image for image in policy.images if image.asset_id == "runtime-image:argocd-copyutil" ) self.assertEqual(canary.source_index_digest, staging.source_index_digest) self.assertEqual(canary.platform_digest, staging.platform_digest) self.assertEqual(canary.push_ref, staging.push_ref) self.assertEqual(canary.workload, staging.workload) self.assertEqual( canary.target_registry_digest, "sha256:8b7e25d7e6036d6750c5bda8920cbd80d8902c25771666f6767dc97bd7fab8fc", ) self.assertTrue( 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_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] batch = { image.asset_id: image for image in policy.images if image.asset_id in { "runtime-image:argocd-applicationset-controller", "runtime-image:argocd-notifications-controller", } } self.assertEqual( set(batch), { "runtime-image:argocd-applicationset-controller", "runtime-image:argocd-notifications-controller", }, ) self.assertEqual( { ( image.workload.kind, image.workload.namespace, image.workload.name, image.workload.container, image.workload.container_kind, ) for image in batch.values() }, { ( "deployment", "argocd", "argocd-applicationset-controller", "argocd-applicationset-controller", "container", ), ( "deployment", "argocd", "argocd-notifications-controller", "argocd-notifications-controller", "container", ), }, ) for image in batch.values(): self.assertEqual(image.source_index_digest, staging.source_index_digest) self.assertEqual(image.platform_digest, staging.platform_digest) self.assertEqual(image.push_ref, staging.push_ref) self.assertEqual( image.target_registry_digest, "sha256:8b7e25d7e6036d6750c5bda8920cbd80d8902c25771666f6767dc97bd7fab8fc", ) self.assertTrue( image.runtime_ref.endswith("@" + image.target_registry_digest) ) def test_argocd_repo_server_batch_reuses_verified_staging_artifact(self) -> None: policy = controller.load_policy(POLICY_PATH) staging = controller.load_staging_policy(STAGING_POLICY_PATH).images[0] batch = { image.asset_id: image for image in policy.images if image.asset_id in { "runtime-image:argocd-repo-server", "runtime-image:argocd-repo-server-copyutil", } } self.assertEqual( set(batch), { "runtime-image:argocd-repo-server", "runtime-image:argocd-repo-server-copyutil", }, ) self.assertEqual( { ( image.workload.kind, image.workload.namespace, image.workload.name, image.workload.container, image.workload.container_kind, ) for image in batch.values() }, { ( "deployment", "argocd", "argocd-repo-server", "argocd-repo-server", "container", ), ( "deployment", "argocd", "argocd-repo-server", "copyutil", "init_container", ), }, ) for image in batch.values(): self.assertEqual(image.source_index_digest, staging.source_index_digest) self.assertEqual(image.platform_digest, staging.platform_digest) self.assertEqual(image.push_ref, staging.push_ref) self.assertEqual( image.target_registry_digest, "sha256:8b7e25d7e6036d6750c5bda8920cbd80d8902c25771666f6767dc97bd7fab8fc", ) self.assertTrue( image.runtime_ref.endswith("@" + image.target_registry_digest) ) def test_argocd_control_plane_batch_reuses_verified_staging_artifact(self) -> None: policy = controller.load_policy(POLICY_PATH) staging = controller.load_staging_policy(STAGING_POLICY_PATH).images[0] batch = { image.asset_id: image for image in policy.images if image.asset_id in { "runtime-image:argocd-server", "runtime-image:argocd-application-controller", } } self.assertEqual( set(batch), { "runtime-image:argocd-server", "runtime-image:argocd-application-controller", }, ) self.assertEqual( { ( image.workload.kind, image.workload.namespace, image.workload.name, image.workload.container, image.workload.container_kind, ) for image in batch.values() }, { ( "deployment", "argocd", "argocd-server", "argocd-server", "container", ), ( "statefulset", "argocd", "argocd-application-controller", "argocd-application-controller", "container", ), }, ) for image in batch.values(): self.assertEqual(image.source_index_digest, staging.source_index_digest) self.assertEqual(image.platform_digest, staging.platform_digest) self.assertEqual(image.push_ref, staging.push_ref) self.assertEqual( image.target_registry_digest, "sha256:8b7e25d7e6036d6750c5bda8920cbd80d8902c25771666f6767dc97bd7fab8fc", ) self.assertTrue( image.runtime_ref.endswith("@" + image.target_registry_digest) ) def test_argocd_redis_init_reuses_verified_staging_artifact(self) -> None: policy = controller.load_policy(POLICY_PATH) staging = controller.load_staging_policy(STAGING_POLICY_PATH).images[0] image = next( image for image in policy.images if image.asset_id == "runtime-image:argocd-redis-secret-init" ) self.assertEqual(image.source_index_digest, staging.source_index_digest) self.assertEqual(image.platform_digest, staging.platform_digest) self.assertEqual(image.push_ref, staging.push_ref) self.assertEqual(image.workload.kind, "deployment") self.assertEqual(image.workload.namespace, "argocd") self.assertEqual(image.workload.name, "argocd-redis") self.assertEqual(image.workload.container, "secret-init") self.assertEqual(image.workload.container_kind, "init_container") self.assertEqual( image.target_registry_digest, "sha256:8b7e25d7e6036d6750c5bda8920cbd80d8902c25771666f6767dc97bd7fab8fc", ) self.assertTrue(image.runtime_ref.endswith("@" + image.target_registry_digest)) def test_argocd_redis_main_reuses_verified_staging_artifact(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:redis-8.2.3-canary" ) image = next( image for image in policy.images if image.asset_id == "runtime-image:argocd-redis" ) self.assertEqual(image.source_index_digest, staging.source_index_digest) self.assertEqual(image.platform_digest, staging.platform_digest) self.assertEqual(image.push_ref, staging.push_ref) self.assertEqual(image.workload.kind, "deployment") self.assertEqual(image.workload.namespace, "argocd") self.assertEqual(image.workload.name, "argocd-redis") self.assertEqual(image.workload.container, "redis") self.assertEqual(image.workload.container_kind, "container") self.assertEqual( image.target_registry_digest, "sha256:1e1de310669186e8a85a676ed5352f6078b09658ee578508f7608fe7beff78f0", ) self.assertTrue(image.runtime_ref.endswith("@" + image.target_registry_digest)) def test_policy_rejects_external_pull_and_mutable_runtime_ref(self) -> None: payload = json.loads(POLICY_PATH.read_text(encoding="utf-8")) with tempfile.TemporaryDirectory() as temp: path = Path(temp) / "policy.json" payload["source_contract"]["external_pull_allowed"] = True path.write_text(json.dumps(payload), encoding="utf-8") with self.assertRaisesRegex( controller.ControllerError, "external_pull_must_be_disabled" ): controller.load_policy(path) payload["source_contract"]["external_pull_allowed"] = False payload["images"][0][ "runtime_ref" ] = "192.168.0.110:5000/awoooi/runtime-mirror/dex:latest" path.write_text(json.dumps(payload), encoding="utf-8") with self.assertRaisesRegex( controller.ControllerError, "runtime_ref_digest_mismatch" ): controller.load_policy(path) def test_archive_verifier_requires_source_platform_manifest(self) -> None: policy = controller.load_policy(POLICY_PATH) image = policy.images[0] with tempfile.TemporaryDirectory() as temp: archive_path = Path(temp) / "image.tar" member = "blobs/sha256/" + image.platform_digest.removeprefix("sha256:") with tarfile.open(archive_path, "w") as archive: info = tarfile.TarInfo(member) info.size = 2 archive.addfile(info, BytesIO(b"{}")) self.assertTrue( controller._archive_contains_digest(archive_path, image.platform_digest) ) self.assertFalse( controller._archive_contains_digest( archive_path, image.target_registry_digest ) ) def test_target_digest_verifier_uses_internal_registry_transport(self) -> None: image = controller.load_policy(POLICY_PATH).images[0] with patch.object(controller.subprocess, "run") as run: run.return_value.returncode = 0 self.assertTrue(controller._target_digest_present(image)) command = run.call_args.args[0] self.assertEqual(command[:3], ["docker", "manifest", "inspect"]) 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} kured = (ROOT / "k8s/kured/kured.yaml").read_text(encoding="utf-8") event = (ROOT / "k8s/observability/event-exporter.yaml").read_text( encoding="utf-8" ) kube_state_metrics = ( ROOT / "k8s/kube-state-metrics/kube-state-metrics.yaml" ).read_text(encoding="utf-8") node_problem_detector = (ROOT / "k8s/npd/node-problem-detector.yaml").read_text( encoding="utf-8" ) self.assertIn(by_asset["runtime-image:kured"].runtime_ref, kured) self.assertIn(by_asset["runtime-image:event-exporter"].runtime_ref, event) self.assertIn( by_asset["runtime-image:kube-state-metrics"].runtime_ref, kube_state_metrics, ) self.assertIn( by_asset["runtime-image:node-problem-detector"].runtime_ref, node_problem_detector, ) self.assertNotIn("ghcr.io", kured) self.assertNotIn("ghcr.io", event) self.assertNotIn("registry.k8s.io", kube_state_metrics) self.assertNotIn("registry.k8s.io", node_problem_detector) def test_controller_and_cd_never_pull_external_images(self) -> None: controller_source = CONTROLLER_PATH.read_text(encoding="utf-8") workflow = (ROOT / ".gitea/workflows/cd.yaml").read_text(encoding="utf-8") self.assertNotIn("docker pull", controller_source) self.assertNotIn('"images", "pull"', controller_source) self.assertIn("Mirror Runtime Cache Images to Harbor", workflow) self.assertIn("Apply Runtime Image Mirror Policy", workflow) self.assertIn("runtime-image-mirror-staging-policy.json", workflow) self.assertIn("runtime_image_staging_stage=verified", workflow) self.assertIn("runtime_image_mirror_controller.py", workflow) mirror_block = workflow[ workflow.index( "- name: Mirror Runtime Cache Images to Harbor" ) : workflow.index("# ── API 鏡像建置") ] self.assertNotIn("HARBOR_PASSWORD", mirror_block) def test_staging_reuses_verified_internal_artifact_without_export(self) -> None: policy = self._single_staging_image_policy() image = policy.images[0] descriptor = controller.RegistryDescriptor( digest="sha256:" + "1" * 64, config_digest=image.source_config_digest, ) with tempfile.TemporaryDirectory() as temp: receipt_path = Path(temp) / "receipt.json" with ( patch.object( controller.RuntimeCache, "verify_source", return_value="quay.io/argoproj/argocd:v3.3.6", ), patch.object( controller.RuntimeCache, "source_config_digest", return_value=image.source_config_digest, ), patch.object(controller.RuntimeCache, "export") as export, patch.object( controller, "_registry_descriptor", return_value=descriptor ), patch.object(controller, "_registry_digest_present", return_value=True), patch.object(controller, "_run") as run, ): receipt = controller.stage_images( policy, trace_id="aia-p0-006-02b-existing", receipt_path=receipt_path, ssh_key=Path("/unused/key"), known_hosts=Path("/unused/known-hosts"), apply=True, ) self.assertEqual(receipt["state"], "verified") self.assertEqual(receipt["missing_before_count"], 0) self.assertEqual(receipt["images"][0]["execution"], "already_present") self.assertTrue(receipt["images"][0]["provenance_config_digest_match"]) export.assert_not_called() run.assert_not_called() def test_staging_exports_cached_source_and_verifies_remote_provenance( self, ) -> None: policy = self._single_staging_image_policy() image = policy.images[0] descriptor = controller.RegistryDescriptor( digest="sha256:" + "2" * 64, config_digest=image.source_config_digest, ) with tempfile.TemporaryDirectory() as temp: receipt_path = Path(temp) / "receipt.json" with ( patch.object( controller.RuntimeCache, "verify_source", return_value="quay.io/argoproj/argocd:v3.3.6", ), patch.object( controller.RuntimeCache, "source_config_digest", return_value=image.source_config_digest, ), patch.object(controller.RuntimeCache, "export") as export, patch.object( controller, "_registry_descriptor", side_effect=[None, descriptor], ), patch.object(controller, "_local_image_present", return_value=False), patch.object(controller, "_archive_contains_digest", return_value=True), patch.object( controller, "_local_config_digest", return_value=image.source_config_digest, ), patch.object(controller, "_registry_digest_present", return_value=True), patch.object(controller, "_remove_local_image") as remove, patch.object(controller, "_run") as run, ): receipt = controller.stage_images( policy, trace_id="aia-p0-006-02b-stage", receipt_path=receipt_path, ssh_key=Path("/unused/key"), known_hosts=Path("/unused/known-hosts"), apply=True, ) self.assertEqual(receipt["state"], "verified") self.assertEqual(receipt["missing_before_count"], 1) self.assertEqual(receipt["images"][0]["execution"], "staged_from_runtime_cache") self.assertEqual( receipt["images"][0]["target_registry_digest"], descriptor.digest ) export.assert_called_once() commands = [call.args[0] for call in run.call_args_list] self.assertEqual([command[1] for command in commands], ["load", "tag", "push"]) self.assertTrue(all("pull" not in command for command in commands)) self.assertEqual(remove.call_count, 2) def test_staging_rejects_internal_artifact_with_wrong_provenance(self) -> None: policy = self._single_staging_image_policy() image = policy.images[0] descriptor = controller.RegistryDescriptor( digest="sha256:" + "3" * 64, config_digest="sha256:" + "4" * 64, ) with tempfile.TemporaryDirectory() as temp: with ( patch.object( controller.RuntimeCache, "verify_source", return_value="quay.io/argoproj/argocd:v3.3.6", ), patch.object( controller.RuntimeCache, "source_config_digest", return_value=image.source_config_digest, ), patch.object( controller, "_registry_descriptor", return_value=descriptor ), self.assertRaisesRegex( controller.ControllerError, "internal_registry_provenance_mismatch", ), ): controller.stage_images( policy, trace_id="aia-p0-006-02b-wrong-provenance", receipt_path=Path(temp) / "receipt.json", ssh_key=Path("/unused/key"), known_hosts=Path("/unused/known-hosts"), apply=True, ) def test_verified_mirror_receipt_is_bound_to_trace_and_policy(self) -> None: policy = controller.load_policy(POLICY_PATH) receipt = { "schema_version": controller.RECEIPT_SCHEMA_VERSION, "trace_id": "aia-p0-006-02a-test", "work_item_id": policy.work_item_id, "policy_checksum": policy.checksum, "state": "verified", "verified_count": len(policy.images), } with tempfile.TemporaryDirectory() as temp: path = Path(temp) / "receipt.json" path.write_text(json.dumps(receipt), encoding="utf-8") self.assertEqual( controller._read_mirror_receipt(path, policy, "aia-p0-006-02a-test"), receipt, ) with self.assertRaisesRegex( controller.ControllerError, "mirror_receipt_not_verified" ): controller._read_mirror_receipt(path, policy, "different-trace") def test_init_container_read_and_patch_use_init_container_spec(self) -> None: image = self._image_with_container_kind("init_container") kubernetes = controller.Kubernetes( use_sudo=False, kubeconfig=None, server=None, ) workload = { "spec": { "template": { "spec": { "containers": [{"name": "other", "image": "other:image"}], "initContainers": [ { "name": image.workload.container, "image": "source:init", } ], } } } } with patch.object(kubernetes, "workload", return_value=workload): self.assertEqual(kubernetes.current_image(image), "source:init") with patch.object(kubernetes, "run") as run: kubernetes.patch_image(image, image.runtime_ref, dry_run=True) args = run.call_args.args[0] patch_payload = json.loads(args[args.index("--patch") + 1]) pod_spec = patch_payload["spec"]["template"]["spec"] self.assertNotIn("containers", pod_spec) self.assertEqual( pod_spec["initContainers"], [{"name": image.workload.container, "image": image.runtime_ref}], ) 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: image = self._image_with_container_kind("init_container") kubernetes = controller.Kubernetes( use_sudo=False, kubeconfig=None, server=None, ) workload = { "spec": {"selector": {"matchLabels": {"app": "runtime-mirror-test"}}} } pods = { "items": [ { "metadata": {"name": "runtime-mirror-test-1"}, "status": { "initContainerStatuses": [ { "name": image.workload.container, "ready": False, "state": {"terminated": {"exitCode": 0}}, "imageID": "registry.local/repo@" + image.target_registry_digest, } ] }, } ] } with ( patch.object(kubernetes, "workload", return_value=workload), patch.object(kubernetes, "run") as run, ): run.return_value.stdout = json.dumps(pods) self.assertEqual(kubernetes.verify_pods(image), 1) def test_init_container_verifier_rejects_incomplete_or_wrong_digest(self) -> None: image = self._image_with_container_kind("init_container") workload = { "spec": {"selector": {"matchLabels": {"app": "runtime-mirror-test"}}} } cases = ( ( { "state": {"running": {}}, "imageID": "repo@" + image.target_registry_digest, }, "pod_init_container_not_completed", ), ( { "state": {"terminated": {"exitCode": 0}}, "imageID": "repo@sha256:" + "0" * 64, }, "pod_image_digest_mismatch", ), ) for status, error_code in cases: with self.subTest(error_code=error_code): status["name"] = image.workload.container pods = { "items": [ {"metadata": {}, "status": {"initContainerStatuses": [status]}} ] } kubernetes = controller.Kubernetes( use_sudo=False, kubeconfig=None, server=None, ) with ( patch.object(kubernetes, "workload", return_value=workload), patch.object(kubernetes, "run") as run, self.assertRaisesRegex(controller.ControllerError, error_code), ): 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) 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()