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