Coverage for scripts / live_release_validation / inventory / project.py: 100.00%
141 statements
« prev ^ index » next coverage.py v7.13.5, created at 2026-09-14 22:07 +0000
1"""Aggregate project-resource collection, baselines, and absence proofs."""
3from __future__ import annotations
5import json
6from collections.abc import Iterable
7from typing import Any
9from ._shared import (
10 _GLOBAL_PROJECT_RESOURCE_CATEGORIES,
11 _PROJECT_RESOURCE_CATEGORIES,
12 _PROJECT_RESOURCE_SCANNERS,
13 _REGIONAL_PROJECT_RESOURCE_CATEGORIES,
14 _project_owned_name,
15)
16from .ecr import (
17 collect_ecr_inventory,
18)
19from .scanners import (
20 _global_accelerator_control_region,
21 _list_api_gateway_v1_apis,
22 _list_api_gateway_v2_apis,
23 _list_cloudwatch_log_groups,
24 _list_cluster_volumes,
25 _list_dynamodb_tables,
26 _list_eks_clusters,
27 _list_global_accelerators,
28 _list_instance_inventory,
29 _list_instances,
30 _list_lambda_functions,
31 _list_load_balancers,
32 _list_project_backup_resources,
33 _list_project_ec2_networking,
34 _list_project_ecr_repositories,
35 _list_project_iam_resources,
36 _list_project_kms_keys,
37 _list_project_s3_buckets,
38 _list_project_tagged_resources,
39 _list_secrets,
40 _list_sqs_queues,
41 _list_target_groups,
42)
43from .stacks import (
44 collect_project_stacks,
45 collect_stack_inventory,
46 describe_stack_fingerprint,
47)
50def capture_baseline(
51 session: Any,
52 *,
53 enabled_regions: Iterable[str],
54 ecr_regions: Iterable[str],
55 protected_stack_names: Iterable[str],
56) -> dict[str, Any]:
57 """Capture protected CloudFormation and complete ECR baselines."""
58 protected_names = set(protected_stack_names)
59 stack_inventory = collect_stack_inventory(session, enabled_regions)
60 protected: dict[str, list[dict[str, Any]]] = {}
61 for region, stacks in stack_inventory.items():
62 fingerprints = []
63 for stack in stacks:
64 if stack["name"] not in protected_names:
65 continue
66 fingerprint = describe_stack_fingerprint(session, region, stack["stack_id"])
67 if fingerprint is None:
68 raise RuntimeError(
69 f"Protected stack disappeared while fingerprinting: {region}:{stack['name']}"
70 )
71 fingerprints.append(fingerprint)
72 if fingerprints:
73 protected[region] = sorted(
74 fingerprints,
75 key=lambda item: (item["name"], item["stack_id"]),
76 )
77 return {
78 "enabled_regions": sorted(set(enabled_regions)),
79 "ecr_regions": sorted(set(ecr_regions)),
80 "protected_stack_names": sorted(protected_names),
81 "protected_stacks": protected,
82 "ecr_repositories": collect_ecr_inventory(session, ecr_regions),
83 }
86def _tagged_ecr_surface(repositories: Any) -> Any:
87 """Return ECR repositories with untagged images dropped.
89 Copying a multi-arch image into a mirror repository leaves the per-platform
90 child manifests untagged: only the manifest list carries the tag. Those
91 children are not an addressable surface — nothing can reference them by tag,
92 and the retained-image acceptance mechanism keys on tags
93 (``retained_ecr_image_deltas`` -> ``_image_with_tag``), so an untagged child
94 can never be declared and can never be accepted.
96 Comparing them therefore made the check unsatisfiable: a run that mirrors a
97 multi-arch image into a repository that already existed in the baseline
98 always reported drift no matter how correct it was. Observed live on a run
99 whose Volcano mirror repositories held identical tags before and after while
100 their untagged child count grew from 8 to 12.
102 Dropping untagged images keeps every guarantee that is actually enforceable:
103 a repository appearing or disappearing is still a difference, and so is any
104 tag that is added, removed, or repointed to a different digest.
105 """
106 if not isinstance(repositories, list):
107 return repositories
108 comparable = []
109 for repository in repositories:
110 if not isinstance(repository, dict):
111 comparable.append(repository)
112 continue
113 images = repository.get("images")
114 if not isinstance(images, list):
115 comparable.append(repository)
116 continue
117 comparable.append({**repository, "images": [i for i in images if _image_tags(i)]})
118 return comparable
121def _image_tags(image: Any) -> list[Any]:
122 """Tags carried by one ECR image record, tolerating malformed entries."""
123 if not isinstance(image, dict):
124 return []
125 tags = image.get("tags")
126 return list(tags) if isinstance(tags, list) else []
129def compare_baseline(expected: dict[str, Any], actual: dict[str, Any]) -> list[dict[str, Any]]:
130 """Return exact protected-stack/ECR differences."""
131 differences: list[dict[str, Any]] = []
132 for category in ("protected_stacks", "ecr_repositories"):
133 before_by_region = expected.get(category) or {}
134 after_by_region = actual.get(category) or {}
135 for region in sorted(set(before_by_region) | set(after_by_region)):
136 before = before_by_region.get(region, [])
137 after = after_by_region.get(region, [])
138 if category == "ecr_repositories":
139 before = _tagged_ecr_surface(before)
140 after = _tagged_ecr_surface(after)
141 if before != after:
142 differences.append(
143 {
144 "category": category,
145 "region": region,
146 "before": before,
147 "after": after,
148 }
149 )
150 return differences
153def collect_project_resources(
154 session: Any,
155 *,
156 enabled_regions: Iterable[str],
157 expected_account: str,
158 project_name: str,
159 seed_region: str,
160 validation_run_id: str | None = None,
161) -> dict[str, Any]:
162 """Collect project resources with explicit, fail-closed scanner coverage."""
163 regions = sorted(set(enabled_regions))
164 partition = session.get_partition_for_region(seed_region)
165 if not partition:
166 raise RuntimeError(f"Could not resolve AWS partition for {seed_region}")
167 if len(expected_account) != 12 or not expected_account.isdigit():
168 raise RuntimeError("EC2 existence authority requires an exact 12-digit account ID")
170 service_names = (
171 "resourcegroupstaggingapi",
172 "eks",
173 "sqs",
174 "dynamodb",
175 "elbv2",
176 "ec2",
177 "ecr",
178 "kms",
179 "lambda",
180 "apigateway",
181 "apigatewayv2",
182 "logs",
183 "secretsmanager",
184 "backup",
185 )
186 service_regions = {
187 service: set(session.get_available_regions(service, partition_name=partition))
188 for service in service_names
189 }
190 regional: dict[str, dict[str, list[Any]]] = {
191 region: {category: [] for category in _REGIONAL_PROJECT_RESOURCE_CATEGORIES}
192 for region in regions
193 }
194 authoritative_eks_clusters: dict[str, list[str]] = {}
195 authoritative_ec2_resources: dict[str, dict[str, list[str]]] = {}
196 completed_scanners: list[str] = []
197 scanner_regions: dict[str, list[str]] = {}
199 cloudformation_stacks = collect_project_stacks(session, regions, project_name)
200 scanner_regions["cloudformation_stacks"] = regions
201 completed_scanners.append("cloudformation_stacks")
203 regional_collectors = (
204 (
205 "resource_groups_tagging_api",
206 "resourcegroupstaggingapi",
207 "tagged_resources",
208 _list_project_tagged_resources,
209 ),
210 ("eks_clusters", "eks", "eks_clusters", _list_eks_clusters),
211 ("sqs_queues", "sqs", "sqs_queues", _list_sqs_queues),
212 ("dynamodb_tables", "dynamodb", "dynamodb_tables", _list_dynamodb_tables),
213 ("load_balancers", "elbv2", "load_balancers", _list_load_balancers),
214 ("target_groups", "elbv2", "target_groups", _list_target_groups),
215 ("ec2_instances", "ec2", "instances", _list_instances),
216 (
217 "ecr_repositories",
218 "ecr",
219 "ecr_repositories",
220 _list_project_ecr_repositories,
221 ),
222 ("kms_keys", "kms", "kms_keys", _list_project_kms_keys),
223 ("lambda_functions", "lambda", "lambda_functions", _list_lambda_functions),
224 (
225 "api_gateway_v1_apis",
226 "apigateway",
227 "api_gateway_v1_apis",
228 _list_api_gateway_v1_apis,
229 ),
230 (
231 "api_gateway_v2_apis",
232 "apigatewayv2",
233 "api_gateway_v2_apis",
234 _list_api_gateway_v2_apis,
235 ),
236 (
237 "cloudwatch_log_groups",
238 "logs",
239 "cloudwatch_log_groups",
240 _list_cloudwatch_log_groups,
241 ),
242 ("secrets_manager", "secretsmanager", "secrets", _list_secrets),
243 # Last of the regional collectors: the CSI driver's volumes are the one
244 # category identified purely by a Kubernetes tag, so they are scanned
245 # independently of the stack- and project-tag matching above.
246 ("cluster_volumes", "ec2", "cluster_volumes", _list_cluster_volumes),
247 )
248 for scanner, service, category, collector in regional_collectors:
249 applicable_regions = sorted(set(regions) & service_regions[service])
250 scanner_regions[scanner] = applicable_regions
251 for region in applicable_regions:
252 if scanner == "eks_clusters":
253 cluster_names = _list_eks_clusters(session, region, None)
254 authoritative_eks_clusters[region] = cluster_names
255 regional[region][category] = [
256 name for name in cluster_names if _project_owned_name(name, project_name)
257 ]
258 elif scanner == "kms_keys":
259 regional[region][category] = _list_project_kms_keys(
260 session,
261 region,
262 project_name,
263 validation_run_id,
264 )
265 elif scanner == "ec2_instances":
266 project_instances, all_instances = _list_instance_inventory(
267 session,
268 region,
269 project_name,
270 )
271 regional[region][category] = project_instances
272 authoritative_ec2_resources.setdefault(region, {})[category] = all_instances
273 else:
274 regional[region][category] = collector(session, region, project_name)
275 completed_scanners.append(scanner)
277 if scanner == "ec2_instances":
278 scanner_regions["ec2_networking"] = applicable_regions
279 for region in applicable_regions:
280 project_networking, authoritative_networking = _list_project_ec2_networking(
281 session,
282 region,
283 project_name,
284 regional[region]["instances"],
285 )
286 regional[region].update(project_networking)
287 authoritative_ec2_resources.setdefault(region, {}).update(authoritative_networking)
288 completed_scanners.append("ec2_networking")
290 backup_regions = sorted(set(regions) & service_regions["backup"])
291 scanner_regions["aws_backup"] = backup_regions
292 for region in backup_regions:
293 regional[region].update(_list_project_backup_resources(session, region, project_name))
294 completed_scanners.append("aws_backup")
296 s3_buckets = _list_project_s3_buckets(session, seed_region, project_name)
297 scanner_regions["s3_buckets"] = ["global"]
298 completed_scanners.append("s3_buckets")
300 iam_resources = _list_project_iam_resources(session, seed_region, project_name)
301 scanner_regions["iam"] = ["global"]
302 completed_scanners.append("iam")
304 global_accelerator_region = _global_accelerator_control_region(session, seed_region)
305 global_accelerators = _list_global_accelerators(
306 session,
307 global_accelerator_region,
308 project_name,
309 )
310 scanner_regions["global_accelerators"] = (
311 [global_accelerator_region] if global_accelerator_region else []
312 )
313 completed_scanners.append("global_accelerators")
315 coverage = {
316 "complete": completed_scanners == list(_PROJECT_RESOURCE_SCANNERS),
317 "required_scanners": list(_PROJECT_RESOURCE_SCANNERS),
318 "completed_scanners": completed_scanners,
319 "scanner_regions": scanner_regions,
320 "enabled_regions": regions,
321 "resource_categories": list(_PROJECT_RESOURCE_CATEGORIES),
322 }
323 if not coverage["complete"]:
324 raise RuntimeError(
325 "Project resource inventory did not run every required scanner: "
326 + json.dumps(coverage, sort_keys=True)
327 )
329 populated_regional = {
330 region: resources for region, resources in regional.items() if any(resources.values())
331 }
332 return {
333 "coverage": coverage,
334 "authority_scope": {"partition": partition, "account": expected_account},
335 "cloudformation_stacks": cloudformation_stacks,
336 "authoritative_eks_clusters": authoritative_eks_clusters,
337 "authoritative_ec2_resources": authoritative_ec2_resources,
338 "regional": populated_regional,
339 "global_accelerators": global_accelerators,
340 "s3_buckets": s3_buckets,
341 **iam_resources,
342 }
345def summarize_project_resources(inventory: dict[str, Any]) -> dict[str, int]:
346 """Flatten every residual resource category into report-friendly counts."""
347 summary = dict.fromkeys(_PROJECT_RESOURCE_CATEGORIES, 0)
348 summary["cloudformation_stacks"] = sum(
349 len(items) for items in inventory.get("cloudformation_stacks", {}).values()
350 )
351 for resources in inventory.get("regional", {}).values():
352 for category in _REGIONAL_PROJECT_RESOURCE_CATEGORIES:
353 summary[category] += len(resources.get(category, []))
354 for category in _GLOBAL_PROJECT_RESOURCE_CATEGORIES:
355 summary[category] = len(inventory.get(category, []))
356 return summary
359def project_resources_are_absent(inventory: dict[str, Any]) -> bool:
360 """Return true only for an explicitly complete, all-zero inventory."""
361 coverage = inventory.get("coverage")
362 if not isinstance(coverage, dict) or coverage.get("complete") is not True:
363 return False
364 required = coverage.get("required_scanners")
365 completed = coverage.get("completed_scanners")
366 categories = coverage.get("resource_categories")
367 if required != list(_PROJECT_RESOURCE_SCANNERS):
368 return False
369 if completed != list(_PROJECT_RESOURCE_SCANNERS):
370 return False
371 if categories != list(_PROJECT_RESOURCE_CATEGORIES):
372 return False
373 return all(count == 0 for count in summarize_project_resources(inventory).values())