Coverage for scripts / live_release_validation / inventory / stacks.py: 100.00%

1"""CloudFormation stack discovery, description, and fingerprinting.""" 

2 

3from __future__ import annotations 

4 

5import hashlib 

6import json 

7from collections.abc import Iterable 

8from typing import Any 

9 

10from botocore.exceptions import ClientError 

11 

12from ._shared import ( 

13 _normalize_json_text, 

14 _project_owned_name, 

15 _tags_to_dict, 

16) 

17 

18 

19def discover_enabled_regions(session: Any, seed_region: str) -> list[str]: 

20 """Return every enabled Region in the caller's partition that supports CFN.""" 

21 partition = session.get_partition_for_region(seed_region) 

22 if not partition: 

23 raise RuntimeError(f"Could not resolve AWS partition for {seed_region}") 

24 available = set(session.get_available_regions("cloudformation", partition_name=partition)) 

25 ec2 = session.client("ec2", region_name=seed_region) 

26 response = ec2.describe_regions(AllRegions=False) 

27 enabled = { 

28 str(item["RegionName"]) 

29 for item in response.get("Regions", []) 

30 if item.get("RegionName") 

31 and item.get("OptInStatus") in {None, "opt-in-not-required", "opted-in"} 

32 } 

33 regions = sorted(enabled & available) 

34 if not regions: 

35 raise RuntimeError( 

36 f"No enabled CloudFormation Regions were discovered in partition {partition}" 

37 ) 

38 return regions 

39 

40 

41def list_active_stacks(session: Any, region: str) -> list[dict[str, str]]: 

42 """List every non-deleted CloudFormation stack in one Region.""" 

43 client = session.client("cloudformation", region_name=region) 

44 stacks: list[dict[str, str]] = [] 

45 for page in client.get_paginator("list_stacks").paginate(): 

46 for summary in page.get("StackSummaries", []): 

47 status = str(summary.get("StackStatus") or "") 

48 if status == "DELETE_COMPLETE": 

49 continue 

50 name = str(summary.get("StackName") or "") 

51 stack_id = str(summary.get("StackId") or "") 

52 if name: 

53 stacks.append({"name": name, "stack_id": stack_id, "status": status}) 

54 return sorted(stacks, key=lambda item: (item["name"], item["stack_id"])) 

55 

56 

57def describe_stack(session: Any, region: str, stack_name: str) -> dict[str, Any] | None: 

58 """Describe one stack; return None only for authoritative nonexistence.""" 

59 client = session.client("cloudformation", region_name=region) 

60 try: 

61 response = client.describe_stacks(StackName=stack_name) 

62 except ClientError as exc: 

63 error = exc.response.get("Error", {}) 

64 if ( 

65 error.get("Code") == "ValidationError" 

66 and "does not exist" in str(error.get("Message", "")).lower() 

67 ): 

68 return None 

69 raise 

70 stacks = response.get("Stacks", []) 

71 if not stacks: 

72 return None 

73 stack = stacks[0] 

74 return { 

75 "name": str(stack.get("StackName") or stack_name), 

76 "stack_id": str(stack.get("StackId") or ""), 

77 "status": str(stack.get("StackStatus") or ""), 

78 "parameters": sorted( 

79 ( 

80 { 

81 "key": str(item["ParameterKey"]), 

82 "value": str(item.get("ParameterValue") or ""), 

83 "resolved_value": str(item.get("ResolvedValue") or ""), 

84 } 

85 for item in stack.get("Parameters", []) 

86 if item.get("ParameterKey") is not None 

87 ), 

88 key=lambda item: item["key"], 

89 ), 

90 "outputs": { 

91 str(item["OutputKey"]): str(item["OutputValue"]) 

92 for item in stack.get("Outputs", []) 

93 if item.get("OutputKey") is not None and item.get("OutputValue") is not None 

94 }, 

95 "tags": _tags_to_dict(stack.get("Tags", [])), 

96 "termination_protection": bool(stack.get("EnableTerminationProtection", False)), 

97 } 

98 

99 

100def _list_stack_resource_identities(client: Any, stack_id: str) -> list[dict[str, str]]: 

101 """Return every stack resource's exact physical identity in stable order.""" 

102 identities: list[dict[str, str]] = [] 

103 logical_ids: set[str] = set() 

104 for page in client.get_paginator("list_stack_resources").paginate(StackName=stack_id): 

105 for resource in page.get("StackResourceSummaries", []): 

106 logical_id = str(resource.get("LogicalResourceId") or "") 

107 resource_type = str(resource.get("ResourceType") or "") 

108 physical_id = str(resource.get("PhysicalResourceId") or "") 

109 if not logical_id or not resource_type or not physical_id: 

110 raise RuntimeError( 

111 "CloudFormation omitted a protected stack resource identity for " 

112 f"{stack_id}: {json.dumps(resource, sort_keys=True, default=str)}" 

113 ) 

114 if logical_id in logical_ids: 

115 raise RuntimeError( 

116 f"CloudFormation duplicated protected stack resource {stack_id}:{logical_id}" 

117 ) 

118 logical_ids.add(logical_id) 

119 identities.append( 

120 { 

121 "logical_id": logical_id, 

122 "resource_type": resource_type, 

123 "physical_id": physical_id, 

124 } 

125 ) 

126 return sorted( 

127 identities, 

128 key=lambda item: (item["logical_id"], item["resource_type"], item["physical_id"]), 

129 ) 

130 

131 

132def describe_stack_fingerprint( 

133 session: Any, 

134 region: str, 

135 stack_name: str, 

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

137 """Return template, policy, and exact physical-resource fingerprints.""" 

138 stack = describe_stack(session, region, stack_name) 

139 if stack is None: 

140 return None 

141 client = session.client("cloudformation", region_name=region) 

142 template = _normalize_json_text( 

143 client.get_template( 

144 StackName=stack["stack_id"], 

145 TemplateStage="Original", 

146 ).get("TemplateBody", "") 

147 ) 

148 if isinstance(template, (dict, list)): 

149 template_bytes = json.dumps( 

150 template, 

151 sort_keys=True, 

152 separators=(",", ":"), 

153 ).encode("utf-8") 

154 else: 

155 template_bytes = str(template).encode("utf-8") 

156 

157 try: 

158 stack_policy = _normalize_json_text( 

159 client.get_stack_policy(StackName=stack["stack_id"]).get("StackPolicyBody") 

160 ) 

161 except ClientError as exc: 

162 error = exc.response.get("Error", {}) 

163 message = str(error.get("Message", "")).lower() 

164 if error.get("Code") == "ValidationError" and "stack policy" in message: 

165 stack_policy = None 

166 else: 

167 raise 

168 return { 

169 **stack, 

170 "template_sha256": hashlib.sha256(template_bytes).hexdigest(), 

171 "stack_policy": stack_policy, 

172 "physical_resources": _list_stack_resource_identities(client, stack["stack_id"]), 

173 } 

174 

175 

176def collect_stack_inventory( 

177 session: Any, regions: Iterable[str] 

178) -> dict[str, list[dict[str, str]]]: 

179 """Collect active stacks across Regions with deterministic ordering.""" 

180 return {region: list_active_stacks(session, region) for region in sorted(set(regions))} 

181 

182 

183def collect_project_stacks( 

184 session: Any, 

185 regions: Iterable[str], 

186 project_name: str, 

187) -> dict[str, list[dict[str, str]]]: 

188 """Collect all project-prefixed stacks, including unexpected/orphan stacks.""" 

189 inventory = collect_stack_inventory(session, regions) 

190 return { 

191 region: [stack for stack in stacks if _project_owned_name(stack["name"], project_name)] 

192 for region, stacks in inventory.items() 

193 if any(_project_owned_name(stack["name"], project_name) for stack in stacks) 

194 }