Coverage for scripts / live_release_validation / cleanup / retained.py: 100.00%

1"""Schedule and account for intentionally retained resources.""" 

2 

3from __future__ import annotations 

4 

5import copy 

6import json 

7from typing import Any 

8 

9from botocore.exceptions import ClientError 

10 

11from ..cleanup.ecr import _cleanup_new_ecr_images, _cleanup_new_ecr_repositories 

12from ..cleanup.log_groups import ( 

13 _cleanup_owned_log_groups, 

14) 

15from ..constants import ( 

16 _KMS_PENDING_WINDOW_DAYS, 

17 _RUN_STACK_TAG, 

18 _LogGroupCleanupError, 

19) 

20from ..models import RunContext, utc_now 

21from ..ownership.kms import ( 

22 _kms_tags, 

23 _validated_owned_kms_identity, 

24) 

25 

26 

27def _schedule_retained_kms_keys(ctx: RunContext) -> dict[str, Any]: 

28 records = ctx.checkpoint.state.get("owned_kms_keys", []) 

29 retained_records = [ 

30 record 

31 for record in records 

32 if _validated_owned_kms_identity(ctx, record)[3] == "harness-schedule" 

33 ] 

34 if retained_records and not ctx.settings.confirm_kms_key_deletion: 

35 raise RuntimeError("Retained KMS keys exist but this identity did not confirm key deletion") 

36 results: list[dict[str, Any]] = [] 

37 for record in records: 

38 region, key_id, arn, cleanup_policy = _validated_owned_kms_identity(ctx, record) 

39 record.setdefault("cleanup_policy", cleanup_policy) 

40 kms = ctx.session.client("kms", region_name=region) 

41 try: 

42 metadata = kms.describe_key(KeyId=key_id).get("KeyMetadata", {}) 

43 except ClientError as exc: 

44 if exc.response.get("Error", {}).get("Code") == "NotFoundException": 

45 record["scheduled"] = True 

46 record["deleted"] = True 

47 results.append( 

48 { 

49 "arn": arn, 

50 "cleanup_policy": cleanup_policy, 

51 "already_absent": True, 

52 } 

53 ) 

54 ctx.persist() 

55 continue 

56 raise 

57 if metadata.get("Arn") != arn: 

58 raise RuntimeError(f"KMS key ARN changed for {key_id}") 

59 tags = _kms_tags(kms, key_id) 

60 if tags.get(_RUN_STACK_TAG) != record["run_tag"]: 

61 raise RuntimeError(f"KMS run ownership changed for {arn}") 

62 

63 state = str(metadata.get("KeyState") or "") 

64 if state != "PendingDeletion" and cleanup_policy == "harness-schedule": 

65 if state not in {"Enabled", "Disabled"}: 

66 raise RuntimeError(f"KMS key {arn} is {state}; refusing to schedule deletion") 

67 kms.schedule_key_deletion( 

68 KeyId=key_id, 

69 PendingWindowInDays=_KMS_PENDING_WINDOW_DAYS, 

70 ) 

71 metadata = kms.describe_key(KeyId=key_id).get("KeyMetadata", {}) 

72 state = str(metadata.get("KeyState") or "") 

73 if state != "PendingDeletion": 

74 raise RuntimeError( 

75 f"Expected {cleanup_policy} KMS key {arn} to be PendingDeletion; found {state}" 

76 ) 

77 deletion_date = metadata.get("DeletionDate") 

78 record["scheduled"] = True 

79 record["deletion_date"] = deletion_date.isoformat() if deletion_date is not None else None 

80 if not record["deletion_date"]: 

81 raise RuntimeError(f"Pending-deletion KMS key omitted its deletion date: {arn}") 

82 results.append( 

83 { 

84 "arn": arn, 

85 "state": state, 

86 "cleanup_policy": cleanup_policy, 

87 "deletion_date": record["deletion_date"], 

88 } 

89 ) 

90 ctx.persist() 

91 return { 

92 "keys": results, 

93 "deletion_window": { 

94 "harness_schedule_days": _KMS_PENDING_WINDOW_DAYS, 

95 "cloudformation_delete": "observed per key deletion_date", 

96 }, 

97 } 

98 

99 

100def _retained_resource_cleanup(ctx: RunContext) -> dict[str, Any]: 

101 result: dict[str, Any] = {"started_at": utc_now(), "errors": []} 

102 try: 

103 result["cloudwatch_logs"] = _cleanup_owned_log_groups(ctx) 

104 except Exception as exc: # noqa: BLE001 - preserve partial evidence 

105 if isinstance(exc, _LogGroupCleanupError): 

106 result["cloudwatch_logs"] = copy.deepcopy(exc.details) 

107 result["errors"].append( 

108 {"phase": "cloudwatch-logs", "error": f"{type(exc).__name__}: {exc}"} 

109 ) 

110 try: 

111 result["ecr_images"] = _cleanup_new_ecr_images(ctx) 

112 except Exception as exc: # noqa: BLE001 - preserve partial evidence 

113 result["errors"].append({"phase": "ecr-images", "error": f"{type(exc).__name__}: {exc}"}) 

114 try: 

115 result["ecr_repositories"] = _cleanup_new_ecr_repositories(ctx) 

116 except Exception as exc: # noqa: BLE001 - preserve partial evidence 

117 result["errors"].append( 

118 {"phase": "ecr-repositories", "error": f"{type(exc).__name__}: {exc}"} 

119 ) 

120 try: 

121 result["kms"] = _schedule_retained_kms_keys(ctx) 

122 except Exception as exc: # noqa: BLE001 - preserve partial evidence 

123 result["errors"].append({"phase": "kms", "error": f"{type(exc).__name__}: {exc}"}) 

124 result["ended_at"] = utc_now() 

125 ctx.checkpoint.state.setdefault("retained_cleanup_attempts", []).append(result) 

126 ctx.persist() 

127 if result["errors"]: 

128 raise RuntimeError( 

129 "Retained resource cleanup failed: " + json.dumps(result["errors"], sort_keys=True) 

130 ) 

131 return result