feat(security): stage Redis runtime cache canary
Some checks failed
CD Pipeline / workflow-shape (push) Successful in 0s
CD Pipeline / cancel-stale-cd (push) Has been skipped
CD Pipeline / build-and-deploy (push) Has been cancelled
CD Pipeline / tests (push) Has been cancelled
CD Pipeline / post-deploy-checks (push) Has been cancelled

This commit is contained in:
ogt
2026-07-15 02:29:53 +08:00
parent 0f681ca277
commit 19b7445c1a
2 changed files with 54 additions and 10 deletions

View File

@@ -6,6 +6,7 @@ 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
@@ -41,6 +42,10 @@ class RuntimeImageMirrorControllerTest(unittest.TestCase):
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")
@@ -85,7 +90,7 @@ class RuntimeImageMirrorControllerTest(unittest.TestCase):
):
self._image_with_container_kind("ephemeral_container")
def test_staging_policy_is_runtime_cache_only_and_init_canary_scoped(
def test_staging_policy_is_runtime_cache_only_and_bounded_canary_scoped(
self,
) -> None:
policy = controller.load_staging_policy(STAGING_POLICY_PATH)
@@ -93,12 +98,38 @@ 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), 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.assertEqual(len(policy.images), 2)
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",
},
)
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")
self.assertNotIn("github.com", raw)
self.assertNotIn("ghcr.io", raw)
self.assertIn('"external_pull_allowed": false', raw)
@@ -428,7 +459,7 @@ class RuntimeImageMirrorControllerTest(unittest.TestCase):
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)
policy = self._single_staging_image_policy()
image = policy.images[0]
descriptor = controller.RegistryDescriptor(
digest="sha256:" + "1" * 64,
@@ -473,7 +504,7 @@ class RuntimeImageMirrorControllerTest(unittest.TestCase):
def test_staging_exports_cached_source_and_verifies_remote_provenance(
self,
) -> None:
policy = controller.load_staging_policy(STAGING_POLICY_PATH)
policy = self._single_staging_image_policy()
image = policy.images[0]
descriptor = controller.RegistryDescriptor(
digest="sha256:" + "2" * 64,
@@ -531,7 +562,7 @@ class RuntimeImageMirrorControllerTest(unittest.TestCase):
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)
policy = self._single_staging_image_policy()
image = policy.images[0]
descriptor = controller.RegistryDescriptor(
digest="sha256:" + "3" * 64,