fix(security): stage multi-platform runtime image
Some checks failed
CD Pipeline / workflow-shape (push) Successful in 0s
CD Pipeline / cancel-stale-cd (push) Has been skipped
CD Pipeline / tests (push) Successful in 1m16s
CD Pipeline / build-and-deploy (push) Failing after 4m9s
CD Pipeline / post-deploy-checks (push) Has been skipped

This commit is contained in:
ogt
2026-07-15 09:22:14 +08:00
parent dcaecb6801
commit 41c069875c
3 changed files with 135 additions and 13 deletions

View File

@@ -238,7 +238,7 @@ class RuntimeImageMirrorControllerTest(unittest.TestCase):
self.assertEqual(policy.work_item_id, "AIA-P0-006-02B")
self.assertEqual(policy.risk_level, "high")
self.assertEqual(len(policy.images), 5)
self.assertEqual(len(policy.images), 6)
by_asset = {image.asset_id: image for image in policy.images}
self.assertEqual(
set(by_asset),
@@ -247,6 +247,7 @@ class RuntimeImageMirrorControllerTest(unittest.TestCase):
"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:sealed-secrets-controller-0.26.0-canary",
"runtime-image:vpa-recommender-1.6.0-canary",
},
)
@@ -335,6 +336,32 @@ class RuntimeImageMirrorControllerTest(unittest.TestCase):
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")
sealed_secrets = by_asset[
"runtime-image:sealed-secrets-controller-0.26.0-canary"
]
self.assertEqual(
sealed_secrets.source_index_digest,
"sha256:74cd190f424b591dd82a234685c8c81d50b33af807e03470bc12b23d202e0106",
)
self.assertEqual(
sealed_secrets.platform_digest,
"sha256:1827b36853e124ac0dd2554f7cd55c04952ecb7c38ff19a4a8d93d633811ed68",
)
self.assertEqual(
sealed_secrets.source_config_digest,
"sha256:bbfc8314cb4978b252fb313b84e1d51a7b0c040befd79ecc903ea952485a3348",
)
self.assertEqual(
sealed_secrets.runtime_repository, "sealed-secrets-controller"
)
self.assertEqual(sealed_secrets.workload.kind, "deployment")
self.assertEqual(sealed_secrets.workload.namespace, "kube-system")
self.assertEqual(sealed_secrets.workload.name, "sealed-secrets-controller")
self.assertEqual(
sealed_secrets.workload.container, "sealed-secrets-controller"
)
self.assertEqual(sealed_secrets.workload.container_kind, "container")
self.assertNotIn("github.com", raw)
self.assertNotIn("ghcr.io", raw)
self.assertIn('"external_pull_allowed": false', raw)
@@ -711,6 +738,26 @@ class RuntimeImageMirrorControllerTest(unittest.TestCase):
)
)
def test_local_platform_digest_reads_platform_descriptor(self) -> None:
digest = "sha256:" + "1" * 64
completed = controller.subprocess.CompletedProcess(
args=[],
returncode=0,
stdout=json.dumps({"digest": digest}),
stderr="",
)
with patch.object(controller, "_run", return_value=completed) as run:
self.assertEqual(
controller._local_platform_digest("source:tag", "linux/amd64"),
digest,
)
command = run.call_args.args[0]
self.assertEqual(command[:3], ["docker", "image", "inspect"])
self.assertIn("--platform", command)
self.assertIn("linux/amd64", command)
self.assertIn("--format={{json .Descriptor}}", command)
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:
@@ -888,11 +935,13 @@ class RuntimeImageMirrorControllerTest(unittest.TestCase):
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, "_archive_contains_digest", return_value=True
) as archive_contains,
patch.object(
controller,
"_local_config_digest",
return_value=image.source_config_digest,
"_local_platform_digest",
return_value=image.platform_digest,
),
patch.object(controller, "_registry_digest_present", return_value=True),
patch.object(controller, "_remove_local_image") as remove,
@@ -916,6 +965,11 @@ class RuntimeImageMirrorControllerTest(unittest.TestCase):
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.assertEqual(archive_contains.call_count, 2)
self.assertIn("--platform", commands[0])
self.assertIn(policy.platform, commands[0])
self.assertIn("--platform", commands[2])
self.assertIn(policy.platform, commands[2])
self.assertTrue(all("pull" not in command for command in commands))
self.assertEqual(remove.call_count, 2)