wooo/awoooi
1
1
Fork 0
Code Issues 2 Pull Requests 1 Actions 3 Packages Projects Releases Wiki Activity

feat(security): verify mirrored init containers

Browse Source
This commit is contained in:
ogt
2026-07-14 21:26:13 +08:00
parent 5b86dbbf17
commit c8edd73971
2 changed files with 174 additions and 5 deletions
Download Patch File Download Diff File Expand all files Collapse all files

139
scripts/security/tests/test_runtime_image_mirror_controller.py
View File

@@ -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()