Coverage for scripts / live_release_validation / inventory / ecr.py: 100.00%
113 statements
« prev ^ index » next coverage.py v7.13.5, created at 2026-09-14 22:07 +0000
1"""ECR repository, image, and manifest inventory."""
3from __future__ import annotations
5import json
6from collections.abc import Iterable
7from typing import Any
9from botocore.exceptions import ClientError
11from ._shared import (
12 _ECR_MANIFEST_MEDIA_TYPES,
13 _normalize_json_text,
14 _tags_to_dict,
15)
18def _optional_ecr_configuration(
19 client: Any,
20 operation: str,
21 *,
22 repository_name: str,
23 response_key: str,
24 not_found_code: str,
25) -> Any:
26 try:
27 response = getattr(client, operation)(repositoryName=repository_name)
28 except ClientError as exc:
29 if exc.response.get("Error", {}).get("Code") == not_found_code:
30 return None
31 raise
32 return _normalize_json_text(response.get(response_key))
35def _select_manifest_record(
36 *,
37 repository_name: str,
38 digest: str,
39 detail: dict[str, Any],
40 records: Iterable[dict[str, Any]],
41) -> dict[str, Any]:
42 """Select one immutable native manifest, collapsing only exact duplicates."""
43 native_media_type = str(detail.get("imageManifestMediaType") or "")
44 if native_media_type and native_media_type not in _ECR_MANIFEST_MEDIA_TYPES:
45 raise RuntimeError(
46 f"ECR image {repository_name}@{digest} has unsupported native media type "
47 f"{native_media_type!r}"
48 )
50 response_records = list(records)
51 unique_native: dict[tuple[str, str], dict[str, Any]] = {}
52 observed_media_types: set[str] = set()
53 for record in response_records:
54 record_digest = str((record.get("imageId") or {}).get("imageDigest") or "")
55 if record_digest != digest:
56 raise RuntimeError(
57 f"ECR manifest lookup returned unexpected digest {record_digest!r} "
58 f"for {repository_name}@{digest}"
59 )
60 response_media_type = str(record.get("imageManifestMediaType") or "")
61 effective_media_type = response_media_type or native_media_type
62 if effective_media_type not in _ECR_MANIFEST_MEDIA_TYPES:
63 raise RuntimeError(
64 f"ECR manifest lookup returned unsupported media type "
65 f"{effective_media_type!r} for {repository_name}@{digest}"
66 )
67 observed_media_types.add(effective_media_type)
68 raw_manifest = record.get("imageManifest")
69 if not isinstance(raw_manifest, str) or not raw_manifest.strip():
70 raise RuntimeError(
71 f"ECR manifest lookup omitted manifest bytes for {repository_name}@{digest}"
72 )
73 if native_media_type and effective_media_type != native_media_type:
74 continue
75 unique_native.setdefault((effective_media_type, raw_manifest), record)
77 if len(unique_native) != 1:
78 raise RuntimeError(
79 f"ECR manifest lookup could not select one immutable native record for "
80 f"{repository_name}@{digest}: {len(response_records)} response record(s), "
81 f"{len(unique_native)} unique native record(s), media types "
82 f"{sorted(observed_media_types)}"
83 )
84 return next(iter(unique_native.values()))
87def _collect_repository_images(client: Any, repository_name: str) -> list[dict[str, Any]]:
88 """Collect every digest, tag set, and exact native manifest in one repository."""
89 details: dict[str, dict[str, Any]] = {}
90 for page in client.get_paginator("describe_images").paginate(repositoryName=repository_name):
91 for image in page.get("imageDetails", []):
92 digest = str(image.get("imageDigest") or "")
93 if not digest:
94 raise RuntimeError(f"ECR image in {repository_name} omitted its digest")
95 previous = details.get(digest)
96 if previous is not None and previous != image:
97 raise RuntimeError(
98 f"ECR describe-images returned conflicting details for "
99 f"{repository_name}@{digest}"
100 )
101 details[digest] = image
103 manifests: dict[str, dict[str, Any]] = {}
104 digests = sorted(details)
105 for offset in range(0, len(digests), 100):
106 requested = digests[offset : offset + 100]
107 response = client.batch_get_image(
108 repositoryName=repository_name,
109 imageIds=[{"imageDigest": digest} for digest in requested],
110 acceptedMediaTypes=list(_ECR_MANIFEST_MEDIA_TYPES),
111 )
112 failures = response.get("failures", [])
113 if failures:
114 raise RuntimeError(
115 f"ECR manifest lookup failed for {repository_name}: "
116 + json.dumps(failures, sort_keys=True)
117 )
118 grouped: dict[str, list[dict[str, Any]]] = {digest: [] for digest in requested}
119 for image in response.get("images", []):
120 digest = str((image.get("imageId") or {}).get("imageDigest") or "")
121 if digest not in grouped:
122 raise RuntimeError(
123 f"ECR manifest lookup returned unexpected digest {digest!r} "
124 f"for {repository_name}"
125 )
126 grouped[digest].append(image)
127 missing = [digest for digest, records in grouped.items() if not records]
128 if missing:
129 raise RuntimeError(
130 f"ECR manifest lookup omitted {repository_name} digests: " + ", ".join(missing)
131 )
132 for digest, records in grouped.items():
133 manifests[digest] = _select_manifest_record(
134 repository_name=repository_name,
135 digest=digest,
136 detail=details[digest],
137 records=records,
138 )
140 images: list[dict[str, Any]] = []
141 for digest, detail in details.items():
142 manifest = manifests[digest]
143 pushed_at = detail.get("imagePushedAt")
144 images.append(
145 {
146 "digest": digest,
147 "tags": sorted(str(tag) for tag in detail.get("imageTags") or []),
148 "size_bytes": int(detail.get("imageSizeInBytes") or 0),
149 "pushed_at": pushed_at.isoformat() if pushed_at is not None else None,
150 "manifest_media_type": str(
151 manifest.get("imageManifestMediaType")
152 or detail.get("imageManifestMediaType")
153 or ""
154 ),
155 "artifact_media_type": str(detail.get("artifactMediaType") or ""),
156 "manifest": _normalize_json_text(manifest.get("imageManifest", "")),
157 }
158 )
159 return sorted(images, key=lambda item: item["digest"])
162def describe_ecr_image_by_tag(
163 session: Any,
164 *,
165 region: str,
166 repository_name: str,
167 tag: str,
168) -> dict[str, Any] | None:
169 """Resolve one ECR tag to its exact digest and native manifest.
171 ``None`` means ECR authoritatively reported that the tag is absent. Any
172 ambiguous, incomplete, or failed lookup raises so callers cannot turn a
173 best-effort read into deletion authority.
174 """
175 client = session.client("ecr", region_name=region)
176 try:
177 response = client.describe_images(
178 repositoryName=repository_name,
179 imageIds=[{"imageTag": tag}],
180 )
181 except ClientError as exc:
182 if exc.response.get("Error", {}).get("Code") in {
183 "ImageNotFoundException",
184 "RepositoryNotFoundException",
185 }:
186 return None
187 raise
189 details = response.get("imageDetails", [])
190 if not details:
191 raise RuntimeError(
192 f"ECR tag lookup omitted an authoritative identity for {region}:{repository_name}:{tag}"
193 )
194 if len(details) != 1:
195 raise RuntimeError(
196 f"ECR tag {region}:{repository_name}:{tag} resolved to {len(details)} images"
197 )
198 detail = details[0]
199 digest = str(detail.get("imageDigest") or "")
200 tags = sorted(str(image_tag) for image_tag in detail.get("imageTags") or [])
201 if not digest or tag not in tags:
202 raise RuntimeError(
203 f"ECR tag lookup returned an invalid identity for {region}:{repository_name}:{tag}"
204 )
206 manifest_response = client.batch_get_image(
207 repositoryName=repository_name,
208 imageIds=[{"imageDigest": digest}],
209 acceptedMediaTypes=list(_ECR_MANIFEST_MEDIA_TYPES),
210 )
211 failures = manifest_response.get("failures", [])
212 if failures:
213 raise RuntimeError(
214 f"ECR manifest lookup failed for {repository_name}:{tag}: "
215 + json.dumps(failures, sort_keys=True)
216 )
217 manifest = _select_manifest_record(
218 repository_name=repository_name,
219 digest=digest,
220 detail=detail,
221 records=manifest_response.get("images", []),
222 )
223 pushed_at = detail.get("imagePushedAt")
224 return {
225 "digest": digest,
226 "tags": tags,
227 "size_bytes": int(detail.get("imageSizeInBytes") or 0),
228 "pushed_at": pushed_at.isoformat() if pushed_at is not None else None,
229 "manifest_media_type": str(
230 manifest.get("imageManifestMediaType") or detail.get("imageManifestMediaType") or ""
231 ),
232 "artifact_media_type": str(detail.get("artifactMediaType") or ""),
233 "manifest": _normalize_json_text(manifest.get("imageManifest", "")),
234 }
237def collect_ecr_inventory(session: Any, regions: Iterable[str]) -> dict[str, list[dict[str, Any]]]:
238 """Collect exact ECR repository identity/configuration in each topology Region."""
239 inventory: dict[str, list[dict[str, Any]]] = {}
240 for region in sorted(set(regions)):
241 client = session.client("ecr", region_name=region)
242 repositories: list[dict[str, Any]] = []
243 for page in client.get_paginator("describe_repositories").paginate():
244 for repository in page.get("repositories", []):
245 name = str(repository.get("repositoryName") or "")
246 arn = str(repository.get("repositoryArn") or "")
247 tags = _tags_to_dict(client.list_tags_for_resource(resourceArn=arn).get("tags", []))
248 repositories.append(
249 {
250 "name": name,
251 "arn": arn,
252 "registry_id": str(repository.get("registryId") or ""),
253 "uri": str(repository.get("repositoryUri") or ""),
254 "created_at": (
255 repository["createdAt"].isoformat()
256 if repository.get("createdAt") is not None
257 else None
258 ),
259 "tag_mutability": str(repository.get("imageTagMutability") or ""),
260 "tag_mutability_exclusions": sorted(
261 repository.get("imageTagMutabilityExclusionFilters") or [],
262 key=lambda item: json.dumps(item, sort_keys=True),
263 ),
264 "scan_configuration": dict(
265 repository.get("imageScanningConfiguration") or {}
266 ),
267 "encryption": dict(repository.get("encryptionConfiguration") or {}),
268 "lifecycle_policy": _optional_ecr_configuration(
269 client,
270 "get_lifecycle_policy",
271 repository_name=name,
272 response_key="lifecyclePolicyText",
273 not_found_code="LifecyclePolicyNotFoundException",
274 ),
275 "repository_policy": _optional_ecr_configuration(
276 client,
277 "get_repository_policy",
278 repository_name=name,
279 response_key="policyText",
280 not_found_code="RepositoryPolicyNotFoundException",
281 ),
282 "tags": tags,
283 "images": _collect_repository_images(client, name),
284 }
285 )
286 inventory[region] = sorted(repositories, key=lambda item: item["name"])
287 return inventory