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

1"""Ownership-matching primitives and category lists shared by the scanners. 

2 

3A resource counts as project-owned only through an explicit signal: its 

4CloudFormation stack-name tag, a ``gco:project`` tag, or a name/ARN under 

5the project prefix. Everything here is deliberately conservative, because 

6a false positive would let teardown consider a pre-existing account 

7resource in scope.""" 

8 

9from __future__ import annotations 

10 

11import json 

12from collections.abc import Iterable 

13from typing import Any 

14 

15_ECR_MANIFEST_MEDIA_TYPES = ( 

16 "application/vnd.docker.distribution.manifest.v1+json", 

17 "application/vnd.docker.distribution.manifest.v2+json", 

18 "application/vnd.docker.distribution.manifest.list.v2+json", 

19 "application/vnd.oci.image.manifest.v1+json", 

20 "application/vnd.oci.image.index.v1+json", 

21) 

22 

23 

24_GLOBAL_ACCELERATOR_CONTROL_REGIONS = {"aws": "us-west-2"} 

25 

26 

27_REGIONAL_PROJECT_RESOURCE_CATEGORIES = ( 

28 "tagged_resources", 

29 "eks_clusters", 

30 "sqs_queues", 

31 "dynamodb_tables", 

32 "load_balancers", 

33 "target_groups", 

34 "instances", 

35 "cluster_volumes", 

36 "vpcs", 

37 "subnets", 

38 "nat_gateways", 

39 "flow_logs", 

40 "network_interfaces", 

41 "security_groups", 

42 "elastic_ips", 

43 "ecr_repositories", 

44 "kms_keys", 

45 "lambda_functions", 

46 "api_gateway_v1_apis", 

47 "api_gateway_v2_apis", 

48 "cloudwatch_log_groups", 

49 "secrets", 

50 "backup_vaults", 

51 "backup_plans", 

52 "backup_selections", 

53 "backup_recovery_points", 

54) 

55 

56 

57_GLOBAL_PROJECT_RESOURCE_CATEGORIES = ( 

58 "global_accelerators", 

59 "s3_buckets", 

60 "iam_roles", 

61 "iam_policies", 

62 "iam_instance_profiles", 

63 "iam_users", 

64 "iam_groups", 

65) 

66 

67 

68_PROJECT_RESOURCE_CATEGORIES = ( 

69 "cloudformation_stacks", 

70 *_REGIONAL_PROJECT_RESOURCE_CATEGORIES, 

71 *_GLOBAL_PROJECT_RESOURCE_CATEGORIES, 

72) 

73 

74 

75_PROJECT_RESOURCE_SCANNERS = ( 

76 "cloudformation_stacks", 

77 "resource_groups_tagging_api", 

78 "eks_clusters", 

79 "sqs_queues", 

80 "dynamodb_tables", 

81 "load_balancers", 

82 "target_groups", 

83 "ec2_instances", 

84 "ec2_networking", 

85 "ecr_repositories", 

86 "kms_keys", 

87 "lambda_functions", 

88 "api_gateway_v1_apis", 

89 "api_gateway_v2_apis", 

90 "cloudwatch_log_groups", 

91 "secrets_manager", 

92 "cluster_volumes", 

93 "aws_backup", 

94 "s3_buckets", 

95 "iam", 

96 "global_accelerators", 

97) 

98 

99 

100def _project_owned_name(name: str, project_name: str) -> bool: 

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

102 

103 

104def _tags_to_dict(tags: Iterable[dict[str, Any]]) -> dict[str, str]: 

105 return { 

106 str(tag.get("Key")): str(tag.get("Value")) for tag in tags if tag.get("Key") is not None 

107 } 

108 

109 

110def _tags_are_project_owned(tags: dict[str, str], project_name: str) -> bool: 

111 stack_name = tags.get("aws:cloudformation:stack-name", "") 

112 explicit_project = tags.get("gco:project", "") 

113 return _project_owned_name(stack_name, project_name) or explicit_project == project_name 

114 

115 

116def _normalize_json_text(value: Any) -> Any: 

117 if not isinstance(value, str): 

118 return value 

119 try: 

120 return json.loads(value) 

121 except json.JSONDecodeError: 

122 return value 

123 

124 

125def _mapping_tags(tags: Any) -> dict[str, str]: 

126 if tags is None: 

127 return {} 

128 if not isinstance(tags, dict): 

129 raise RuntimeError("AWS returned tags in an unexpected format") 

130 return {str(key): str(value) for key, value in tags.items()} 

131 

132 

133def _name_or_path_is_project_owned(value: str, project_name: str) -> bool: 

134 if _project_owned_name(value, project_name): 

135 return True 

136 components = [component for component in value.replace(":", "/").split("/") if component] 

137 return any(_project_owned_name(component, project_name) for component in components) 

138 

139 

140def _arn_is_project_owned(arn: str, project_name: str) -> bool: 

141 parts = arn.split(":", 5) 

142 if len(parts) != 6: 

143 return False 

144 components = [component for component in parts[5].replace(":", "/").split("/") if component] 

145 if len(components) > 1: 

146 components = components[1:] 

147 return any(_project_owned_name(component, project_name) for component in components) 

148 

149 

150def _ec2_resource_is_project_owned(resource: dict[str, Any], project_name: str) -> bool: 

151 tags = _tags_to_dict(resource.get("Tags", [])) 

152 return _tags_are_project_owned(tags, project_name) or _project_owned_name( 

153 tags.get("Name", ""), project_name 

154 ) 

155 

156 

157def _iam_resource_is_project_owned( 

158 name: str, 

159 path: str, 

160 tags: dict[str, str], 

161 project_name: str, 

162) -> bool: 

163 return ( 

164 _project_owned_name(name, project_name) 

165 or _name_or_path_is_project_owned(path, project_name) 

166 or _tags_are_project_owned(tags, project_name) 

167 )