From c8edd739719cb3409d7898ee7dc451a8cb376bda Mon Sep 17 00:00:00 2001 From: ogt Date: Tue, 14 Jul 2026 21:26:13 +0800 Subject: [PATCH] feat(security): verify mirrored init containers --- .../runtime_image_mirror_controller.py | 40 ++++- .../test_runtime_image_mirror_controller.py | 139 ++++++++++++++++++ 2 files changed, 174 insertions(+), 5 deletions(-) diff --git a/scripts/security/runtime_image_mirror_controller.py b/scripts/security/runtime_image_mirror_controller.py index 7565356cd..2ea86f509 100644 --- a/scripts/security/runtime_image_mirror_controller.py +++ b/scripts/security/runtime_image_mirror_controller.py @@ -20,6 +20,7 @@ from typing import Any, Sequence SCHEMA_VERSION = "awoooi_runtime_image_mirror_policy_v1" RECEIPT_SCHEMA_VERSION = "awoooi_runtime_image_mirror_receipt_v1" ALLOWED_KINDS = {"deployment", "daemonset", "statefulset"} +ALLOWED_CONTAINER_KINDS = {"container", "init_container"} DIGEST_PATTERN = re.compile(r"^sha256:[0-9a-f]{64}$") NAME_PATTERN = re.compile(r"^[a-z0-9](?:[a-z0-9.-]*[a-z0-9])?$") TRACE_PATTERN = re.compile(r"^[A-Za-z0-9._:-]{3,160}$") @@ -35,6 +36,7 @@ class Workload: namespace: str name: str container: str + container_kind: str @dataclass(frozen=True) @@ -119,7 +121,7 @@ def load_policy(path: Path) -> Policy: raise ControllerError("image_policy_empty") images: list[ImagePolicy] = [] asset_ids: set[str] = set() - workloads: set[tuple[str, str, str, str]] = set() + workloads: set[tuple[str, str, str, str, str]] = set() for row in image_rows: if not isinstance(row, dict) or not isinstance(row.get("workload"), dict): raise ControllerError("image_policy_invalid") @@ -152,12 +154,18 @@ def load_policy(path: Path) -> Policy: namespace=_require_name(workload_row.get("namespace"), "namespace"), name=_require_name(workload_row.get("name"), "workload_name"), container=_require_name(workload_row.get("container"), "container"), + container_kind=_require_text( + workload_row.get("container_kind", "container"), "container_kind" + ).lower(), ) + if workload.container_kind not in ALLOWED_CONTAINER_KINDS: + raise ControllerError("container_kind_not_allowed") workload_key = ( workload.kind, workload.namespace, workload.name, workload.container, + workload.container_kind, ) if workload_key in workloads: raise ControllerError("duplicate_workload_target") @@ -476,13 +484,26 @@ class Kubernetes: except json.JSONDecodeError as exc: raise ControllerError("workload_readback_invalid") from exc + @staticmethod + def _spec_container_field(item: ImagePolicy) -> str: + if item.workload.container_kind == "init_container": + return "initContainers" + return "containers" + + @staticmethod + def _status_container_field(item: ImagePolicy) -> str: + if item.workload.container_kind == "init_container": + return "initContainerStatuses" + return "containerStatuses" + def current_image(self, item: ImagePolicy) -> str: workload = self.workload(item) + container_field = self._spec_container_field(item) containers = ( workload.get("spec", {}) .get("template", {}) .get("spec", {}) - .get("containers", []) + .get(container_field, []) ) values = [ row.get("image") @@ -494,12 +515,13 @@ class Kubernetes: return values[0] def patch_image(self, item: ImagePolicy, image_ref: str, *, dry_run: bool) -> None: + container_field = self._spec_container_field(item) patch = json.dumps( { "spec": { "template": { "spec": { - "containers": [ + container_field: [ {"name": item.workload.container, "image": image_ref} ] } @@ -566,7 +588,8 @@ class Kubernetes: raise ControllerError("workload_pods_missing") verified = 0 for pod in active: - statuses = pod.get("status", {}).get("containerStatuses", []) + status_field = self._status_container_field(item) + statuses = pod.get("status", {}).get(status_field, []) matches = [ status for status in statuses @@ -576,7 +599,14 @@ class Kubernetes: if len(matches) != 1: raise ControllerError("pod_container_status_missing") status = matches[0] - if status.get("ready") is not True: + if item.workload.container_kind == "init_container": + terminated = status.get("state", {}).get("terminated", {}) + completed = status.get("ready") is True or ( + isinstance(terminated, dict) and terminated.get("exitCode") == 0 + ) + if not completed: + raise ControllerError("pod_init_container_not_completed") + elif status.get("ready") is not True: raise ControllerError("pod_container_not_ready") image_id = status.get("imageID") if not isinstance(image_id, str) or not image_id.endswith( diff --git a/scripts/security/tests/test_runtime_image_mirror_controller.py b/scripts/security/tests/test_runtime_image_mirror_controller.py index e3a715c14..6476d402a 100644 --- a/scripts/security/tests/test_runtime_image_mirror_controller.py +++ b/scripts/security/tests/test_runtime_image_mirror_controller.py @@ -31,6 +31,15 @@ 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") @@ -61,6 +70,19 @@ class RuntimeImageMirrorControllerTest(unittest.TestCase): 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_policy_rejects_external_pull_and_mutable_runtime_ref(self) -> None: payload = json.loads(POLICY_PATH.read_text(encoding="utf-8")) @@ -182,6 +204,123 @@ class RuntimeImageMirrorControllerTest(unittest.TestCase): ): 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()