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 ..cleanup.transaction_search import ( 

16 TransactionSearchRestoreError, 

17 _restore_transaction_search, 

18) 

19from ..constants import ( 

20 _KMS_PENDING_WINDOW_DAYS, 

21 _RUN_STACK_TAG, 

22 _LogGroupCleanupError, 

23) 

24from ..models import RunContext, utc_now 

25from ..ownership.kms import ( 

26 _kms_tags, 

27 _validated_owned_kms_identity, 

28) 

29 

30 

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

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

33 retained_records = [ 

34 record 

35 for record in records 

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

37 ] 

38 if retained_records and not ctx.settings.confirm_kms_key_deletion: 

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

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

41 for record in records: 

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

43 record.setdefault("cleanup_policy", cleanup_policy) 

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

45 try: 

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

47 except ClientError as exc: 

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

49 record["scheduled"] = True 

50 record["deleted"] = True 

51 results.append( 

52 { 

53 "arn": arn, 

54 "cleanup_policy": cleanup_policy, 

55 "already_absent": True, 

56 } 

57 ) 

58 ctx.persist() 

59 continue 

60 raise 

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

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

63 tags = _kms_tags(kms, key_id) 

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

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

66 

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

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

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

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

71 kms.schedule_key_deletion( 

72 KeyId=key_id, 

73 PendingWindowInDays=_KMS_PENDING_WINDOW_DAYS, 

74 ) 

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

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

77 if state != "PendingDeletion": 

78 raise RuntimeError( 

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

80 ) 

81 deletion_date = metadata.get("DeletionDate") 

82 record["scheduled"] = True 

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

84 if not record["deletion_date"]: 

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

86 results.append( 

87 { 

88 "arn": arn, 

89 "state": state, 

90 "cleanup_policy": cleanup_policy, 

91 "deletion_date": record["deletion_date"], 

92 } 

93 ) 

94 ctx.persist() 

95 return { 

96 "keys": results, 

97 "deletion_window": { 

98 "harness_schedule_days": _KMS_PENDING_WINDOW_DAYS, 

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

100 }, 

101 } 

102 

103 

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

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

106 try: 

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

108 except Exception as exc: # preserve partial evidence 

109 if isinstance(exc, _LogGroupCleanupError): 

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

111 result["errors"].append( 

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

113 ) 

114 try: 

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

116 except Exception as exc: # preserve partial evidence 

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

118 try: 

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

120 except Exception as exc: # preserve partial evidence 

121 result["errors"].append( 

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

123 ) 

124 try: 

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

126 except Exception as exc: # preserve partial evidence 

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

128 # Account-level state the stacks' deletion never reverts: the enabler's 

129 # Delete is a no-op, so the run puts Transaction Search back itself. 

130 try: 

131 result["transaction_search"] = _restore_transaction_search(ctx) 

132 except Exception as exc: # preserve partial evidence 

133 if isinstance(exc, TransactionSearchRestoreError): 

134 result["transaction_search"] = copy.deepcopy(exc.details) 

135 result["errors"].append( 

136 {"phase": "transaction-search", "error": f"{type(exc).__name__}: {exc}"} 

137 ) 

138 result["ended_at"] = utc_now() 

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

140 ctx.persist() 

141 if result["errors"]: 

142 raise RuntimeError( 

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

144 ) 

145 return result