Coverage for scripts / live_release_validation / ownership / ecr.py: 100.00%
297 statements
« prev ^ index » next coverage.py v7.13.5, created at 2026-10-28 16:19 +0000
1"""ECR expectation, ownership checkpointing, and baseline stripping."""
3from __future__ import annotations
5import copy
6import json
7import re
8from collections.abc import Mapping
9from pathlib import Path
10from typing import Any
12from ..constants import (
13 _RUN_STACK_TAG,
14)
15from ..context import (
16 _topology_regions,
17)
18from ..inventory import (
19 collect_ecr_inventory,
20)
21from ..models import RunContext
22from ..protected import (
23 _EC2_TAGGED_RESOURCE_IDENTITIES,
24 _PROTECTED_GLOBAL_RESOURCE_CATEGORIES,
25 _PROTECTED_REGIONAL_RESOURCE_CATEGORIES,
26 _baseline_protected_identities,
27 _ec2_tagged_resource_identity,
28 _eks_pod_parent_cluster,
29 _matches_protected_physical_identity,
30 _tagged_resource_is_protected,
31)
33#: Checkpoint key for the image targets of a release the run deployed before
34#: the validated checkout (the upgrade harness's base release).
35PRIOR_RELEASE_ECR_IMAGES_KEY = "prior_release_ecr_images"
37#: The chart catalogue the image mirror derives its images from, relative to
38#: a checkout (``cli/_image_mirror.py`` reads its own checkout's copy).
39_CHARTS_YAML = Path("lambda") / "helm-installer" / "charts.yaml"
42def _strip_baseline_ecr(
43 project_inventory: dict[str, Any], baseline: dict[str, Any]
44) -> dict[str, Any]:
45 """Strip only exact protected identities and authoritatively absent tag records."""
46 protected_stack_ids, protected_resource_ids = _baseline_protected_identities(baseline)
47 baseline_ecr_names: dict[str, set[str]] = {}
48 baseline_ecr_arns: dict[str, set[str]] = {}
49 for raw_region, repositories in (baseline.get("ecr_repositories") or {}).items():
50 region = str(raw_region)
51 for repository in repositories:
52 name = str(repository.get("name") or "")
53 arn = str(repository.get("arn") or "")
54 if name:
55 baseline_ecr_names.setdefault(region, set()).add(name)
56 if arn:
57 baseline_ecr_arns.setdefault(region, set()).add(arn)
59 inventory = copy.deepcopy(project_inventory)
60 stacks_by_region = inventory.get("cloudformation_stacks")
61 if isinstance(stacks_by_region, dict):
62 for region, stacks in list(stacks_by_region.items()):
63 exact_stack_ids = protected_stack_ids.get(str(region), set())
64 remaining = [
65 stack for stack in stacks if str(stack.get("stack_id") or "") not in exact_stack_ids
66 ]
67 if remaining:
68 stacks_by_region[region] = remaining
69 else:
70 stacks_by_region.pop(region)
72 authoritative_clusters = inventory.get("authoritative_eks_clusters")
73 authoritative_ec2_resources = inventory.get("authoritative_ec2_resources")
74 authority_scope = inventory.get("authority_scope")
75 expected_partition = (
76 str(authority_scope.get("partition") or "") if isinstance(authority_scope, Mapping) else ""
77 )
78 expected_account = (
79 str(authority_scope.get("account") or "") if isinstance(authority_scope, Mapping) else ""
80 )
81 has_exact_authority_scope = bool(
82 expected_partition and re.fullmatch(r"\d{12}", expected_account)
83 )
84 coverage = inventory.get("coverage")
85 coverage_complete = isinstance(coverage, Mapping) and coverage.get("complete") is True
86 completed_scanners = (
87 {str(scanner) for scanner in coverage.get("completed_scanners", [])}
88 if isinstance(coverage, Mapping) and isinstance(coverage.get("completed_scanners"), list)
89 else set()
90 )
91 scanner_regions = coverage.get("scanner_regions") if isinstance(coverage, Mapping) else None
92 eks_scanner_regions = (
93 {str(region) for region in scanner_regions.get("eks_clusters", [])}
94 if isinstance(scanner_regions, Mapping)
95 and isinstance(scanner_regions.get("eks_clusters"), list)
96 else set()
97 )
98 instance_scanner_regions = (
99 {str(region) for region in scanner_regions.get("ec2_instances", [])}
100 if isinstance(scanner_regions, Mapping)
101 and isinstance(scanner_regions.get("ec2_instances"), list)
102 else set()
103 )
104 network_scanner_regions = (
105 {str(region) for region in scanner_regions.get("ec2_networking", [])}
106 if isinstance(scanner_regions, Mapping)
107 and isinstance(scanner_regions.get("ec2_networking"), list)
108 else set()
109 )
110 has_complete_eks_authority = coverage_complete and "eks_clusters" in completed_scanners
111 has_complete_ec2_authority = coverage_complete and {
112 "ec2_instances",
113 "ec2_networking",
114 }.issubset(completed_scanners)
115 ec2_scanner_regions = instance_scanner_regions & network_scanner_regions
116 for region, resources in list(inventory.get("regional", {}).items()):
117 region_key = str(region)
118 region_stack_ids = protected_stack_ids.get(region_key, set())
119 region_resource_ids = protected_resource_ids.get(region_key, {})
120 exact_tagged_arns = {
121 physical_id
122 for physical_ids in region_resource_ids.values()
123 for physical_id in physical_ids
124 if physical_id.startswith("arn:")
125 }
126 exact_tagged_arns.update(region_stack_ids)
127 exact_tagged_arns.update(baseline_ecr_arns.get(region_key, set()))
128 if "tagged_resources" in resources:
129 resources["tagged_resources"] = [
130 record
131 for record in resources.get("tagged_resources", [])
132 if not _tagged_resource_is_protected(
133 record,
134 protected_stack_ids=region_stack_ids,
135 protected_resource_ids=region_resource_ids,
136 exact_arns=exact_tagged_arns,
137 expected_partition=expected_partition,
138 expected_region=region_key,
139 expected_account=expected_account,
140 )
141 ]
143 protected_backup_plan_ids = region_resource_ids.get(
144 "AWS::Backup::BackupPlan",
145 set(),
146 )
147 for resource_type, category in _PROTECTED_REGIONAL_RESOURCE_CATEGORIES.items():
148 if category not in resources:
149 continue
150 physical_ids = region_resource_ids.get(resource_type, set())
151 if physical_ids:
152 resources[category] = [
153 candidate
154 for candidate in resources.get(category, [])
155 if not any(
156 _matches_protected_physical_identity(
157 resource_type,
158 category,
159 candidate,
160 physical_id,
161 protected_backup_plan_ids=protected_backup_plan_ids,
162 )
163 for physical_id in physical_ids
164 )
165 ]
167 if "ecr_repositories" in resources:
168 resources["ecr_repositories"] = [
169 name
170 for name in resources.get("ecr_repositories", [])
171 if name not in baseline_ecr_names.get(str(region), set())
172 ]
174 region_ec2_authority = (
175 authoritative_ec2_resources.get(region_key)
176 if isinstance(authoritative_ec2_resources, dict)
177 else None
178 )
179 if (
180 "tagged_resources" in resources
181 and has_exact_authority_scope
182 and has_complete_ec2_authority
183 and region_key in ec2_scanner_regions
184 and isinstance(region_ec2_authority, dict)
185 ):
186 authoritative_ec2_ids = {
187 category: {str(candidate) for candidate in candidates}
188 for category, candidates in region_ec2_authority.items()
189 if category in {item[0] for item in _EC2_TAGGED_RESOURCE_IDENTITIES.values()}
190 and isinstance(candidates, list)
191 }
192 resources["tagged_resources"] = [
193 record
194 for record in resources["tagged_resources"]
195 if (
196 (
197 identity := _ec2_tagged_resource_identity(
198 str(record.get("arn") or ""),
199 region_key,
200 expected_partition,
201 expected_account,
202 )
203 )
204 is None
205 or identity[0] not in authoritative_ec2_ids
206 or identity[1] in authoritative_ec2_ids[identity[0]]
207 )
208 ]
210 if (
211 "tagged_resources" in resources
212 and has_exact_authority_scope
213 and has_complete_eks_authority
214 and region_key in eks_scanner_regions
215 and isinstance(authoritative_clusters, dict)
216 and region in authoritative_clusters
217 and isinstance(authoritative_clusters[region], list)
218 ):
219 existing_clusters = {str(name) for name in authoritative_clusters[region]}
220 resources["tagged_resources"] = [
221 record
222 for record in resources["tagged_resources"]
223 if (
224 (
225 parent := _eks_pod_parent_cluster(
226 str(record.get("arn") or ""),
227 region_key,
228 expected_partition,
229 expected_account,
230 )
231 )
232 is None
233 or parent in existing_clusters
234 )
235 ]
237 if not any(resources.values()):
238 inventory["regional"].pop(region)
240 for resource_type, category in _PROTECTED_GLOBAL_RESOURCE_CATEGORIES.items():
241 if category not in inventory:
242 continue
243 physical_ids = {
244 physical_id
245 for resources_by_type in protected_resource_ids.values()
246 for physical_id in resources_by_type.get(resource_type, set())
247 }
248 if physical_ids:
249 inventory[category] = [
250 candidate
251 for candidate in inventory.get(category, [])
252 if not any(
253 _matches_protected_physical_identity(
254 resource_type,
255 category,
256 candidate,
257 physical_id,
258 )
259 for physical_id in physical_ids
260 )
261 ]
262 return inventory
265def _strip_expected_retained_ecr(
266 ctx: RunContext,
267 final_baseline: dict[str, Any],
268) -> tuple[dict[str, Any], dict[str, list[dict[str, Any]]]]:
269 """Remove only exact checkpointed ECR residuals from comparison inventory."""
270 sanitized = copy.deepcopy(final_baseline)
271 repositories_by_region = sanitized.setdefault("ecr_repositories", {})
272 baseline_repositories = {
273 (str(region), str(repository["name"])): repository
274 for region, repositories in (ctx.checkpoint.baseline or {})
275 .get("ecr_repositories", {})
276 .items()
277 for repository in repositories
278 }
279 accepted: dict[str, list[dict[str, Any]]] = {
280 "repositories": [],
281 "image_deltas": [],
282 }
283 created_keys: set[tuple[str, str]] = set()
285 for record in ctx.checkpoint.state.get("created_ecr_repositories", []):
286 region = str(record["region"])
287 name = str(record["name"])
288 repository_key = (region, name)
289 if repository_key in created_keys or repository_key in baseline_repositories:
290 raise RuntimeError(f"Invalid retained ECR repository authority for {region}:{name}")
291 created_keys.add(repository_key)
292 repositories = repositories_by_region.get(region, [])
293 matches = [repository for repository in repositories if repository.get("name") == name]
294 if len(matches) > 1:
295 raise RuntimeError(f"Final ECR inventory duplicated {region}:{name}")
296 if not matches:
297 accepted["repositories"].append(
298 {"region": region, "name": name, "arn": record["arn"], "already_absent": True}
299 )
300 continue
301 repository = matches[0]
302 if _ecr_creation_identity(repository) != record.get("creation_identity"):
303 raise RuntimeError(f"Retained ECR repository identity changed for {region}:{name}")
304 if (repository.get("tags") or {}).get(_RUN_STACK_TAG) != record.get("run_tag"):
305 raise RuntimeError(f"Retained ECR repository run tag changed for {record['arn']}")
306 repositories.remove(repository)
307 accepted["repositories"].append(
308 {
309 "region": region,
310 "name": name,
311 "arn": record["arn"],
312 "retained": True,
313 "inventory": repository,
314 }
315 )
317 observed_delta_keys: set[tuple[str, str, str]] = set()
318 for record in ctx.checkpoint.state.get("retained_ecr_image_deltas", []):
319 region = str(record["region"])
320 name = str(record["repository"])
321 tag = str(record["tag"])
322 image_key = (region, name, tag)
323 if image_key in observed_delta_keys or (region, name) in created_keys:
324 raise RuntimeError(f"Invalid retained ECR image-delta authority for {image_key}")
325 observed_delta_keys.add(image_key)
326 baseline_repository = baseline_repositories.get((region, name))
327 if baseline_repository is None or _image_with_tag(baseline_repository, tag) is not None:
328 raise RuntimeError(
329 f"Retained ECR image delta is not absent from the baseline: {image_key}"
330 )
331 repositories = repositories_by_region.get(region, [])
332 matches = [repository for repository in repositories if repository.get("name") == name]
333 if len(matches) != 1:
334 raise RuntimeError(
335 f"Baseline ECR repository changed before final comparison: {image_key}"
336 )
337 repository = matches[0]
338 images = repository.get("images", [])
339 tagged_images = [image for image in images if tag in image.get("tags", [])]
340 if len(tagged_images) > 1:
341 raise RuntimeError(f"Retained ECR tag resolves to multiple images: {image_key}")
342 if not tagged_images:
343 accepted["image_deltas"].append(
344 {"region": region, "repository": name, "tag": tag, "already_absent": True}
345 )
346 continue
347 image = tagged_images[0]
348 if _ecr_image_identity(image) != record.get("identity"):
349 raise RuntimeError(f"Retained ECR image identity changed for {image_key}")
350 remaining_tags = sorted(value for value in image.get("tags", []) if value != tag)
351 baseline_digest_matches = [
352 baseline_image
353 for baseline_image in baseline_repository.get("images", [])
354 if baseline_image.get("digest") == image.get("digest")
355 ]
356 if len(baseline_digest_matches) > 1:
357 raise RuntimeError(f"Baseline ECR digest is ambiguous for {image_key}")
358 if remaining_tags or baseline_digest_matches:
359 image["tags"] = remaining_tags
360 else:
361 images.remove(image)
362 accepted["image_deltas"].append(
363 {
364 "region": region,
365 "repository": name,
366 "tag": tag,
367 "digest": record["identity"]["digest"],
368 "retained": True,
369 }
370 )
372 return sanitized, accepted
375def _strip_accepted_retained_ecr(
376 project_inventory: dict[str, Any],
377 accepted: dict[str, list[dict[str, Any]]],
378) -> dict[str, Any]:
379 """Exclude exact retained repositories after final identity revalidation."""
380 allowed = {
381 (str(item["region"]), str(item["name"]))
382 for item in accepted.get("repositories", [])
383 if item.get("retained")
384 }
385 inventory = copy.deepcopy(project_inventory)
386 for region, resources in list(inventory.get("regional", {}).items()):
387 resources["ecr_repositories"] = [
388 name
389 for name in resources.get("ecr_repositories", [])
390 if (str(region), str(name)) not in allowed
391 ]
392 if not any(resources.values()):
393 inventory["regional"].pop(region)
394 return inventory
397def _merge_expected_ecr_target(
398 targets: dict[tuple[str, str, str], dict[str, Any]],
399 *,
400 region: str,
401 repository: str,
402 tag: str,
403 source: dict[str, str],
404) -> None:
405 if not region or not repository or not tag:
406 raise RuntimeError(f"Invalid expected ECR target: {region}:{repository}:{tag}")
407 key = (region, repository, tag)
408 target = targets.setdefault(
409 key,
410 {"region": region, "repository": repository, "tag": tag, "sources": []},
411 )
412 if source not in target["sources"]:
413 target["sources"].append(source)
416def _expected_ecr_images(
417 ctx: RunContext,
418 stack_names: list[str],
419 *,
420 root: Path | None = None,
421) -> list[dict[str, Any]]:
422 """Derive exact CDK-asset and configured mirror tags without AWS writes.
424 ``root`` is the checkout whose ``cdk.out``, ``cdk.json``, and chart
425 catalogue are read; it defaults to the validated checkout. The upgrade
426 harness also passes the private checkout of the release it deploys first,
427 whose mirror copies the images its own charts name.
428 """
429 checkout = ctx.settings.repo_root if root is None else root
430 targets: dict[tuple[str, str, str], dict[str, Any]] = {}
431 assembly = checkout / "cdk.out"
432 for stack_name in stack_names:
433 path = assembly / f"{stack_name}.assets.json"
434 try:
435 document = json.loads(path.read_text(encoding="utf-8"))
436 except (OSError, UnicodeError, json.JSONDecodeError) as exc:
437 raise RuntimeError(f"Could not read cloud assembly assets {path}: {exc}") from exc
438 docker_images = document.get("dockerImages") if isinstance(document, dict) else None
439 if not isinstance(docker_images, dict):
440 raise RuntimeError(f"Cloud assembly {path} omitted dockerImages")
441 for asset_id, asset in docker_images.items():
442 destinations = asset.get("destinations") if isinstance(asset, dict) else None
443 if not isinstance(destinations, dict):
444 raise RuntimeError(f"Docker asset {stack_name}:{asset_id} has no destinations")
445 for destination in destinations.values():
446 if not isinstance(destination, dict):
447 raise RuntimeError(f"Docker asset {stack_name}:{asset_id} is malformed")
448 _merge_expected_ecr_target(
449 targets,
450 region=str(destination.get("region") or ""),
451 repository=str(destination.get("repositoryName") or ""),
452 tag=str(destination.get("imageTag") or ""),
453 source={"kind": "cdk-asset", "stack": stack_name, "asset_id": str(asset_id)},
454 )
456 from cli import _image_mirror
458 mirror_config = _image_mirror.read_mirror_config(checkout / "cdk.json")
459 if mirror_config["enabled"]:
460 source_refs = _image_mirror.collect_source_refs(
461 _image_mirror.load_charts_config(checkout / _CHARTS_YAML)
462 )
463 for region in ctx.deployment_regions:
464 plan = _image_mirror.plan_from_sources(
465 source_refs,
466 f"validation.invalid.{region}",
467 mirror_config["ecr_namespace"],
468 )
469 for item in plan:
470 _merge_expected_ecr_target(
471 targets,
472 region=region,
473 repository=item.dest_repo,
474 tag=item.tag,
475 source={"kind": "configured-mirror", "source_ref": item.source_ref},
476 )
478 return [
479 {
480 **target,
481 "sources": sorted(target["sources"], key=lambda item: json.dumps(item, sort_keys=True)),
482 }
483 for _key, target in sorted(targets.items())
484 ]
487def _checkpointed_ecr_image_targets(ctx: RunContext) -> list[dict[str, Any]]:
488 """Every image target this run may publish, from the checkpoint.
490 Normally the validated checkout's ``expected_ecr_images``. A run that
491 first deployed a prior release (the upgrade harness) also checkpointed
492 that release's targets; a target both graphs share appears once, with
493 the sources of both.
494 """
495 current = ctx.checkpoint.state.get("expected_ecr_images", [])
496 prior = ctx.checkpoint.state.get(PRIOR_RELEASE_ECR_IMAGES_KEY)
497 if prior is None:
498 return list(current)
499 targets: dict[tuple[str, str, str], dict[str, Any]] = {}
500 for item in [*current, *prior]:
501 for source in item.get("sources") or [{"kind": "unrecorded"}]:
502 _merge_expected_ecr_target(
503 targets,
504 region=str(item.get("region") or ""),
505 repository=str(item.get("repository") or ""),
506 tag=str(item.get("tag") or ""),
507 source=source,
508 )
509 return [
510 {
511 **target,
512 "sources": sorted(target["sources"], key=lambda item: json.dumps(item, sort_keys=True)),
513 }
514 for _key, target in sorted(targets.items())
515 ]
518def _ecr_creation_identity(repository: dict[str, Any]) -> dict[str, Any]:
519 """Return immutable-enough ECR creation fields for delete authorization."""
520 return {
521 "name": str(repository.get("name") or ""),
522 "arn": str(repository.get("arn") or ""),
523 "registry_id": str(repository.get("registry_id") or ""),
524 "created_at": repository.get("created_at"),
525 }
528def _record_ecr_repository_creation(
529 ctx: RunContext,
530 region: str,
531 repository: Mapping[str, Any],
532) -> None:
533 """Persist the synchronous create_repository acknowledgement before any copy."""
534 name = str(repository.get("repositoryName") or "")
535 arn = str(repository.get("repositoryArn") or "")
536 registry_id = str(repository.get("registryId") or "")
537 created_at_raw = repository.get("createdAt")
538 if created_at_raw is None:
539 created_at = ""
540 elif hasattr(created_at_raw, "isoformat"):
541 created_at = str(created_at_raw.isoformat())
542 else:
543 created_at = str(created_at_raw)
544 expected = {
545 (str(item["region"]), str(item["repository"]))
546 for item in _checkpointed_ecr_image_targets(ctx)
547 }
548 baseline_names = {
549 (str(baseline_region), str(item["name"]))
550 for baseline_region, repositories in (ctx.checkpoint.baseline or {})
551 .get("ecr_repositories", {})
552 .items()
553 for item in repositories
554 }
555 key = (region, name)
556 if key not in expected or key in baseline_names:
557 raise RuntimeError(f"Unexpected ECR repository creation acknowledgement: {region}:{name}")
558 if not arn or not registry_id or not created_at:
559 raise RuntimeError(f"ECR creation acknowledgement is incomplete for {region}:{name}")
560 creation_identity = {
561 "name": name,
562 "arn": arn,
563 "registry_id": registry_id,
564 "created_at": created_at,
565 }
566 with ctx.state_lock:
567 records = ctx.checkpoint.state.setdefault("created_ecr_repositories", [])
568 matches = [
569 item for item in records if item.get("region") == region and item.get("name") == name
570 ]
571 if len(matches) > 1:
572 raise RuntimeError(f"Duplicate ECR creation acknowledgements for {region}:{name}")
573 candidate = {
574 "region": region,
575 "name": name,
576 "arn": arn,
577 "creation_identity": creation_identity,
578 "run_tag": ctx.settings.run_id,
579 "cleanup_policy": "retain-no-conditional-delete",
580 }
581 if matches and matches[0] != candidate:
582 raise RuntimeError(f"ECR creation acknowledgement changed for {region}:{name}")
583 if not matches:
584 records.append(candidate)
585 ctx.persist_callback(ctx.checkpoint)
588def _checkpoint_new_ecr_repositories(ctx: RunContext) -> list[dict[str, Any]]:
589 """Reconcile only repositories backed by persisted create acknowledgements."""
590 records = ctx.checkpoint.state.get("created_ecr_repositories", [])
591 if not records:
592 return []
593 current = collect_ecr_inventory(
594 ctx.session,
595 {str(item["region"]) for item in records},
596 )
597 current_by_key = {
598 (region, str(repository["name"])): repository
599 for region, repositories in current.items()
600 for repository in repositories
601 }
602 for record in records:
603 key = (str(record["region"]), str(record["name"]))
604 repository = current_by_key.get(key)
605 if repository is None:
606 record["observed_absent"] = True
607 continue
608 if _ecr_creation_identity(repository) != record.get("creation_identity"):
609 raise RuntimeError(f"ECR repository identity changed for {key[0]}:{key[1]}")
610 if (repository.get("tags") or {}).get(_RUN_STACK_TAG) != record.get("run_tag"):
611 raise RuntimeError(f"ECR repository run tag changed for {record['arn']}")
612 record["last_observed"] = _ecr_creation_identity(repository)
613 ctx.persist()
614 return copy.deepcopy(records)
617def _ecr_image_identity(image: dict[str, Any]) -> dict[str, Any]:
618 return {
619 "digest": str(image.get("digest") or ""),
620 "manifest_media_type": str(image.get("manifest_media_type") or ""),
621 "artifact_media_type": str(image.get("artifact_media_type") or ""),
622 "manifest": image.get("manifest"),
623 }
626def _image_with_tag(repository: dict[str, Any], tag: str) -> dict[str, Any] | None:
627 matches = [image for image in repository.get("images", []) if tag in image.get("tags", [])]
628 if len(matches) > 1:
629 raise RuntimeError(f"ECR tag {repository.get('name')}:{tag} resolved to multiple digests")
630 return matches[0] if matches else None
633def _checkpoint_new_ecr_images(ctx: RunContext) -> list[dict[str, Any]]:
634 """Observe ECR deltas without converting mutable tags into delete authority."""
635 baseline = ctx.checkpoint.baseline
636 if baseline is None:
637 raise RuntimeError("Cannot reconcile ECR images without a baseline")
638 baseline_repositories = {
639 (region, str(repository["name"])): repository
640 for region, repositories in baseline.get("ecr_repositories", {}).items()
641 for repository in repositories
642 }
643 current = collect_ecr_inventory(
644 ctx.session,
645 baseline.get("ecr_regions") or _topology_regions(ctx),
646 )
647 current_repositories = {
648 (region, str(repository["name"])): repository
649 for region, repositories in current.items()
650 for repository in repositories
651 }
653 deltas: list[dict[str, Any]] = []
654 for expected in _checkpointed_ecr_image_targets(ctx):
655 key = (
656 str(expected["region"]),
657 str(expected["repository"]),
658 str(expected["tag"]),
659 )
660 baseline_repository = baseline_repositories.get(key[:2])
661 if baseline_repository is None:
662 continue
663 current_repository = current_repositories.get(key[:2])
664 if current_repository is None:
665 raise RuntimeError(f"Baseline ECR repository disappeared: {key[0]}:{key[1]}")
666 before = _image_with_tag(baseline_repository, key[2])
667 now = _image_with_tag(current_repository, key[2])
668 if before is not None:
669 if now is None or _ecr_image_identity(now) != _ecr_image_identity(before):
670 raise RuntimeError(f"Baseline ECR tag changed during validation: {key}")
671 continue
672 if now is None:
673 continue
674 identity = _ecr_image_identity(now)
675 if not identity["digest"]:
676 raise RuntimeError(f"Expected ECR tag lacks a digest: {key}")
677 deltas.append(
678 {
679 "region": key[0],
680 "repository": key[1],
681 "tag": key[2],
682 "identity": identity,
683 "sources": expected.get("sources", []),
684 "cleanup_policy": "retain-no-conditional-delete",
685 }
686 )
687 with ctx.state_lock:
688 previous = ctx.checkpoint.state.get("retained_ecr_image_deltas")
689 if previous is not None and previous != deltas:
690 raise RuntimeError("Observed ECR image deltas changed during validation")
691 ctx.checkpoint.state["retained_ecr_image_deltas"] = deltas
692 ctx.checkpoint.state["owned_ecr_images"] = []
693 ctx.persist_callback(ctx.checkpoint)
694 return copy.deepcopy(deltas)