from __future__ import annotations import importlib.util import json import sys import tarfile import tempfile import unittest 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 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), 5) 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 ) ) self.assertTrue( all(image.workload.container_kind == "container" for image in policy.images) ) 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_init_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), 1) image = policy.images[0] self.assertEqual(image.runtime_repository, "argocd") self.assertEqual(image.workload.name, "argocd-dex-server") self.assertEqual(image.workload.container, "copyutil") self.assertEqual(image.workload.container_kind, "init_container") self.assertNotIn("github.com", raw) self.assertNotIn("ghcr.io", raw) self.assertIn('"external_pull_allowed": false', raw) 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 = controller.load_staging_policy(STAGING_POLICY_PATH) 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 = controller.load_staging_policy(STAGING_POLICY_PATH) 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 = controller.load_staging_policy(STAGING_POLICY_PATH) 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()