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

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

Browse Source
This commit is contained in:
ogt
2026-07-14 21:37:32 +08:00
parent c8edd73971
commit e54e0f3a27
4 changed files with 561 additions and 7 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -14,6 +14,7 @@ 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():
@@ -84,6 +85,24 @@ class RuntimeImageMirrorControllerTest(unittest.TestCase):
):
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:
@@ -174,6 +193,8 @@ class RuntimeImageMirrorControllerTest(unittest.TestCase):
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(
@@ -182,6 +203,145 @@ 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)
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 = {