Coverage for scripts / live_release_validation / ownership / cleanup_role.py: 100.00%

1"""The delegated log-cleanup helper stack and its scoped IAM role.""" 

2 

3from __future__ import annotations 

4 

5import hashlib 

6import json 

7import re 

8import time 

9import uuid 

10from collections.abc import Mapping 

11from typing import Any 

12 

13from botocore.exceptions import ClientError 

14 

15from ..constants import ( 

16 _LOG_CLEANUP_HELPER_NAMESPACE, 

17 _LOG_CLEANUP_HELPER_RUN_TAG, 

18 _LOG_CLEANUP_HELPER_STACK_PREFIX, 

19 _LOG_CLEANUP_HELPER_TOKEN_TAG, 

20 _LOG_CLEANUP_ROLE_OUTPUT, 

21 _LOG_CLEANUP_ROLE_POLICY_NAME, 

22 _LOG_CLEANUP_ROLE_RUN_TAG, 

23 _LOG_CLEANUP_ROLE_TOKEN_TAG, 

24 _LOG_CLEANUP_SESSION_SECONDS, 

25 _LOG_CLEANUP_STACK_POLL_ATTEMPTS, 

26 _LOG_CLEANUP_STACK_POLL_SECONDS, 

27 _LOG_CLEANUP_TOKEN_TAG, 

28 _RUN_STACK_TAG, 

29) 

30from ..inventory import ( 

31 describe_stack, 

32) 

33from ..models import RunContext, utc_now 

34from ..ownership.log_groups import ( 

35 _validated_owned_log_group_identity, 

36) 

37 

38 

39def _canonical_json(value: Any) -> str: 

40 return json.dumps(value, separators=(",", ":"), sort_keys=True) 

41 

42 

43def _cleanup_principal_identity(ctx: RunContext, caller_arn: str) -> dict[str, str]: 

44 """Resolve a renewable caller session to one immutable IAM principal.""" 

45 region = ctx.config.global_region 

46 partition = ctx.session.get_partition_for_region(region) 

47 account = ctx.settings.expected_account 

48 if not partition: 

49 raise RuntimeError(f"Could not resolve AWS partition for cleanup authority in {region}") 

50 if not caller_arn or "*" in caller_arn: 

51 raise RuntimeError("Cleanup authority principal ARN is empty or contains a wildcard") 

52 

53 iam = ctx.session.client("iam", region_name=region) 

54 iam_prefix = f"arn:{partition}:iam::{account}:" 

55 if caller_arn.startswith(f"{iam_prefix}user/"): 

56 user_name = caller_arn.rsplit("/", 1)[-1] 

57 user = iam.get_user(UserName=user_name).get("User") 

58 principal_arn = str((user or {}).get("Arn") or "") 

59 principal_id = str((user or {}).get("UserId") or "") 

60 if principal_arn != caller_arn or not principal_id: 

61 raise RuntimeError(f"IAM returned an invalid user identity for {caller_arn}") 

62 return {"arn": principal_arn, "principal_id": principal_id} 

63 

64 if caller_arn.startswith(f"{iam_prefix}role/"): 

65 role_name = caller_arn.rsplit("/", 1)[-1] 

66 role = iam.get_role(RoleName=role_name).get("Role") 

67 principal_arn = str((role or {}).get("Arn") or "") 

68 principal_id = str((role or {}).get("RoleId") or "") 

69 if principal_arn != caller_arn or not principal_id: 

70 raise RuntimeError(f"IAM returned an invalid role identity for {caller_arn}") 

71 return {"arn": principal_arn, "principal_id": principal_id} 

72 

73 assumed_prefix = f"arn:{partition}:sts::{account}:assumed-role/" 

74 if caller_arn.startswith(assumed_prefix): 

75 role_session = caller_arn.removeprefix(assumed_prefix) 

76 role_resource, separator, session_name = role_session.rpartition("/") 

77 role_name = role_resource.rsplit("/", 1)[-1] 

78 if not separator or not role_name or not session_name: 

79 raise RuntimeError(f"Malformed assumed-role caller ARN: {caller_arn}") 

80 role = iam.get_role(RoleName=role_name).get("Role") 

81 principal_arn = str((role or {}).get("Arn") or "") 

82 principal_id = str((role or {}).get("RoleId") or "") 

83 if ( 

84 not principal_arn.startswith(f"{iam_prefix}role/") 

85 or principal_arn.rsplit("/", 1)[-1] != role_name 

86 or not principal_id 

87 ): 

88 raise RuntimeError(f"IAM returned an invalid underlying role identity for {caller_arn}") 

89 return {"arn": principal_arn, "principal_id": principal_id} 

90 raise RuntimeError( 

91 f"Log cleanup requires an exact IAM user or STS assumed-role caller; found {caller_arn}" 

92 ) 

93 

94 

95def _log_cleanup_policy( 

96 ctx: RunContext, 

97 cleanup_token: str, 

98 records: list[dict[str, Any]], 

99) -> tuple[dict[str, Any], str]: 

100 partitions: set[str] = set() 

101 for record in records: 

102 region, _name = _validated_owned_log_group_identity(ctx, record) 

103 partition = ctx.session.get_partition_for_region(region) 

104 if not partition: 

105 raise RuntimeError(f"Could not resolve AWS partition for log cleanup in {region}") 

106 partitions.add(partition) 

107 if len(partitions) != 1: 

108 raise RuntimeError("Log cleanup requires all authorized groups to share one AWS partition") 

109 partition = next(iter(partitions)) 

110 return ( 

111 { 

112 "Version": "2012-10-17", 

113 "Statement": [ 

114 { 

115 "Effect": "Allow", 

116 "Action": "logs:DeleteLogGroup", 

117 "Resource": ( 

118 f"arn:{partition}:logs:*:{ctx.settings.expected_account}:log-group:*" 

119 ), 

120 "Condition": { 

121 "StringEquals": { 

122 f"aws:ResourceTag/{_RUN_STACK_TAG}": ctx.settings.run_id, 

123 f"aws:ResourceTag/{_LOG_CLEANUP_TOKEN_TAG}": cleanup_token, 

124 } 

125 }, 

126 } 

127 ], 

128 }, 

129 partition, 

130 ) 

131 

132 

133def _log_cleanup_helper_spec(ctx: RunContext) -> dict[str, Any] | None: 

134 records = ctx.checkpoint.state.get("owned_log_groups", []) 

135 if not isinstance(records, list): 

136 raise RuntimeError("Checkpoint owned_log_groups must be a list") 

137 if not records: 

138 return None 

139 cleanup_token = str(ctx.checkpoint.state.get("log_group_cleanup_token") or "") 

140 if not re.fullmatch(r"[0-9a-f]{32}", cleanup_token): 

141 raise RuntimeError("Checkpoint log-group cleanup token is malformed") 

142 if any(not isinstance(record, dict) for record in records): 

143 raise RuntimeError("Checkpoint owned_log_groups must contain objects") 

144 policy, partition = _log_cleanup_policy(ctx, cleanup_token, records) 

145 helper_region = str(ctx.config.global_region) 

146 if ctx.session.get_partition_for_region(helper_region) != partition: 

147 raise RuntimeError("Cleanup helper Region is outside the log groups' AWS partition") 

148 

149 existing_helper = ctx.checkpoint.state.get("log_cleanup_helper") 

150 if existing_helper is not None and not isinstance(existing_helper, dict): 

151 raise RuntimeError("Checkpoint log_cleanup_helper must be an object") 

152 if isinstance(existing_helper, dict): 

153 first_caller_arn = str(existing_helper.get("first_caller_arn") or "") 

154 trusted_principal_arn = str(existing_helper.get("trusted_principal_arn") or "") 

155 trusted_principal_id = str(existing_helper.get("trusted_principal_id") or "") 

156 if not first_caller_arn or not trusted_principal_arn or not trusted_principal_id: 

157 raise RuntimeError("Checkpoint cleanup helper lacks immutable caller identity") 

158 else: 

159 first_caller_arn = str(ctx.checkpoint.state.get("account_arn") or "") 

160 principal_identity = _cleanup_principal_identity(ctx, first_caller_arn) 

161 trusted_principal_arn = principal_identity["arn"] 

162 trusted_principal_id = principal_identity["principal_id"] 

163 expected_iam_prefix = f"arn:{partition}:iam::{ctx.settings.expected_account}:" 

164 if ( 

165 not trusted_principal_arn.startswith( 

166 (f"{expected_iam_prefix}user/", f"{expected_iam_prefix}role/") 

167 ) 

168 or "*" in trusted_principal_arn 

169 or not re.fullmatch(r"[A-Z0-9]+", trusted_principal_id) 

170 ): 

171 raise RuntimeError("Checkpoint cleanup helper canonical principal is invalid") 

172 stable_id = uuid.uuid5( 

173 _LOG_CLEANUP_HELPER_NAMESPACE, 

174 f"{partition}:{ctx.settings.expected_account}:{ctx.settings.run_id}:{cleanup_token}", 

175 ).hex[:20] 

176 stack_name = f"{_LOG_CLEANUP_HELPER_STACK_PREFIX}-{stable_id}" 

177 role_name = stack_name 

178 project_name = str(ctx.config.project_name) 

179 if any( 

180 name == project_name or name.startswith((f"{project_name}-", f"{project_name}/")) 

181 for name in (stack_name, role_name) 

182 ): 

183 raise RuntimeError("Cleanup helper identity overlaps project inventory naming") 

184 role_arn = f"arn:{partition}:iam::{ctx.settings.expected_account}:role/{role_name}" 

185 trust_policy = { 

186 "Version": "2012-10-17", 

187 "Statement": [ 

188 { 

189 "Effect": "Allow", 

190 "Principal": {"AWS": trusted_principal_arn}, 

191 "Action": "sts:AssumeRole", 

192 "Condition": {"StringEquals": {"sts:ExternalId": cleanup_token}}, 

193 } 

194 ], 

195 } 

196 template = { 

197 "AWSTemplateFormatVersion": "2010-09-09", 

198 "Description": "Temporary least-privilege role for live-validation log cleanup", 

199 "Resources": { 

200 "CleanupRole": { 

201 "Type": "AWS::IAM::Role", 

202 "Properties": { 

203 "RoleName": role_name, 

204 "MaxSessionDuration": 3600, 

205 "AssumeRolePolicyDocument": trust_policy, 

206 "Policies": [ 

207 { 

208 "PolicyName": _LOG_CLEANUP_ROLE_POLICY_NAME, 

209 "PolicyDocument": policy, 

210 } 

211 ], 

212 "Tags": [ 

213 {"Key": _LOG_CLEANUP_ROLE_RUN_TAG, "Value": ctx.settings.run_id}, 

214 {"Key": _LOG_CLEANUP_ROLE_TOKEN_TAG, "Value": cleanup_token}, 

215 ], 

216 }, 

217 } 

218 }, 

219 "Outputs": {_LOG_CLEANUP_ROLE_OUTPUT: {"Value": {"Fn::GetAtt": ["CleanupRole", "Arn"]}}}, 

220 } 

221 template_body = _canonical_json(template) 

222 return { 

223 "schema_version": 1, 

224 "region": helper_region, 

225 "stack_name": stack_name, 

226 "role_name": role_name, 

227 "role_arn": role_arn, 

228 "partition": partition, 

229 "run_id": ctx.settings.run_id, 

230 "cleanup_token": cleanup_token, 

231 "first_caller_arn": first_caller_arn, 

232 "trusted_principal_arn": trusted_principal_arn, 

233 "trusted_principal_id": trusted_principal_id, 

234 "role_policy": policy, 

235 "trust_policy": trust_policy, 

236 "template": template, 

237 "template_body": template_body, 

238 "template_sha256": hashlib.sha256(template_body.encode("utf-8")).hexdigest(), 

239 } 

240 

241 

242def _prepare_log_cleanup_helper_record( 

243 ctx: RunContext, 

244 spec: Mapping[str, Any], 

245) -> dict[str, Any]: 

246 immutable_keys = ( 

247 "schema_version", 

248 "region", 

249 "stack_name", 

250 "role_name", 

251 "role_arn", 

252 "partition", 

253 "run_id", 

254 "cleanup_token", 

255 "trusted_principal_arn", 

256 "trusted_principal_id", 

257 "template_sha256", 

258 ) 

259 with ctx.state_lock: 

260 record = ctx.checkpoint.state.get("log_cleanup_helper") 

261 if record is None: 

262 record = {key: spec[key] for key in immutable_keys} 

263 record.update( 

264 { 

265 "first_caller_arn": spec["first_caller_arn"], 

266 "active_stack_id": None, 

267 "lifecycle": "prepared", 

268 "create_sequence": 0, 

269 "stack_history": [], 

270 } 

271 ) 

272 ctx.checkpoint.state["log_cleanup_helper"] = record 

273 elif not isinstance(record, dict): 

274 raise RuntimeError("Checkpoint log_cleanup_helper must be an object") 

275 elif any(record.get(key) != spec[key] for key in immutable_keys): 

276 raise RuntimeError("Checkpoint log cleanup helper identity changed") 

277 ctx.persist_callback(ctx.checkpoint) 

278 return record 

279 

280 

281def _helper_stack_id_prefix(ctx: RunContext, spec: Mapping[str, Any]) -> str: 

282 return ( 

283 f"arn:{spec['partition']}:cloudformation:{spec['region']}:" 

284 f"{ctx.settings.expected_account}:stack/{spec['stack_name']}/" 

285 ) 

286 

287 

288def _record_log_cleanup_helper_stack( 

289 ctx: RunContext, 

290 spec: Mapping[str, Any], 

291 stack_id: str, 

292 status: str, 

293) -> None: 

294 if not stack_id.startswith(_helper_stack_id_prefix(ctx, spec)): 

295 raise RuntimeError(f"Cleanup helper returned an invalid stack ID: {stack_id}") 

296 with ctx.state_lock: 

297 record = _prepare_log_cleanup_helper_record(ctx, spec) 

298 active_stack_id = str(record.get("active_stack_id") or "") 

299 if active_stack_id and active_stack_id != stack_id: 

300 raise RuntimeError("Cleanup helper stack generation changed without absence proof") 

301 history = record.setdefault("stack_history", []) 

302 if not isinstance(history, list): 

303 raise RuntimeError("Checkpoint cleanup helper stack_history must be a list") 

304 if not any(item.get("stack_id") == stack_id for item in history if isinstance(item, dict)): 

305 history.append({"stack_id": stack_id, "first_observed_at": utc_now()}) 

306 record["active_stack_id"] = stack_id 

307 record["lifecycle"] = status 

308 record["last_observed_at"] = utc_now() 

309 ctx.persist_callback(ctx.checkpoint) 

310 

311 

312def _mark_log_cleanup_helper_absent( 

313 ctx: RunContext, 

314 stack_id: str | None, 

315) -> None: 

316 with ctx.state_lock: 

317 record = ctx.checkpoint.state.get("log_cleanup_helper") 

318 if not isinstance(record, dict): 

319 return 

320 active_stack_id = str(record.get("active_stack_id") or "") 

321 if stack_id and active_stack_id and active_stack_id != stack_id: 

322 raise RuntimeError("Cleanup helper absence proof refers to a different stack") 

323 record["active_stack_id"] = None 

324 record["lifecycle"] = "deleted" 

325 record["last_deleted_stack_id"] = stack_id 

326 record["deleted_at"] = utc_now() 

327 ctx.persist_callback(ctx.checkpoint) 

328 

329 

330def _template_document(template_body: Any) -> dict[str, Any]: 

331 if isinstance(template_body, str): 

332 try: 

333 template_body = json.loads(template_body) 

334 except json.JSONDecodeError as exc: 

335 raise RuntimeError("Cleanup helper template is not canonical JSON") from exc 

336 if not isinstance(template_body, dict): 

337 raise RuntimeError("Cleanup helper template is not a JSON object") 

338 return template_body 

339 

340 

341def _validate_log_cleanup_helper_stack( 

342 ctx: RunContext, 

343 spec: Mapping[str, Any], 

344 stack: Mapping[str, Any], 

345) -> str: 

346 stack_id = str(stack.get("stack_id") or "") 

347 if ( 

348 stack.get("name") != spec["stack_name"] 

349 or not stack_id.startswith(_helper_stack_id_prefix(ctx, spec)) 

350 or stack.get("termination_protection") 

351 ): 

352 raise RuntimeError("Cleanup helper CloudFormation identity is invalid") 

353 tags = stack.get("tags") or {} 

354 if ( 

355 tags.get(_LOG_CLEANUP_HELPER_RUN_TAG) != spec["run_id"] 

356 or tags.get(_LOG_CLEANUP_HELPER_TOKEN_TAG) != spec["cleanup_token"] 

357 or tags.get("gco:project") is not None 

358 or tags.get("Project") is not None 

359 ): 

360 raise RuntimeError("Cleanup helper CloudFormation tags are invalid") 

361 cfn = ctx.session.client("cloudformation", region_name=spec["region"]) 

362 body = _template_document( 

363 cfn.get_template(StackName=stack_id, TemplateStage="Original").get("TemplateBody") 

364 ) 

365 observed_hash = hashlib.sha256(_canonical_json(body).encode("utf-8")).hexdigest() 

366 if observed_hash != spec["template_sha256"]: 

367 raise RuntimeError("Cleanup helper CloudFormation template changed") 

368 return stack_id 

369 

370 

371def _validate_log_cleanup_helper_role( 

372 ctx: RunContext, 

373 spec: Mapping[str, Any], 

374 helper_record: dict[str, Any], 

375 stack_id: str, 

376) -> dict[str, str]: 

377 iam = ctx.session.client("iam", region_name=spec["region"]) 

378 role = iam.get_role(RoleName=spec["role_name"]).get("Role") 

379 if not isinstance(role, dict): 

380 raise RuntimeError("IAM omitted the cleanup helper role") 

381 tags = { 

382 str(item.get("Key")): str(item.get("Value") or "") 

383 for item in role.get("Tags", []) 

384 if item.get("Key") is not None 

385 } 

386 if ( 

387 str(role.get("RoleName") or "") != spec["role_name"] 

388 or str(role.get("Arn") or "") != spec["role_arn"] 

389 or str(role.get("Path") or "") != "/" 

390 or int(role.get("MaxSessionDuration") or 0) != 3600 

391 or role.get("AssumeRolePolicyDocument") != spec["trust_policy"] 

392 or tags.get(_LOG_CLEANUP_ROLE_RUN_TAG) != spec["run_id"] 

393 or tags.get(_LOG_CLEANUP_ROLE_TOKEN_TAG) != spec["cleanup_token"] 

394 ): 

395 raise RuntimeError("Cleanup helper IAM role identity changed") 

396 inline = iam.list_role_policies(RoleName=spec["role_name"]) 

397 if inline.get("IsTruncated") or inline.get("PolicyNames") != [_LOG_CLEANUP_ROLE_POLICY_NAME]: 

398 raise RuntimeError("Cleanup helper IAM inline policies changed") 

399 role_policy = iam.get_role_policy( 

400 RoleName=spec["role_name"], 

401 PolicyName=_LOG_CLEANUP_ROLE_POLICY_NAME, 

402 ).get("PolicyDocument") 

403 if role_policy != spec["role_policy"]: 

404 raise RuntimeError("Cleanup helper IAM delete policy changed") 

405 attached = iam.list_attached_role_policies(RoleName=spec["role_name"]) 

406 if attached.get("IsTruncated") or attached.get("AttachedPolicies"): 

407 raise RuntimeError("Cleanup helper IAM role gained a managed policy") 

408 created = role.get("CreateDate") 

409 identity = { 

410 "arn": str(role["Arn"]), 

411 "role_id": str(role.get("RoleId") or ""), 

412 "created_at": created.isoformat() if created is not None else "", 

413 } 

414 if not identity["role_id"] or not identity["created_at"]: 

415 raise RuntimeError("IAM omitted immutable cleanup role identity") 

416 history = helper_record.get("stack_history") 

417 if not isinstance(history, list): 

418 raise RuntimeError("Checkpoint cleanup helper stack_history must be a list") 

419 generation = next( 

420 ( 

421 item 

422 for item in history 

423 if isinstance(item, dict) and str(item.get("stack_id") or "") == stack_id 

424 ), 

425 None, 

426 ) 

427 if generation is None: 

428 raise RuntimeError("Cleanup role identity has no exact helper stack generation") 

429 observed = generation.get("observed_role_identity") 

430 if observed is None: 

431 generation["observed_role_identity"] = identity 

432 ctx.persist() 

433 elif observed != identity: 

434 raise RuntimeError("Cleanup helper IAM role generation changed within its stack") 

435 return identity 

436 

437 

438def _wait_for_log_cleanup_helper( 

439 ctx: RunContext, 

440 spec: Mapping[str, Any], 

441 stack_id: str, 

442 *, 

443 deleting: bool, 

444) -> dict[str, Any] | None: 

445 for _attempt in range(_LOG_CLEANUP_STACK_POLL_ATTEMPTS): 

446 stack = describe_stack(ctx.session, str(spec["region"]), stack_id) 

447 status = str((stack or {}).get("status") or "") 

448 if deleting and (stack is None or status == "DELETE_COMPLETE"): 

449 return None 

450 if not deleting and stack is not None and status == "CREATE_COMPLETE": 

451 return stack 

452 if status == "DELETE_FAILED": 

453 raise RuntimeError(f"Cleanup helper stack deletion failed: {stack_id}") 

454 if not deleting and stack is not None and not status.endswith("_IN_PROGRESS"): 

455 raise RuntimeError(f"Cleanup helper stack creation ended in {status}: {stack_id}") 

456 time.sleep(_LOG_CLEANUP_STACK_POLL_SECONDS) 

457 operation = "deletion" if deleting else "creation" 

458 raise RuntimeError(f"Cleanup helper stack {operation} timed out: {stack_id}") 

459 

460 

461def _current_cleanup_trusted_principal(ctx: RunContext) -> tuple[str, dict[str, str]]: 

462 identity = ctx.session.client("sts", region_name=ctx.config.global_region).get_caller_identity() 

463 account = str(identity.get("Account") or "") 

464 caller_arn = str(identity.get("Arn") or "") 

465 if account != ctx.settings.expected_account: 

466 raise RuntimeError("Cleanup helper caller account changed") 

467 return caller_arn, _cleanup_principal_identity(ctx, caller_arn) 

468 

469 

470def _ensure_log_cleanup_helper(ctx: RunContext) -> dict[str, Any]: 

471 records = ctx.checkpoint.state.get("owned_log_groups", []) 

472 if not isinstance(records, list): 

473 raise RuntimeError("Checkpoint owned_log_groups must be a list") 

474 if not records or all(bool(record.get("deleted")) for record in records): 

475 return {"needed": False} 

476 spec = _log_cleanup_helper_spec(ctx) 

477 if spec is None: 

478 return {"needed": False} 

479 helper_record = _prepare_log_cleanup_helper_record(ctx, spec) 

480 caller_arn, current_principal = _current_cleanup_trusted_principal(ctx) 

481 if ( 

482 current_principal["arn"] != spec["trusted_principal_arn"] 

483 or current_principal["principal_id"] != spec["trusted_principal_id"] 

484 ): 

485 raise RuntimeError("Cleanup helper caller principal changed since authority creation") 

486 

487 region = str(spec["region"]) 

488 cfn = ctx.session.client("cloudformation", region_name=region) 

489 stack: dict[str, Any] | None = None 

490 active_stack_id = str(helper_record.get("active_stack_id") or "") 

491 if active_stack_id: 

492 stack = describe_stack(ctx.session, region, active_stack_id) 

493 if stack is not None and stack.get("status") == "DELETE_COMPLETE": 

494 _mark_log_cleanup_helper_absent(ctx, active_stack_id) 

495 active_stack_id = "" 

496 stack = None 

497 if stack is None: 

498 named_stack = describe_stack(ctx.session, region, str(spec["stack_name"])) 

499 if named_stack is not None and named_stack.get("status") != "DELETE_COMPLETE": 

500 named_stack_id = _validate_log_cleanup_helper_stack(ctx, spec, named_stack) 

501 if active_stack_id and named_stack_id != active_stack_id: 

502 raise RuntimeError("A different cleanup helper stack generation appeared") 

503 _record_log_cleanup_helper_stack( 

504 ctx, 

505 spec, 

506 named_stack_id, 

507 str(named_stack.get("status") or ""), 

508 ) 

509 active_stack_id = named_stack_id 

510 stack = named_stack 

511 if stack is not None and stack.get("status") == "DELETE_IN_PROGRESS": 

512 _wait_for_log_cleanup_helper(ctx, spec, active_stack_id, deleting=True) 

513 _mark_log_cleanup_helper_absent(ctx, active_stack_id) 

514 active_stack_id = "" 

515 stack = None 

516 

517 if stack is None: 

518 with ctx.state_lock: 

519 helper_record = _prepare_log_cleanup_helper_record(ctx, spec) 

520 helper_record["create_sequence"] = int(helper_record.get("create_sequence") or 0) + 1 

521 sequence = helper_record["create_sequence"] 

522 helper_record["lifecycle"] = "create-intent" 

523 helper_record["create_intent_at"] = utc_now() 

524 ctx.persist_callback(ctx.checkpoint) 

525 token = f"live-validation-{spec['stack_name']}-{sequence}" 

526 try: 

527 response = cfn.create_stack( 

528 StackName=spec["stack_name"], 

529 TemplateBody=spec["template_body"], 

530 Capabilities=["CAPABILITY_NAMED_IAM"], 

531 ClientRequestToken=token[:128], 

532 EnableTerminationProtection=False, 

533 OnFailure="ROLLBACK", 

534 TimeoutInMinutes=10, 

535 Tags=[ 

536 {"Key": _LOG_CLEANUP_HELPER_RUN_TAG, "Value": spec["run_id"]}, 

537 {"Key": _LOG_CLEANUP_HELPER_TOKEN_TAG, "Value": spec["cleanup_token"]}, 

538 ], 

539 ) 

540 active_stack_id = str(response.get("StackId") or "") 

541 _record_log_cleanup_helper_stack(ctx, spec, active_stack_id, "CREATE_IN_PROGRESS") 

542 except ClientError as exc: 

543 if exc.response.get("Error", {}).get("Code") != "AlreadyExistsException": 

544 raise 

545 stack = describe_stack(ctx.session, region, str(spec["stack_name"])) 

546 if stack is None: 

547 raise RuntimeError("Cleanup helper name exists but cannot be described") from exc 

548 active_stack_id = _validate_log_cleanup_helper_stack(ctx, spec, stack) 

549 _record_log_cleanup_helper_stack( 

550 ctx, 

551 spec, 

552 active_stack_id, 

553 str(stack.get("status") or ""), 

554 ) 

555 

556 stack = _wait_for_log_cleanup_helper(ctx, spec, active_stack_id, deleting=False) 

557 if stack is None: 

558 raise RuntimeError("Cleanup helper disappeared after creation") 

559 _validate_log_cleanup_helper_stack(ctx, spec, stack) 

560 outputs = stack.get("outputs") or {} 

561 if outputs.get(_LOG_CLEANUP_ROLE_OUTPUT) != spec["role_arn"]: 

562 raise RuntimeError("Cleanup helper role output changed") 

563 helper_record = _prepare_log_cleanup_helper_record(ctx, spec) 

564 _validate_log_cleanup_helper_role(ctx, spec, helper_record, active_stack_id) 

565 _record_log_cleanup_helper_stack(ctx, spec, active_stack_id, "CREATE_COMPLETE") 

566 return { 

567 "needed": True, 

568 "region": region, 

569 "stack_id": active_stack_id, 

570 "stack_name": spec["stack_name"], 

571 "role_arn": spec["role_arn"], 

572 "partition": spec["partition"], 

573 "caller_arn": caller_arn, 

574 "trusted_principal_arn": spec["trusted_principal_arn"], 

575 "session_policy": spec["role_policy"], 

576 "external_id": spec["cleanup_token"], 

577 } 

578 

579 

580def _delete_log_cleanup_helper(ctx: RunContext) -> dict[str, Any]: 

581 helper_record = ctx.checkpoint.state.get("log_cleanup_helper") 

582 if helper_record is None: 

583 return {"needed": False, "deleted": True} 

584 if not isinstance(helper_record, dict): 

585 raise RuntimeError("Checkpoint log_cleanup_helper must be an object") 

586 spec = _log_cleanup_helper_spec(ctx) 

587 if spec is None: 

588 raise RuntimeError("Cleanup helper exists without log-group authority records") 

589 helper_record = _prepare_log_cleanup_helper_record(ctx, spec) 

590 region = str(spec["region"]) 

591 cfn = ctx.session.client("cloudformation", region_name=region) 

592 active_stack_id = str(helper_record.get("active_stack_id") or "") 

593 stack = describe_stack(ctx.session, region, active_stack_id) if active_stack_id else None 

594 if stack is not None and stack.get("status") == "DELETE_COMPLETE": 

595 stack = None 

596 if stack is None: 

597 named_stack = describe_stack(ctx.session, region, str(spec["stack_name"])) 

598 if named_stack is not None and named_stack.get("status") != "DELETE_COMPLETE": 

599 named_stack_id = _validate_log_cleanup_helper_stack(ctx, spec, named_stack) 

600 if active_stack_id and named_stack_id != active_stack_id: 

601 raise RuntimeError("Refusing to delete a replacement cleanup helper stack") 

602 active_stack_id = named_stack_id 

603 stack = named_stack 

604 _record_log_cleanup_helper_stack( 

605 ctx, 

606 spec, 

607 active_stack_id, 

608 str(stack.get("status") or ""), 

609 ) 

610 if stack is None: 

611 _mark_log_cleanup_helper_absent(ctx, active_stack_id or None) 

612 return { 

613 "needed": bool(active_stack_id), 

614 "deleted": True, 

615 "already_absent": True, 

616 "stack_id": active_stack_id or None, 

617 } 

618 

619 active_stack_id = _validate_log_cleanup_helper_stack(ctx, spec, stack) 

620 status = str(stack.get("status") or "") 

621 if status == "CREATE_COMPLETE": 

622 _validate_log_cleanup_helper_role(ctx, spec, helper_record, active_stack_id) 

623 if status != "DELETE_IN_PROGRESS": 

624 with ctx.state_lock: 

625 helper_record["lifecycle"] = "delete-intent" 

626 helper_record["delete_intent_at"] = utc_now() 

627 ctx.persist_callback(ctx.checkpoint) 

628 cfn.delete_stack( 

629 StackName=active_stack_id, 

630 ClientRequestToken=( 

631 f"delete-{spec['stack_name']}-{active_stack_id.rsplit('/', 1)[-1]}"[:128] 

632 ), 

633 ) 

634 helper_record["lifecycle"] = "DELETE_IN_PROGRESS" 

635 ctx.persist() 

636 _wait_for_log_cleanup_helper(ctx, spec, active_stack_id, deleting=True) 

637 replacement = describe_stack(ctx.session, region, str(spec["stack_name"])) 

638 if replacement is not None and replacement.get("status") != "DELETE_COMPLETE": 

639 raise RuntimeError("Cleanup helper stack name was replaced during deletion") 

640 _mark_log_cleanup_helper_absent(ctx, active_stack_id) 

641 return {"needed": True, "deleted": True, "stack_id": active_stack_id} 

642 

643 

644class TagConditionedLogDeleter: 

645 """Hand out ``logs`` clients whose DeleteLogGroup is tag-conditioned. 

646 

647 Deleting a retained log group is only safe while the group still carries 

648 both of this run's authority tags. Rather than trust a read that happened 

649 moments earlier, deletion goes through a delegated role whose session 

650 policy makes every ``logs:DeleteLogGroup`` call conditional on those exact 

651 tag values: if a foreign generation replaced the group between the 

652 immediate pre-delete read and the request, AWS refuses the call instead of 

653 destroying someone else's log data. 

654 

655 The helper stack, role, and STS session are created on first use, so a run 

656 whose checkpointed groups all turn out to be absent never provisions any of 

657 them. ``authorization`` is the report evidence for that decision and stays 

658 ``{"needed": False}`` until a session is actually established. 

659 """ 

660 

661 def __init__(self, ctx: RunContext) -> None: 

662 self._ctx = ctx 

663 self._clients: dict[str, Any] = {} 

664 self._credentials: dict[str, Any] | None = None 

665 self.authorization: dict[str, Any] = {"needed": False} 

666 

667 def client(self, region: str) -> Any: 

668 """Return the tag-conditioned ``logs`` client for one Region.""" 

669 if self._credentials is None: 

670 self._establish_session() 

671 credentials = self._credentials 

672 if credentials is None: 

673 raise RuntimeError("Log cleanup session was not established") 

674 if region not in self._clients: 

675 self._clients[region] = self._ctx.session.client( 

676 "logs", 

677 region_name=region, 

678 aws_access_key_id=credentials["AccessKeyId"], 

679 aws_secret_access_key=credentials["SecretAccessKey"], 

680 aws_session_token=credentials["SessionToken"], 

681 ) 

682 return self._clients[region] 

683 

684 def _establish_session(self) -> None: 

685 """Assume the scoped cleanup role once and verify the exact principal.""" 

686 ctx = self._ctx 

687 helper = _ensure_log_cleanup_helper(ctx) 

688 if not helper.get("needed"): 

689 raise RuntimeError("Log cleanup role was not created for pending groups")