Coverage for scripts / live_release_validation / actions / preflight.py: 100.00%
100 statements
« prev ^ index » next coverage.py v7.13.5, created at 2026-09-14 22:07 +0000
1"""preflight: verify exact git, account, configuration, and ownership identity."""
3from __future__ import annotations
5import json
6import shutil
7from pathlib import Path
8from typing import Any
10from ..constants import (
11 _CLUSTER_TUNNEL_ACTIONS,
12 _HEALTHY_STACK_STATUSES,
13)
14from ..context import (
15 _direct_regional_access_enabled,
16 _resolve_branch,
17 _run_git,
18 _topology_regions,
19 _validate_profile,
20)
21from ..inventory import (
22 collect_project_stacks,
23 describe_stack,
24 discover_enabled_regions,
25)
26from ..models import RunContext, RunSettings
27from ..ownership.ecr import (
28 _expected_ecr_images,
29)
30from ..ownership.stacks import (
31 _reconcile_stack_ownership,
32)
34_GIB = float(1024**3)
37def _nearest_existing(path: Path) -> Path:
38 """Walk up until a path that exists (the report dir may not yet)."""
39 candidate = path
40 while not candidate.exists() and candidate.parent != candidate:
41 candidate = candidate.parent
42 return candidate
45def _check_free_disk(settings: RunSettings) -> dict[str, float]:
46 """Refuse to deploy from a host that cannot absorb this run's disk usage.
48 ``deploy`` builds every service image locally before publishing, the
49 checkpoint grows to tens of megabytes, and the container runtime's image
50 store lives under the home volume on macOS. A host that fills up mid-run
51 fails the image build, then fails to persist the checkpoint, and that
52 second failure aborts the guaranteed cleanup too — leaving stacks behind
53 that only a resume can reclaim. Measure the floor before anything is
54 created. Every probed location is reported so the operator sees where
55 the space went; the check fails on the first location below the floor.
56 """
57 floor_gib = float(settings.min_free_disk_gib)
58 probes = {
59 "repo_root": Path(settings.repo_root),
60 "report_dir": Path(settings.report_dir),
61 "home": Path.home(),
62 }
63 observed: dict[str, float] = {}
64 short: list[str] = []
65 for label, path in probes.items():
66 free_gib = shutil.disk_usage(_nearest_existing(path)).free / _GIB
67 observed[label] = round(free_gib, 2)
68 if free_gib < floor_gib:
69 short.append(f"{label} ({path}) has {free_gib:.1f} GiB free")
70 if short:
71 raise RuntimeError(
72 f"Free disk space is below the {floor_gib:g} GiB floor deploy needs for "
73 "container image builds and checkpoint persistence: "
74 + "; ".join(short)
75 + ". Reclaim space (stale container images from earlier runs are the usual "
76 "culprit) or lower --min-free-disk-gib, then rerun."
77 )
78 return observed
81def action_preflight(ctx: RunContext) -> dict[str, Any]:
82 """Validate exact git/AWS/config identity and prove project ownership."""
83 settings = ctx.settings
84 head = _run_git(settings.repo_root, "rev-parse", "HEAD")
85 if head != settings.expected_sha:
86 raise RuntimeError(f"HEAD {head} does not match expected SHA {settings.expected_sha}")
88 branch = _resolve_branch(settings.repo_root)
89 if branch != settings.expected_branch:
90 raise RuntimeError(
91 f"Current branch {branch!r} does not match expected branch {settings.expected_branch!r}"
92 )
94 dirty = _run_git(
95 settings.repo_root,
96 "status",
97 "--porcelain=v1",
98 "--untracked-files=all",
99 )
100 if dirty:
101 raise RuntimeError(
102 "Live validation requires a clean worktree; commit or remove these paths:\n" + dirty
103 )
105 selected = set(ctx.report.selected_actions)
106 session_manager_plugin = None
107 tunnel_actions = sorted(selected & _CLUSTER_TUNNEL_ACTIONS)
108 if tunnel_actions:
109 session_manager_plugin = shutil.which("session-manager-plugin")
110 if session_manager_plugin is None:
111 raise RuntimeError(
112 f"The {', '.join(tunnel_actions)} action(s) reach the private cluster "
113 "endpoint through an SSM tunnel and require the AWS Session Manager plugin "
114 "before deploy. Install session-manager-plugin and ensure it is on PATH, "
115 "then resume."
116 )
118 free_disk_gib: dict[str, float] | None = None
119 if "deploy" in selected and settings.min_free_disk_gib > 0:
120 free_disk_gib = _check_free_disk(settings)
122 identity = ctx.session.client("sts", region_name=ctx.config.global_region).get_caller_identity()
123 account = str(identity.get("Account") or "")
124 if account != settings.expected_account:
125 raise RuntimeError(
126 f"AWS caller account {account or 'unknown'} does not match expected "
127 f"account {settings.expected_account}"
128 )
130 _validate_profile(ctx)
131 if "deploy" in selected and not settings.confirm_kms_key_deletion:
132 raise RuntimeError(
133 "Deployment creates retained EKS encryption keys. Pass "
134 "--confirm-kms-key-deletion to explicitly authorize scheduling only "
135 "this run's exact keys for deletion during cleanup."
136 )
137 direct_regional_access = _direct_regional_access_enabled(ctx)
138 if (
139 len(ctx.deployment_regions) > 1
140 and selected.intersection({"api", "sqs", "central-queue"})
141 and not direct_regional_access
142 ):
143 raise RuntimeError(
144 "Multi-Region Job actions require api_gateway.regional_api_enabled=true; "
145 "the global API cannot prove which same-named regional Job it observed"
146 )
148 enabled_regions = discover_enabled_regions(ctx.session, ctx.config.global_region)
149 target_stacks = ctx.stack_manager.list_stacks()
150 if not target_stacks:
151 raise RuntimeError("CDK returned no target stacks")
152 expected_ecr_images = _expected_ecr_images(ctx, target_stacks)
153 unexpected_names = [
154 name
155 for name in target_stacks
156 if not (name == ctx.config.project_name or name.startswith(f"{ctx.config.project_name}-"))
157 ]
158 if unexpected_names:
159 raise RuntimeError(
160 "Refusing to own non-project CDK stacks: " + ", ".join(sorted(unexpected_names))
161 )
163 target_stack_regions = {
164 stack_name: ctx.stack_manager._get_destroy_region(stack_name)
165 for stack_name in target_stacks
166 }
167 if any(not region for region in target_stack_regions.values()):
168 raise RuntimeError(
169 "Could not resolve target stack Regions: "
170 + json.dumps(target_stack_regions, sort_keys=True)
171 )
172 target_region_set = {str(region) for region in target_stack_regions.values()}
173 unavailable_targets = sorted(target_region_set - set(enabled_regions))
174 if unavailable_targets:
175 raise RuntimeError(
176 "Target Regions are not enabled for this account: " + ", ".join(unavailable_targets)
177 )
179 bootstrap_stacks: dict[str, Any] = {}
180 for region in sorted(target_region_set):
181 bootstrap = describe_stack(ctx.session, region, "CDKToolkit")
182 if bootstrap is None or bootstrap.get("status") not in _HEALTHY_STACK_STATUSES:
183 status = bootstrap.get("status") if bootstrap else "absent"
184 raise RuntimeError(
185 f"Region {region} must already contain a healthy CDKToolkit stack; found {status}. "
186 "Live validation never auto-bootstraps or mutates the protected baseline."
187 )
188 bootstrap_stacks[region] = {
189 "stack_id": bootstrap["stack_id"],
190 "status": bootstrap["status"],
191 }
193 previous_bootstrap = ctx.checkpoint.state.get("bootstrap_stacks")
194 if previous_bootstrap is not None and previous_bootstrap != bootstrap_stacks:
195 raise RuntimeError(
196 "Checkpointed CDKToolkit ARN/status changed; refusing bootstrap adoption"
197 )
198 previous_ecr_targets = ctx.checkpoint.state.get("expected_ecr_images")
199 if previous_ecr_targets is not None and previous_ecr_targets != expected_ecr_images:
200 raise RuntimeError("Cloud-assembly ECR image targets changed since checkpoint creation")
202 existing = collect_project_stacks(
203 ctx.session,
204 enabled_regions,
205 ctx.config.project_name,
206 )
207 if not ctx.checkpoint.deployment_attempted and existing:
208 raise RuntimeError(
209 "Fresh runs refuse pre-existing project stacks because ownership is unproven: "
210 + json.dumps(existing, sort_keys=True)
211 )
213 previous_targets = ctx.checkpoint.state.get("target_stack_regions")
214 if previous_targets is not None and previous_targets != target_stack_regions:
215 raise RuntimeError(
216 "CDK target stacks changed since the checkpoint was created; refusing resume"
217 )
219 ctx.checkpoint.state.update(
220 {
221 "account_arn": str(identity.get("Arn") or ""),
222 "enabled_regions": enabled_regions,
223 "target_stack_regions": target_stack_regions,
224 "topology_regions": list(_topology_regions(ctx)),
225 "bootstrap_stacks": bootstrap_stacks,
226 "expected_ecr_images": expected_ecr_images,
227 "direct_regional_access": direct_regional_access,
228 "preexisting_project_stacks": existing
229 if not ctx.checkpoint.deployment_attempted
230 else ctx.checkpoint.state.get("preexisting_project_stacks", {}),
231 }
232 )
233 ctx.persist()
234 if ctx.checkpoint.deployment_attempted:
235 _reconcile_stack_ownership(ctx)
237 return {
238 "account": account,
239 "caller_arn": identity.get("Arn"),
240 "sha": head,
241 "branch": branch,
242 "profile": settings.profile,
243 "deployment_regions": list(ctx.deployment_regions),
244 "topology_regions": list(_topology_regions(ctx)),
245 "enabled_regions": enabled_regions,
246 "target_stack_regions": target_stack_regions,
247 "bootstrap_stacks": bootstrap_stacks,
248 "expected_ecr_images": expected_ecr_images,
249 "direct_regional_access": direct_regional_access,
250 "session_manager_plugin": session_manager_plugin or "not-required",
251 "min_free_disk_gib": settings.min_free_disk_gib,
252 "free_disk_gib": free_disk_gib if free_disk_gib is not None else "not-required",
253 "kms_key_deletion_confirmed": settings.confirm_kms_key_deletion,
254 "resume": settings.resume,
255 }