Coverage for scripts / live_release_validation / ownership / stacks.py: 100.00%
171 statements
« prev ^ index » next coverage.py v7.13.5, created at 2026-09-14 22:07 +0000
1"""Durable CloudFormation stack ownership and change-set authority."""
3from __future__ import annotations
5import copy
6import json
7from typing import Any
9from ..constants import (
10 _RUN_STACK_TAG,
11)
12from ..inventory import (
13 collect_project_stacks,
14 describe_stack,
15)
16from ..models import RunContext
19def _owned_stacks(ctx: RunContext) -> dict[str, dict[str, dict[str, Any]]]:
20 """Return region-qualified stack ownership records for this schema."""
21 owned = ctx.checkpoint.state.setdefault("owned_stacks", {})
22 if not isinstance(owned, dict):
23 raise RuntimeError("Checkpoint owned_stacks must be an object")
24 for region, records in owned.items():
25 if not isinstance(records, dict):
26 raise RuntimeError(f"Checkpoint stack ownership for {region} is malformed")
27 for stack_name, record in records.items():
28 if not isinstance(record, dict):
29 raise RuntimeError(
30 f"Checkpoint stack ownership for {region}:{stack_name} is malformed"
31 )
32 return owned
35def _owned_stack_record(
36 ctx: RunContext,
37 region: str,
38 stack_name: str,
39) -> dict[str, Any] | None:
40 return _owned_stacks(ctx).get(region, {}).get(stack_name)
43def _require_prepared_stack_authority(
44 record: dict[str, Any],
45 *,
46 region: str,
47 stack_name: str,
48) -> None:
49 if (
50 record.get("authority") != "prepared-change-set"
51 or not record.get("change_set_id")
52 or record.get("change_set_type") not in {"CREATE", "UPDATE"}
53 ):
54 raise RuntimeError(
55 f"Stack {region}:{stack_name} lacks persisted prepared-change-set authority"
56 )
59def _prepared_change_set_authority(
60 ctx: RunContext,
61) -> dict[str, dict[str, dict[str, str]]]:
62 """Return validated per-target preparation history, including legacy checkpoints."""
63 target_regions = ctx.checkpoint.state.get("target_stack_regions")
64 if not isinstance(target_regions, dict):
65 raise RuntimeError("Checkpoint target_stack_regions must be an object")
67 authority: dict[str, dict[str, dict[str, str]]] = {}
68 for stack_name, region_value in target_regions.items():
69 region = str(region_value)
70 record = _owned_stack_record(ctx, region, stack_name)
71 prepared_records: dict[str, dict[str, str]] = {}
72 if record is not None:
73 _require_prepared_stack_authority(
74 record,
75 region=region,
76 stack_name=stack_name,
77 )
78 raw_records = record.get("prepared_change_sets", {})
79 if not isinstance(raw_records, dict):
80 raise RuntimeError(
81 f"Prepared change-set history for {region}:{stack_name} is malformed"
82 )
83 for change_set_id, raw_prepared in raw_records.items():
84 if not isinstance(change_set_id, str) or not isinstance(raw_prepared, dict):
85 raise RuntimeError(
86 f"Prepared change-set history for {region}:{stack_name} is malformed"
87 )
88 prepared = {
89 "change_set_id": str(raw_prepared.get("change_set_id") or ""),
90 "stack_id": str(raw_prepared.get("stack_id") or ""),
91 "change_set_type": str(raw_prepared.get("change_set_type") or ""),
92 }
93 if (
94 prepared["change_set_id"] != change_set_id
95 or prepared["stack_id"] != str(record.get("stack_id") or "")
96 or prepared["change_set_type"] not in {"CREATE", "UPDATE"}
97 ):
98 raise RuntimeError(
99 f"Prepared change-set history for {region}:{stack_name} is inconsistent"
100 )
101 prepared_records[change_set_id] = prepared
103 # Checkpoints written before per-change-set history retained only
104 # the latest preparation. Preserve that exact authority on resume.
105 legacy_change_set_id = str(record.get("change_set_id") or "")
106 if legacy_change_set_id and legacy_change_set_id not in prepared_records:
107 prepared_records[legacy_change_set_id] = {
108 "change_set_id": legacy_change_set_id,
109 "stack_id": str(record.get("stack_id") or ""),
110 "change_set_type": str(record.get("change_set_type") or ""),
111 }
112 authority[stack_name] = prepared_records
113 return authority
116def _record_prepared_stack_identity(
117 ctx: RunContext,
118 stack_name: str,
119 region: str,
120 stack_id: str,
121 change_set_id: str,
122 change_set_type: str,
123) -> None:
124 """Persist causal change-set authority before CloudFormation execution."""
125 if not stack_id or not change_set_id or change_set_type not in {"CREATE", "UPDATE"}:
126 raise RuntimeError(f"Invalid prepared change-set identity for {region}:{stack_name}")
127 with ctx.state_lock:
128 records = _owned_stacks(ctx).setdefault(region, {})
129 previous = records.get(stack_name)
130 core = {"name": stack_name, "region": region, "stack_id": stack_id}
131 if previous is not None:
132 _require_prepared_stack_authority(
133 previous,
134 region=region,
135 stack_name=stack_name,
136 )
137 if any(previous.get(key) != value for key, value in core.items()):
138 raise RuntimeError(
139 f"Prepared stack identity changed for {region}:{stack_name}; refusing adoption"
140 )
141 previous_prepared = (previous or {}).get("prepared_change_sets", {})
142 if not isinstance(previous_prepared, dict):
143 raise RuntimeError(
144 f"Prepared change-set history for {region}:{stack_name} is malformed"
145 )
146 prepared_records = copy.deepcopy(previous_prepared)
147 if previous is not None:
148 legacy_change_set_id = str(previous.get("change_set_id") or "")
149 legacy_record = {
150 "change_set_id": legacy_change_set_id,
151 "stack_id": stack_id,
152 "change_set_type": str(previous.get("change_set_type") or ""),
153 }
154 persisted_legacy = prepared_records.get(legacy_change_set_id)
155 if persisted_legacy is not None and persisted_legacy != legacy_record:
156 raise RuntimeError(
157 f"Prepared change-set history for {region}:{stack_name} is inconsistent"
158 )
159 prepared_records[legacy_change_set_id] = legacy_record
160 prepared_record = {
161 "change_set_id": change_set_id,
162 "stack_id": stack_id,
163 "change_set_type": change_set_type,
164 }
165 existing_prepared = prepared_records.get(change_set_id)
166 if existing_prepared is not None and existing_prepared != prepared_record:
167 raise RuntimeError(
168 f"Prepared change-set identity changed for {region}:{stack_name}; refusing adoption"
169 )
170 prepared_records[change_set_id] = prepared_record
171 records[stack_name] = {
172 **(previous or {}),
173 **core,
174 "run_tag": ctx.settings.run_id,
175 "authority": "prepared-change-set",
176 "change_set_id": change_set_id,
177 "change_set_type": change_set_type,
178 "prepared_change_sets": prepared_records,
179 }
180 ctx.persist_callback(ctx.checkpoint)
183def _record_stack_identity(
184 ctx: RunContext,
185 stack_name: str,
186 region: str,
187 stack: dict[str, Any],
188) -> dict[str, Any]:
189 stack_id = str(stack.get("stack_id") or "")
190 run_tag = str((stack.get("tags") or {}).get(_RUN_STACK_TAG) or "")
191 if stack.get("name") != stack_name or not stack_id:
192 raise RuntimeError(f"CloudFormation returned an invalid identity for {region}:{stack_name}")
193 if run_tag != ctx.settings.run_id:
194 raise RuntimeError(
195 f"Stack {region}:{stack_name} is not tagged for run {ctx.settings.run_id!r}"
196 )
198 with ctx.state_lock:
199 records = _owned_stacks(ctx).get(region)
200 if records is None:
201 raise RuntimeError(
202 f"Stack {region}:{stack_name} was observed without prepared-change-set authority"
203 )
204 previous = records.get(stack_name)
205 if previous is None:
206 raise RuntimeError(
207 f"Stack {region}:{stack_name} was observed without prepared-change-set authority"
208 )
209 _require_prepared_stack_authority(
210 previous,
211 region=region,
212 stack_name=stack_name,
213 )
214 core = {
215 "name": stack_name,
216 "region": region,
217 "stack_id": stack_id,
218 "run_tag": run_tag,
219 }
220 if any(previous.get(key) != value for key, value in core.items()):
221 raise RuntimeError(
222 f"Stack identity changed for {region}:{stack_name}; refusing name-based adoption"
223 )
224 candidate = {**previous, **core}
225 records[stack_name] = candidate
226 ctx.persist_callback(ctx.checkpoint)
227 return candidate
230def _reconcile_stack_ownership(ctx: RunContext) -> dict[str, Any]:
231 """Verify every live project stack by ARN and exact run tag."""
232 target_regions = ctx.checkpoint.state.get("target_stack_regions") or {}
233 enabled_regions = ctx.checkpoint.state.get("enabled_regions") or []
234 if not target_regions or not enabled_regions:
235 raise RuntimeError("Checkpoint lacks target stack Regions or enabled Regions")
237 project_stacks = collect_project_stacks(
238 ctx.session,
239 enabled_regions,
240 ctx.config.project_name,
241 )
242 expected_targets = {
243 (str(region), str(stack_name)) for stack_name, region in target_regions.items()
244 }
245 unexpected = {
246 region: [
247 item for item in stacks if (str(region), str(item["name"])) not in expected_targets
248 ]
249 for region, stacks in project_stacks.items()
250 if any((str(region), str(item["name"])) not in expected_targets for item in stacks)
251 }
252 if unexpected:
253 raise RuntimeError(
254 "Project stacks outside the checkpoint target set were found: "
255 + json.dumps(unexpected, sort_keys=True)
256 )
258 present: dict[str, dict[str, Any]] = {}
259 for stack_name, expected_region in target_regions.items():
260 region = str(expected_region)
261 stack = describe_stack(ctx.session, region, stack_name)
262 if stack is None or stack.get("status") == "DELETE_COMPLETE":
263 continue
264 present.setdefault(region, {})[stack_name] = _record_stack_identity(
265 ctx, stack_name, region, stack
266 )
268 checkpointed = _owned_stacks(ctx)
269 for region, records in checkpointed.items():
270 for stack_name, record in records.items():
271 if target_regions.get(stack_name) != region:
272 raise RuntimeError(
273 f"Checkpoint owns unexpected stack identity {region}:{stack_name}"
274 )
275 if str(record.get("region")) != region:
276 raise RuntimeError(f"Checkpoint Region changed for stack {region}:{stack_name}")
277 return present
280def _authorize_owned_stack(
281 ctx: RunContext,
282 stack_name: str,
283 region: str,
284 stack_id: str,
285) -> None:
286 """Revalidate checkpoint ARN and run tag at a destructive boundary."""
287 record = _owned_stack_record(ctx, region, stack_name)
288 if record is None:
289 raise RuntimeError(f"No checkpointed ownership exists for {region}:{stack_name}")
290 _require_prepared_stack_authority(
291 record,
292 region=region,
293 stack_name=stack_name,
294 )
295 if str(record.get("region")) != region or str(record.get("stack_id")) != stack_id:
296 raise RuntimeError(f"Checkpoint identity changed for {region}:{stack_name}")
297 live = describe_stack(ctx.session, region, stack_id)
298 if live is None:
299 raise RuntimeError(f"Checkpointed stack disappeared before authorization: {stack_id}")
300 if live.get("name") != stack_name or live.get("stack_id") != stack_id:
301 raise RuntimeError(f"CloudFormation identity changed for {region}:{stack_name}")
302 if (live.get("tags") or {}).get(_RUN_STACK_TAG) != ctx.settings.run_id:
303 raise RuntimeError(f"Run ownership changed for {region}:{stack_name}")
306def _resolve_target_stack(
307 ctx: RunContext,
308 *,
309 region: str,
310 stack_name: str,
311 expected_stack_id: str,
312) -> dict[str, Any]:
313 """Resolve live/absent/tombstone/replacement state for one exact target."""
314 exact = describe_stack(ctx.session, region, expected_stack_id) if expected_stack_id else None
315 if exact is not None and exact.get("status") != "DELETE_COMPLETE":
316 if exact.get("name") != stack_name or exact.get("stack_id") != expected_stack_id:
317 raise RuntimeError(f"Exact stack identity changed for {region}:{stack_name}")
318 return {"state": "live", "stack": exact}
320 by_name = describe_stack(ctx.session, region, stack_name)
321 if by_name is None or by_name.get("status") == "DELETE_COMPLETE":
322 return {
323 "state": "absent",
324 "tombstone": exact if exact and exact.get("status") == "DELETE_COMPLETE" else None,
325 }
326 actual_id = str(by_name.get("stack_id") or "")
327 if expected_stack_id and actual_id != expected_stack_id:
328 return {"state": "replacement", "stack": by_name}
329 if not expected_stack_id:
330 return {"state": "uncheckpointed", "stack": by_name}
331 return {"state": "live", "stack": by_name}
334def _verify_target_stack_absence(ctx: RunContext) -> dict[str, Any]:
335 """Prove every target is absent while surfacing same-name replacements."""
336 targets = ctx.checkpoint.state.get("target_stack_regions") or {}
337 if not targets:
338 raise RuntimeError("Checkpoint lacks target stack Regions for absence verification")
339 residual: list[dict[str, Any]] = []
340 absent: list[dict[str, str]] = []
341 for stack_name, raw_region in targets.items():
342 region = str(raw_region)
343 record = _owned_stack_record(ctx, region, stack_name)
344 expected_id = str((record or {}).get("stack_id") or "")
345 resolution = _resolve_target_stack(
346 ctx,
347 region=region,
348 stack_name=stack_name,
349 expected_stack_id=expected_id,
350 )
351 if resolution["state"] == "absent":
352 absent.append({"name": stack_name, "region": region, "stack_id": expected_id})
353 continue
354 stack = resolution["stack"]
355 residual.append(
356 {
357 "name": stack_name,
358 "region": region,
359 "expected_stack_id": expected_id or None,
360 "actual_stack_id": stack.get("stack_id"),
361 "status": stack.get("status"),
362 "kind": resolution["state"],
363 }
364 )
365 return {"all_absent": not residual, "absent": absent, "residual": residual}