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 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.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"), 11) self.assertEqual(container_kinds.count("init_container"), 3) 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_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), 3) 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: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") 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_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_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_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) if __name__ == "__main__": unittest.main()