diff --git a/.gitea/workflows/cd.yaml b/.gitea/workflows/cd.yaml index ef0f9a6ed..5009bbb42 100644 --- a/.gitea/workflows/cd.yaml +++ b/.gitea/workflows/cd.yaml @@ -1647,6 +1647,20 @@ jobs: echo "BLOCKER runtime_image_cache_host_key_unavailable" exit 1 } + STAGING_TRACE_ID="aia-p0-006-02b-${GITHUB_RUN_ID:-${GITEA_RUN_NUMBER:-unknown}}" + STAGING_RECEIPT_PATH="/tmp/awoooi-runtime-image-staging-receipt.json" + rm -f "${STAGING_RECEIPT_PATH}" + python3 scripts/security/runtime_image_mirror_controller.py \ + --policy config/security/runtime-image-mirror-staging-policy.json \ + --trace-id "${STAGING_TRACE_ID}" \ + stage \ + --apply \ + --receipt "${STAGING_RECEIPT_PATH}" \ + --ssh-key "${HOME}/.ssh/deploy_key" \ + --known-hosts "${CACHE_KNOWN_HOSTS}" + test -s "${STAGING_RECEIPT_PATH}" + echo "runtime_image_staging_stage=verified" + TRACE_ID="aia-p0-006-02a-${GITHUB_RUN_ID:-${GITEA_RUN_NUMBER:-unknown}}" RECEIPT_PATH="/tmp/awoooi-runtime-image-mirror-receipt.json" rm -f "${RECEIPT_PATH}" diff --git a/config/security/runtime-image-mirror-staging-policy.json b/config/security/runtime-image-mirror-staging-policy.json new file mode 100644 index 000000000..34b373a88 --- /dev/null +++ b/config/security/runtime-image-mirror-staging-policy.json @@ -0,0 +1,33 @@ +{ + "schema_version": "awoooi_runtime_image_mirror_staging_policy_v1", + "work_item_id": "AIA-P0-006-02B", + "risk_level": "high", + "source_contract": { + "origin": "runtime_cache_only", + "external_pull_allowed": false, + "cache_host": "192.168.0.120", + "cache_user": "wooo", + "platform": "linux/amd64" + }, + "target_contract": { + "push_registry_prefix": "registry.wooo.work/awoooi/runtime-mirror/", + "runtime_registry_prefix": "192.168.0.110:5000/awoooi/runtime-mirror/", + "digest_pin_required": true + }, + "images": [ + { + "asset_id": "runtime-image:argocd-v3.3.6-canary", + "source_index_digest": "sha256:16b92ba472fbb9287459cc52e0ecff07288dff461209955098edb56ce866fe49", + "platform_digest": "sha256:2f25a42949ea69c0dd33f4ce1918c6a01039d6c14a7ecc1d19088504a9d3e94f", + "source_config_digest": "sha256:86219ff1fefea5c8e09a4791d1093f06fca7ac8b1b2d1c56f94d3cb560a77603", + "push_ref": "registry.wooo.work/awoooi/runtime-mirror/argocd:v3.3.6-linux-amd64", + "workload": { + "kind": "deployment", + "namespace": "argocd", + "name": "argocd-dex-server", + "container": "copyutil", + "container_kind": "init_container" + } + } + ] +} diff --git a/scripts/security/runtime_image_mirror_controller.py b/scripts/security/runtime_image_mirror_controller.py index 2ea86f509..6438bf040 100644 --- a/scripts/security/runtime_image_mirror_controller.py +++ b/scripts/security/runtime_image_mirror_controller.py @@ -19,6 +19,8 @@ from typing import Any, Sequence SCHEMA_VERSION = "awoooi_runtime_image_mirror_policy_v1" RECEIPT_SCHEMA_VERSION = "awoooi_runtime_image_mirror_receipt_v1" +STAGING_SCHEMA_VERSION = "awoooi_runtime_image_mirror_staging_policy_v1" +STAGING_RECEIPT_SCHEMA_VERSION = "awoooi_runtime_image_mirror_staging_receipt_v1" ALLOWED_KINDS = {"deployment", "daemonset", "statefulset"} ALLOWED_CONTAINER_KINDS = {"container", "init_container"} DIGEST_PATTERN = re.compile(r"^sha256:[0-9a-f]{64}$") @@ -50,6 +52,17 @@ class ImagePolicy: workload: Workload +@dataclass(frozen=True) +class StagingImage: + asset_id: str + source_index_digest: str + platform_digest: str + source_config_digest: str + push_ref: str + runtime_repository: str + workload: Workload + + @dataclass(frozen=True) class Policy: work_item_id: str @@ -61,6 +74,24 @@ class Policy: checksum: str +@dataclass(frozen=True) +class StagingPolicy: + work_item_id: str + risk_level: str + cache_host: str + cache_user: str + platform: str + runtime_registry_prefix: str + images: tuple[StagingImage, ...] + checksum: str + + +@dataclass(frozen=True) +class RegistryDescriptor: + digest: str + config_digest: str + + def _require_text(value: Any, label: str) -> str: if not isinstance(value, str) or not value.strip(): raise ControllerError(f"invalid_{label}") @@ -193,6 +224,121 @@ def load_policy(path: Path) -> Policy: ) +def load_staging_policy(path: Path) -> StagingPolicy: + try: + payload = json.loads(path.read_text(encoding="utf-8")) + except (OSError, json.JSONDecodeError) as exc: + raise ControllerError("staging_policy_unreadable") from exc + if ( + not isinstance(payload, dict) + or payload.get("schema_version") != STAGING_SCHEMA_VERSION + ): + raise ControllerError("staging_policy_schema_invalid") + + source = payload.get("source_contract") + target = payload.get("target_contract") + if not isinstance(source, dict) or not isinstance(target, dict): + raise ControllerError("staging_policy_contract_missing") + if source.get("origin") != "runtime_cache_only": + raise ControllerError("runtime_cache_only_required") + if source.get("external_pull_allowed") is not False: + raise ControllerError("external_pull_must_be_disabled") + if source.get("platform") != "linux/amd64": + raise ControllerError("unsupported_runtime_platform") + if target.get("digest_pin_required") is not True: + raise ControllerError("digest_pin_required") + + push_prefix = _require_text(target.get("push_registry_prefix"), "push_prefix") + runtime_prefix = _require_text( + target.get("runtime_registry_prefix"), "runtime_prefix" + ) + if push_prefix != "registry.wooo.work/awoooi/runtime-mirror/": + raise ControllerError("push_registry_not_internal") + if runtime_prefix != "192.168.0.110:5000/awoooi/runtime-mirror/": + raise ControllerError("runtime_registry_not_internal") + + image_rows = payload.get("images") + if not isinstance(image_rows, list) or not image_rows: + raise ControllerError("staging_image_policy_empty") + images: list[StagingImage] = [] + asset_ids: set[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("staging_image_policy_invalid") + asset_id = _require_text(row.get("asset_id"), "asset_id") + if asset_id in asset_ids: + raise ControllerError("duplicate_asset_id") + asset_ids.add(asset_id) + + push_ref = _require_text(row.get("push_ref"), "push_ref") + if not push_ref.startswith(push_prefix) or "@" in push_ref: + raise ControllerError("push_ref_invalid") + push_suffix = push_ref.removeprefix(push_prefix) + if ":" not in push_suffix: + raise ControllerError("push_ref_tag_required") + runtime_repository, push_tag = push_suffix.rsplit(":", 1) + if not runtime_repository or any( + not NAME_PATTERN.fullmatch(part) for part in runtime_repository.split("/") + ): + raise ControllerError("runtime_repository_invalid") + _require_name(push_tag, "push_tag") + + workload_row = row["workload"] + kind = _require_text(workload_row.get("kind"), "workload_kind").lower() + if kind not in ALLOWED_KINDS: + raise ControllerError("workload_kind_not_allowed") + workload = Workload( + kind=kind, + 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") + workloads.add(workload_key) + images.append( + StagingImage( + asset_id=asset_id, + source_index_digest=_require_digest( + row.get("source_index_digest"), "source_index_digest" + ), + platform_digest=_require_digest( + row.get("platform_digest"), "platform_digest" + ), + source_config_digest=_require_digest( + row.get("source_config_digest"), "source_config_digest" + ), + push_ref=push_ref, + runtime_repository=runtime_repository, + workload=workload, + ) + ) + + return StagingPolicy( + work_item_id=_require_text(payload.get("work_item_id"), "work_item_id"), + risk_level=_require_text(payload.get("risk_level"), "risk_level"), + cache_host=_require_text(source.get("cache_host"), "cache_host"), + cache_user=_require_name(source.get("cache_user"), "cache_user"), + platform="linux/amd64", + runtime_registry_prefix=runtime_prefix, + images=tuple(images), + checksum=_policy_checksum(payload), + ) + + def _run( command: Sequence[str], *, @@ -222,7 +368,9 @@ def _run( class RuntimeCache: - def __init__(self, policy: Policy, ssh_key: Path, known_hosts: Path) -> None: + def __init__( + self, policy: Policy | StagingPolicy, ssh_key: Path, known_hosts: Path + ) -> None: self.policy = policy self.ssh = [ "ssh", @@ -254,7 +402,7 @@ class RuntimeCache: raise ControllerError("runtime_cache_digest_missing") return sorted(set(matches))[0] - def verify_source(self, image: ImagePolicy) -> str: + def verify_source(self, image: ImagePolicy | StagingImage) -> str: source_ref = self.source_ref(image.source_index_digest) check = _run(self._command("images", "check", f"name=={source_ref}")) if "complete" not in (check.stdout or ""): @@ -277,6 +425,17 @@ class RuntimeCache: raise ControllerError("runtime_cache_platform_digest_mismatch") return source_ref + def source_config_digest(self, platform_digest: str) -> str: + content = _run(self._command("content", "get", platform_digest)).stdout + try: + manifest = json.loads(content or "{}") + except json.JSONDecodeError as exc: + raise ControllerError("runtime_cache_platform_manifest_invalid") from exc + config = manifest.get("config") + if not isinstance(config, dict): + raise ControllerError("runtime_cache_config_digest_missing") + return _require_digest(config.get("digest"), "runtime_cache_config_digest") + def export(self, source_ref: str, output: Path) -> None: with output.open("wb") as handle: completed = subprocess.run( @@ -304,7 +463,7 @@ def _target_digest_ref(image: ImagePolicy) -> str: return f"{_target_repository(image.push_ref)}@{image.target_registry_digest}" -def _target_digest_present(image: ImagePolicy) -> bool: +def _registry_digest_present(digest_ref: str) -> bool: try: completed = subprocess.run( [ @@ -312,7 +471,7 @@ def _target_digest_present(image: ImagePolicy) -> bool: "manifest", "inspect", "--insecure", - _target_digest_ref(image), + digest_ref, ], check=False, stdout=subprocess.DEVNULL, @@ -324,6 +483,52 @@ def _target_digest_present(image: ImagePolicy) -> bool: return completed.returncode == 0 +def _target_digest_present(image: ImagePolicy) -> bool: + return _registry_digest_present(_target_digest_ref(image)) + + +def _registry_descriptor(image_ref: str) -> RegistryDescriptor | None: + try: + completed = subprocess.run( + ["docker", "manifest", "inspect", "--verbose", image_ref], + check=False, + text=True, + stdout=subprocess.PIPE, + stderr=subprocess.DEVNULL, + timeout=20, + ) + except (OSError, subprocess.TimeoutExpired) as exc: + raise ControllerError("internal_registry_descriptor_unavailable") from exc + if completed.returncode != 0: + return None + try: + payload = json.loads(completed.stdout or "{}") + except json.JSONDecodeError as exc: + raise ControllerError("internal_registry_descriptor_invalid") from exc + if not isinstance(payload, dict): + raise ControllerError("internal_registry_descriptor_invalid") + descriptor = payload.get("Descriptor") + manifest = payload.get("SchemaV2Manifest") + if not isinstance(descriptor, dict) or not isinstance(manifest, dict): + raise ControllerError("internal_registry_descriptor_invalid") + config = manifest.get("config") + if not isinstance(config, dict): + raise ControllerError("internal_registry_descriptor_invalid") + return RegistryDescriptor( + digest=_require_digest( + descriptor.get("digest"), "internal_registry_descriptor_digest" + ), + config_digest=_require_digest( + config.get("digest"), "internal_registry_config_digest" + ), + ) + + +def _local_config_digest(image_ref: str) -> str: + output = _run(["docker", "image", "inspect", "--format={{.Id}}", image_ref]).stdout + return _require_digest((output or "").strip(), "local_config_digest") + + def _local_image_present(image_ref: str) -> bool: try: completed = subprocess.run( @@ -445,6 +650,127 @@ def mirror_images( return receipt +def stage_images( + policy: StagingPolicy, + *, + trace_id: str, + receipt_path: Path, + ssh_key: Path, + known_hosts: Path, + apply: bool, +) -> dict[str, Any]: + cache = RuntimeCache(policy, ssh_key, known_hosts) + image_receipts: list[dict[str, Any]] = [] + missing_count = 0 + for image in policy.images: + source_ref = cache.verify_source(image) + source_fingerprint = "sha256:" + hashlib.sha256(source_ref.encode()).hexdigest() + source_config_digest = cache.source_config_digest(image.platform_digest) + if source_config_digest != image.source_config_digest: + raise ControllerError("runtime_cache_config_digest_mismatch") + + descriptor = _registry_descriptor(image.push_ref) + execution = "already_present" + if descriptor is None: + missing_count += 1 + execution = "check_only_missing" + if apply: + source_preexisting = _local_image_present(source_ref) + try: + with tempfile.TemporaryDirectory( + prefix="awoooi-runtime-image-stage-" + ) as temp: + archive = Path(temp) / "image.oci.tar" + cache.export(source_ref, archive) + if not _archive_contains_digest(archive, image.platform_digest): + raise ControllerError("runtime_cache_export_digest_missing") + _run( + ["docker", "load", "--input", str(archive)], + quiet=True, + ) + if _local_config_digest(source_ref) != source_config_digest: + raise ControllerError("local_image_config_digest_mismatch") + _run( + ["docker", "tag", source_ref, image.push_ref], + quiet=True, + ) + _run(["docker", "push", image.push_ref], quiet=True) + finally: + _remove_local_image(image.push_ref) + if not source_preexisting: + _remove_local_image(source_ref) + descriptor = _registry_descriptor(image.push_ref) + if descriptor is None: + raise ControllerError("internal_registry_descriptor_missing") + execution = "staged_from_runtime_cache" + + if descriptor is not None and descriptor.config_digest != source_config_digest: + raise ControllerError("internal_registry_provenance_mismatch") + target_digest_ref = None + runtime_ref = None + target_verified = False + if descriptor is not None: + target_digest_ref = ( + f"{_target_repository(image.push_ref)}@{descriptor.digest}" + ) + runtime_ref = ( + f"{policy.runtime_registry_prefix}{image.runtime_repository}" + f"@{descriptor.digest}" + ) + target_verified = _registry_digest_present(target_digest_ref) + if apply and not target_verified: + raise ControllerError("internal_registry_digest_verifier_failed") + image_receipts.append( + { + "asset_id": image.asset_id, + "source_ref_fingerprint": source_fingerprint, + "source_index_digest": image.source_index_digest, + "platform_digest": image.platform_digest, + "source_config_digest": source_config_digest, + "target_registry_digest": ( + descriptor.digest if descriptor is not None else None + ), + "target_digest_ref": target_digest_ref, + "runtime_ref": runtime_ref, + "source_cache_verified": True, + "target_digest_verified": target_verified, + "provenance_config_digest_match": descriptor is not None, + "execution": execution, + "workload": { + "kind": image.workload.kind, + "namespace": image.workload.namespace, + "name": image.workload.name, + "container": image.workload.container, + "container_kind": image.workload.container_kind, + }, + } + ) + + verified = all(row["target_digest_verified"] for row in image_receipts) + receipt = { + "schema_version": STAGING_RECEIPT_SCHEMA_VERSION, + "trace_id": trace_id, + "work_item_id": policy.work_item_id, + "policy_checksum": policy.checksum, + "stage": "runtime_cache_staging", + "mode": "apply" if apply else "check", + "risk_level": policy.risk_level, + "external_pull_allowed": False, + "source_cache_only": True, + "candidate_count": len(image_receipts), + "missing_before_count": missing_count, + "verified_count": sum( + 1 for row in image_receipts if row["target_digest_verified"] + ), + "state": "verified" if verified else "blocked_with_safe_next_action", + "images": image_receipts, + } + receipt_path.write_text( + json.dumps(receipt, ensure_ascii=True, indent=2) + "\n", encoding="utf-8" + ) + return receipt + + class Kubernetes: def __init__( self, @@ -745,6 +1071,12 @@ def build_parser() -> argparse.ArgumentParser: mirror.add_argument("--ssh-key", type=Path, required=True) mirror.add_argument("--known-hosts", type=Path, required=True) + stage = subparsers.add_parser("stage") + stage.add_argument("--apply", action="store_true") + stage.add_argument("--receipt", type=Path, required=True) + stage.add_argument("--ssh-key", type=Path, required=True) + stage.add_argument("--known-hosts", type=Path, required=True) + apply_parser = subparsers.add_parser("apply") apply_parser.add_argument("--apply", action="store_true") apply_parser.add_argument("--mirror-receipt", type=Path, required=True) @@ -757,7 +1089,18 @@ def build_parser() -> argparse.ArgumentParser: def main(argv: Sequence[str] | None = None) -> int: args = build_parser().parse_args(argv) try: - policy = load_policy(args.policy) + if args.command == "stage": + staging_policy = load_staging_policy(args.policy) + receipt = stage_images( + staging_policy, + trace_id=args.trace_id, + receipt_path=args.receipt, + ssh_key=args.ssh_key, + known_hosts=args.known_hosts, + apply=args.apply, + ) + else: + policy = load_policy(args.policy) if args.command == "mirror": receipt = mirror_images( policy, @@ -767,7 +1110,7 @@ def main(argv: Sequence[str] | None = None) -> int: known_hosts=args.known_hosts, apply=args.apply, ) - else: + elif args.command == "apply": receipt = apply_policy( policy, trace_id=args.trace_id, @@ -783,7 +1126,11 @@ def main(argv: Sequence[str] | None = None) -> int: print( json.dumps( { - "schema_version": RECEIPT_SCHEMA_VERSION, + "schema_version": ( + STAGING_RECEIPT_SCHEMA_VERSION + if args.command == "stage" + else RECEIPT_SCHEMA_VERSION + ), "state": "blocked_with_safe_next_action", "error_code": str(exc), "secret_value_exposed": False, diff --git a/scripts/security/tests/test_runtime_image_mirror_controller.py b/scripts/security/tests/test_runtime_image_mirror_controller.py index 6476d402a..0a5e9b375 100644 --- a/scripts/security/tests/test_runtime_image_mirror_controller.py +++ b/scripts/security/tests/test_runtime_image_mirror_controller.py @@ -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 = {