Coverage for scripts / live_release_validation / ownership / kms.py: 100.00%
178 statements
« prev ^ index » next coverage.py v7.13.5, created at 2026-10-28 16:19 +0000
1"""KMS key identity, retained-key checkpointing, and pending-deletion accounting."""
3from __future__ import annotations
5import copy
6from collections.abc import Mapping
7from typing import Any
9from botocore.exceptions import ClientError
11from ..constants import (
12 _EKS_KEY_LOGICAL_ID,
13 _HEALTHY_STACK_STATUSES,
14 _RUN_STACK_TAG,
15)
16from ..inventory import (
17 describe_stack,
18)
19from ..models import RunContext
20from ..ownership.log_groups import (
21 _checkpoint_owned_log_groups,
22)
23from ..ownership.stacks import (
24 _owned_stack_ids,
25 _owned_stack_record,
26 _owned_stacks,
27)
30def _kms_tags(client: Any, key_id: str) -> dict[str, str]:
31 tags: dict[str, str] = {}
32 marker: str | None = None
33 while True:
34 kwargs = {"KeyId": key_id}
35 if marker:
36 kwargs["Marker"] = marker
37 response = client.list_resource_tags(**kwargs)
38 tags.update(
39 {
40 str(tag["TagKey"]): str(tag.get("TagValue") or "")
41 for tag in response.get("Tags", [])
42 if tag.get("TagKey") is not None
43 }
44 )
45 marker = response.get("NextMarker") if response.get("Truncated") else None
46 if not marker:
47 return tags
50def _validated_owned_kms_identity(
51 ctx: RunContext,
52 record: Mapping[str, Any],
53) -> tuple[str, str, str, str]:
54 """Validate immutable stack-resource authority for one run-owned KMS key."""
55 region = str(record.get("region") or "")
56 key_id = str(record.get("key_id") or "")
57 arn = str(record.get("arn") or "")
58 stack_name = str(record.get("stack_name") or "")
59 stack_id = str(record.get("stack_id") or "")
60 logical_id = str(record.get("logical_id") or "")
61 target_regions = ctx.checkpoint.state.get("target_stack_regions")
62 if not isinstance(target_regions, dict) or str(target_regions.get(stack_name) or "") != region:
63 raise RuntimeError(f"KMS checkpoint target stack is invalid for {arn or key_id}")
64 partition = ctx.session.get_partition_for_region(region)
65 if not partition:
66 raise RuntimeError(f"Could not resolve AWS partition for KMS key in {region}")
67 expected_arn = f"arn:{partition}:kms:{region}:{ctx.settings.expected_account}:key/{key_id}"
68 expected_stack_prefix = (
69 f"arn:{partition}:cloudformation:{region}:{ctx.settings.expected_account}:"
70 f"stack/{stack_name}/"
71 )
72 owned_stack_record = _owned_stack_record(ctx, region, stack_name)
73 # The key's stack may be the current generation or, after an adopted
74 # in-place upgrade, one it replaced (the retained EKS key outlives it).
75 owned_stack_ids = _owned_stack_ids(ctx, region, stack_name)
76 if not key_id or arn != expected_arn:
77 raise RuntimeError(f"KMS checkpoint ARN is invalid for {arn or key_id}")
78 if (
79 not stack_name
80 or not stack_id.startswith(expected_stack_prefix)
81 or stack_id not in owned_stack_ids
82 or (owned_stack_record or {}).get("run_tag") != ctx.settings.run_id
83 ):
84 raise RuntimeError(f"KMS checkpoint stack identity is invalid for {arn}")
85 if (
86 record.get("ownership_authority") != "cloudformation-stack-resource"
87 or not logical_id
88 or record.get("run_tag") != ctx.settings.run_id
89 ):
90 raise RuntimeError(f"KMS checkpoint authority is incomplete for {arn}")
92 retained_identity = (
93 stack_name == f"{ctx.config.project_name}-{region}" and logical_id == _EKS_KEY_LOGICAL_ID
94 )
95 cleanup_policy = str(record.get("cleanup_policy") or "")
96 if not cleanup_policy and retained_identity:
97 cleanup_policy = "harness-schedule"
98 if cleanup_policy == "harness-schedule":
99 if not retained_identity:
100 raise RuntimeError(f"Retained KMS checkpoint identity is invalid for {arn}")
101 elif cleanup_policy == "cloudformation-delete":
102 if retained_identity:
103 raise RuntimeError(f"Retained EKS key cannot use CloudFormation cleanup: {arn}")
104 else:
105 raise RuntimeError(f"KMS checkpoint cleanup policy is invalid for {arn}")
106 return region, key_id, arn, cleanup_policy
109def _validated_retained_kms_identity(
110 ctx: RunContext,
111 record: Mapping[str, Any],
112) -> tuple[str, str, str]:
113 """Validate exact retained-EKS authority before harness-scheduled deletion."""
114 region, key_id, arn, cleanup_policy = _validated_owned_kms_identity(ctx, record)
115 if cleanup_policy != "harness-schedule":
116 raise RuntimeError(f"KMS key is not harness-retained: {arn}")
117 return region, key_id, arn
120def _checkpoint_retained_kms_keys(ctx: RunContext) -> list[dict[str, Any]]:
121 """Capture every exact stack-owned KMS key plus teardown log-group candidates."""
122 _checkpoint_owned_log_groups(ctx)
123 owned_stacks = _owned_stacks(ctx)
124 target_regions = ctx.checkpoint.state.get("target_stack_regions")
125 if not isinstance(target_regions, dict):
126 raise RuntimeError("Checkpoint target_stack_regions must be an object")
127 with ctx.state_lock:
128 records = ctx.checkpoint.state.setdefault("owned_kms_keys", [])
129 if not isinstance(records, list):
130 raise RuntimeError("Checkpoint owned_kms_keys must be a list")
131 by_arn = {str(item.get("arn") or ""): item for item in records if isinstance(item, dict)}
133 for stack_name, raw_region in sorted(target_regions.items()):
134 region = str(raw_region)
135 stack_record = owned_stacks.get(region, {}).get(str(stack_name))
136 if stack_record is None:
137 continue
138 live_stack = describe_stack(ctx.session, region, stack_record["stack_id"])
139 live_source_authority = (
140 live_stack is not None
141 and not str(live_stack.get("status") or "").startswith("DELETE")
142 and live_stack.get("stack_id") == stack_record["stack_id"]
143 and (live_stack.get("tags") or {}).get(_RUN_STACK_TAG) == ctx.settings.run_id
144 )
145 cfn = ctx.session.client("cloudformation", region_name=region)
146 try:
147 pages = cfn.get_paginator("list_stack_resources").paginate(
148 StackName=stack_record["stack_id"]
149 )
150 matching_resources = [
151 item
152 for page in pages
153 for item in page.get("StackResourceSummaries", [])
154 if item.get("ResourceType") == "AWS::KMS::Key"
155 and item.get("LogicalResourceId")
156 and item.get("PhysicalResourceId")
157 ]
158 except ClientError as exc:
159 if (
160 exc.response.get("Error", {}).get("Code") == "ValidationError"
161 and live_stack is None
162 ):
163 continue
164 raise
165 retained_resources = [
166 item
167 for item in matching_resources
168 if str(stack_name) == f"{ctx.config.project_name}-{region}"
169 and str(item.get("LogicalResourceId") or "") == _EKS_KEY_LOGICAL_ID
170 ]
171 if (
172 live_stack is not None
173 and live_stack.get("status") in _HEALTHY_STACK_STATUSES
174 and str(stack_name) == f"{ctx.config.project_name}-{region}"
175 and len(retained_resources) != 1
176 ):
177 raise RuntimeError(
178 f"Expected one retained EKS KMS key in {stack_name}; found "
179 f"{len(retained_resources)}"
180 )
182 for resource in matching_resources:
183 key_id = str(resource["PhysicalResourceId"])
184 logical_id = str(resource["LogicalResourceId"])
185 partition = ctx.session.get_partition_for_region(region)
186 if not partition:
187 raise RuntimeError(f"Could not resolve AWS partition for KMS key in {region}")
188 derived_arn = (
189 f"arn:{partition}:kms:{region}:{ctx.settings.expected_account}:key/{key_id}"
190 )
191 previous = by_arn.get(derived_arn)
192 if previous is None and not live_source_authority:
193 # Deleted-stack tombstones may reconcile exact records that
194 # were persisted pre-destroy, but can never create authority.
195 continue
196 kms = ctx.session.client("kms", region_name=region)
197 try:
198 metadata = kms.describe_key(KeyId=key_id).get("KeyMetadata", {})
199 except ClientError as exc:
200 if exc.response.get("Error", {}).get("Code") != "NotFoundException":
201 raise
202 if previous is not None:
203 previous["scheduled"] = True
204 previous["deleted"] = True
205 continue
206 arn = str(metadata.get("Arn") or "")
207 tags = _kms_tags(kms, key_id)
208 if tags.get(_RUN_STACK_TAG) != ctx.settings.run_id:
209 raise RuntimeError(
210 f"KMS key {arn or key_id} lacks the exact live-validation run tag"
211 )
212 cleanup_policy = (
213 "harness-schedule"
214 if str(stack_name) == f"{ctx.config.project_name}-{region}"
215 and logical_id == _EKS_KEY_LOGICAL_ID
216 else "cloudformation-delete"
217 )
218 deletion_date = metadata.get("DeletionDate")
219 state = str(metadata.get("KeyState") or "")
220 candidate = {
221 "region": region,
222 "key_id": key_id,
223 "arn": arn,
224 "stack_name": str(stack_name),
225 "stack_id": stack_record["stack_id"],
226 "logical_id": logical_id,
227 "ownership_authority": "cloudformation-stack-resource",
228 "cleanup_policy": cleanup_policy,
229 "run_tag": ctx.settings.run_id,
230 "scheduled": state == "PendingDeletion",
231 "deletion_date": (
232 deletion_date.isoformat() if deletion_date is not None else None
233 ),
234 }
235 _validated_owned_kms_identity(ctx, candidate)
236 if arn != derived_arn:
237 raise RuntimeError(f"KMS returned an unexpected ARN for {key_id}: {arn}")
238 previous = by_arn.get(arn)
239 if previous is not None:
240 previous.setdefault("cleanup_policy", cleanup_policy)
241 for key in (
242 "region",
243 "key_id",
244 "arn",
245 "stack_name",
246 "stack_id",
247 "logical_id",
248 "ownership_authority",
249 "cleanup_policy",
250 "run_tag",
251 ):
252 if previous.get(key) != candidate[key]:
253 raise RuntimeError(f"KMS ownership changed for {arn}: {key}")
254 if candidate["scheduled"]:
255 previous["scheduled"] = True
256 previous["deletion_date"] = candidate["deletion_date"]
257 continue
258 refreshed_stack = describe_stack(ctx.session, region, stack_record["stack_id"])
259 if not (
260 refreshed_stack is not None
261 and not str(refreshed_stack.get("status") or "").startswith("DELETE")
262 and refreshed_stack.get("stack_id") == stack_record["stack_id"]
263 and (refreshed_stack.get("tags") or {}).get(_RUN_STACK_TAG)
264 == ctx.settings.run_id
265 ):
266 continue
267 records.append(candidate)
268 by_arn[arn] = candidate
269 ctx.persist_callback(ctx.checkpoint)
270 ctx.persist_callback(ctx.checkpoint)
271 return copy.deepcopy(records)
274def _strip_expected_pending_kms(
275 ctx: RunContext,
276 project_inventory: dict[str, Any],
277) -> tuple[dict[str, Any], list[dict[str, Any]]]:
278 inventory = copy.deepcopy(project_inventory)
279 expected: dict[tuple[str, str], dict[str, Any]] = {}
280 accepted: list[dict[str, Any]] = []
281 for record in ctx.checkpoint.state.get("owned_kms_keys", []):
282 region, key_id, arn, cleanup_policy = _validated_owned_kms_identity(ctx, record)
283 identity = (region, arn)
284 if not record.get("scheduled"):
285 raise RuntimeError(f"Owned KMS key was not scheduled for deletion: {arn}")
286 if identity in expected:
287 raise RuntimeError(f"Duplicate KMS checkpoint identity: {region}:{arn}")
289 kms = ctx.session.client("kms", region_name=region)
290 try:
291 metadata = kms.describe_key(KeyId=key_id).get("KeyMetadata", {})
292 except ClientError as exc:
293 if exc.response.get("Error", {}).get("Code") != "NotFoundException":
294 raise
295 evidence = {
296 "region": region,
297 "key_id": key_id,
298 "arn": arn,
299 "state": "Deleted",
300 "already_absent": True,
301 "stack_id": record["stack_id"],
302 "logical_id": record["logical_id"],
303 "ownership_authority": record["ownership_authority"],
304 "cleanup_policy": cleanup_policy,
305 "run_tag": record["run_tag"],
306 }
307 else:
308 if metadata.get("Arn") != arn:
309 raise RuntimeError(f"KMS key ARN changed for {key_id}")
310 state = str(metadata.get("KeyState") or "")
311 if state != "PendingDeletion":
312 raise RuntimeError(
313 f"Expected {cleanup_policy} KMS key {arn} to be PendingDeletion; found {state}"
314 )
315 tags = _kms_tags(kms, key_id)
316 if tags.get(_RUN_STACK_TAG) != record["run_tag"]:
317 raise RuntimeError(f"KMS run ownership changed for {arn}")
318 deletion_date = metadata.get("DeletionDate")
319 observed_deletion_date = (
320 deletion_date.isoformat() if deletion_date is not None else None
321 )
322 if not observed_deletion_date or observed_deletion_date != record.get("deletion_date"):
323 raise RuntimeError(f"KMS deletion date changed for {arn}")
324 evidence = {
325 "region": region,
326 "key_id": key_id,
327 "arn": arn,
328 "state": state,
329 "description": str(metadata.get("Description") or ""),
330 "deletion_date": observed_deletion_date,
331 "tags": tags,
332 "stack_id": record["stack_id"],
333 "logical_id": record["logical_id"],
334 "ownership_authority": record["ownership_authority"],
335 "cleanup_policy": cleanup_policy,
336 "run_tag": record["run_tag"],
337 }
338 expected[identity] = evidence
339 accepted.append(evidence)
341 for region, resources in list(inventory.get("regional", {}).items()):
342 resources["kms_keys"] = [
343 key
344 for key in resources.get("kms_keys", [])
345 if (region, str(key.get("arn") or "")) not in expected
346 ]
347 if not any(resources.values()):
348 inventory["regional"].pop(region)
349 return inventory, accepted