Coverage for scripts / live_release_validation / protected.py: 100.00%
183 statements
« prev ^ index » next coverage.py v7.13.5, created at 2026-09-14 22:07 +0000
1"""Protected-resource identity matching for the ownership boundary."""
3from __future__ import annotations
5import re
6from collections.abc import Mapping
7from typing import Any
8from urllib.parse import urlsplit
10_PROTECTED_REGIONAL_RESOURCE_CATEGORIES = {
11 "AWS::ApiGateway::RestApi": "api_gateway_v1_apis",
12 "AWS::ApiGatewayV2::Api": "api_gateway_v2_apis",
13 "AWS::Backup::BackupPlan": "backup_plans",
14 "AWS::Backup::BackupSelection": "backup_selections",
15 "AWS::Backup::BackupVault": "backup_vaults",
16 "AWS::DynamoDB::Table": "dynamodb_tables",
17 "AWS::EC2::EIP": "elastic_ips",
18 "AWS::EC2::FlowLog": "flow_logs",
19 "AWS::EC2::Instance": "instances",
20 "AWS::EC2::NatGateway": "nat_gateways",
21 "AWS::EC2::NetworkInterface": "network_interfaces",
22 "AWS::EC2::SecurityGroup": "security_groups",
23 "AWS::EC2::Subnet": "subnets",
24 "AWS::EC2::VPC": "vpcs",
25 "AWS::ECR::Repository": "ecr_repositories",
26 "AWS::EKS::Cluster": "eks_clusters",
27 "AWS::ElasticLoadBalancingV2::LoadBalancer": "load_balancers",
28 "AWS::ElasticLoadBalancingV2::TargetGroup": "target_groups",
29 "AWS::KMS::Key": "kms_keys",
30 "AWS::Lambda::Function": "lambda_functions",
31 "AWS::Logs::LogGroup": "cloudwatch_log_groups",
32 "AWS::SQS::Queue": "sqs_queues",
33 "AWS::SecretsManager::Secret": "secrets",
34}
37_PROTECTED_GLOBAL_RESOURCE_CATEGORIES = {
38 "AWS::GlobalAccelerator::Accelerator": "global_accelerators",
39 "AWS::IAM::Group": "iam_groups",
40 "AWS::IAM::InstanceProfile": "iam_instance_profiles",
41 "AWS::IAM::ManagedPolicy": "iam_policies",
42 "AWS::IAM::Role": "iam_roles",
43 "AWS::IAM::User": "iam_users",
44 "AWS::S3::Bucket": "s3_buckets",
45}
48_IAM_ARN_RESOURCE_KINDS = {
49 "AWS::IAM::Group": "group",
50 "AWS::IAM::InstanceProfile": "instance-profile",
51 "AWS::IAM::ManagedPolicy": "policy",
52 "AWS::IAM::Role": "role",
53 "AWS::IAM::User": "user",
54}
57_BACKUP_ARN_RESOURCE_PREFIXES = {
58 "AWS::Backup::BackupPlan": "backup-plan:",
59 "AWS::Backup::BackupVault": "backup-vault:",
60}
63_EC2_ID_SUFFIX = r"(?:[0-9a-f]{8}|[0-9a-f]{17})"
66_EC2_TAGGED_RESOURCE_IDENTITIES: dict[str, tuple[str, re.Pattern[str]]] = {
67 "elastic-ip": ("elastic_ips", re.compile(rf"eipalloc-{_EC2_ID_SUFFIX}")),
68 "instance": ("instances", re.compile(rf"i-{_EC2_ID_SUFFIX}")),
69 "natgateway": ("nat_gateways", re.compile(rf"nat-{_EC2_ID_SUFFIX}")),
70 "network-interface": ("network_interfaces", re.compile(rf"eni-{_EC2_ID_SUFFIX}")),
71 "security-group": ("security_groups", re.compile(rf"sg-{_EC2_ID_SUFFIX}")),
72 "subnet": ("subnets", re.compile(rf"subnet-{_EC2_ID_SUFFIX}")),
73 "vpc": ("vpcs", re.compile(rf"vpc-{_EC2_ID_SUFFIX}")),
74 "vpc-flow-log": ("flow_logs", re.compile(rf"fl-{_EC2_ID_SUFFIX}")),
75}
78_EKS_CLUSTER_NAME = re.compile(r"[0-9A-Za-z][A-Za-z0-9_-]{0,99}")
81_EKS_ASSOCIATION_ID = re.compile(r"a-[0-9a-z]{17}")
84_KUBERNETES_DNS_LABEL = re.compile(r"[a-z0-9](?:[-a-z0-9]{0,61}[a-z0-9])?")
87_KUBERNETES_POD_UID = re.compile(r"[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}")
90_CLOUDFORMATION_STACK_ID_TAG = "aws:cloudformation:stack-id"
93_SQS_DNS_SUFFIXES = {
94 "aws": "amazonaws.com",
95 "aws-cn": "amazonaws.com.cn",
96 "aws-us-gov": "amazonaws.com",
97}
100def _baseline_protected_identities(
101 baseline: dict[str, Any],
102) -> tuple[dict[str, set[str]], dict[str, dict[str, set[str]]]]:
103 """Validate and index exact protected stack and physical-resource identities."""
104 protected_stacks = baseline.get("protected_stacks")
105 if protected_stacks is None:
106 protected_stacks = {}
107 elif not isinstance(protected_stacks, dict):
108 raise RuntimeError("Baseline protected_stacks must be an object")
110 stack_ids: dict[str, set[str]] = {}
111 resource_ids: dict[str, dict[str, set[str]]] = {}
112 for raw_region, stacks in protected_stacks.items():
113 region = str(raw_region)
114 if not isinstance(stacks, list):
115 raise RuntimeError(f"Protected stack baseline for {region} must be a list")
116 for stack in stacks:
117 if not isinstance(stack, dict):
118 raise RuntimeError(f"Protected stack baseline for {region} is malformed")
119 stack_id = str(stack.get("stack_id") or "")
120 if not stack_id:
121 raise RuntimeError(f"Protected stack baseline for {region} omitted its stack ID")
122 stack_ids.setdefault(region, set()).add(stack_id)
123 resources = stack.get("physical_resources")
124 if not isinstance(resources, list):
125 raise RuntimeError(
126 f"Protected stack baseline {region}:{stack_id} omitted physical resources"
127 )
128 for resource in resources:
129 if not isinstance(resource, dict):
130 raise RuntimeError(
131 f"Protected stack baseline {region}:{stack_id} has a malformed resource"
132 )
133 resource_type = str(resource.get("resource_type") or "")
134 physical_id = str(resource.get("physical_id") or "")
135 logical_id = str(resource.get("logical_id") or "")
136 if not resource_type or not physical_id or not logical_id:
137 raise RuntimeError(
138 f"Protected stack baseline {region}:{stack_id} has an incomplete resource"
139 )
140 resource_ids.setdefault(region, {}).setdefault(resource_type, set()).add(
141 physical_id
142 )
143 return stack_ids, resource_ids
146def _iam_arn_name(arn: str, resource_kind: str) -> str | None:
147 parts = arn.split(":", 5)
148 if len(parts) != 6 or parts[0] != "arn" or parts[2] != "iam":
149 return None
150 prefix = f"{resource_kind}/"
151 resource = parts[5]
152 if not resource.startswith(prefix):
153 return None
154 name = resource[len(prefix) :].rsplit("/", 1)[-1]
155 return name or None
158def _lambda_arn_name(arn: str) -> str | None:
159 parts = arn.split(":", 5)
160 if len(parts) != 6 or parts[0] != "arn" or parts[2] != "lambda":
161 return None
162 resource = parts[5]
163 if not resource.startswith("function:"):
164 return None
165 name = resource.removeprefix("function:")
166 return name if name and ":" not in name else None
169def _backup_arn_physical_id(arn: str, resource_prefix: str) -> str | None:
170 parts = arn.split(":", 5)
171 if len(parts) != 6 or parts[0] != "arn" or parts[2] != "backup":
172 return None
173 resource = parts[5]
174 if not resource.startswith(resource_prefix):
175 return None
176 physical_id = resource.removeprefix(resource_prefix)
177 return physical_id or None
180def _sqs_queue_name_from_physical_id(
181 physical_id: str,
182 *,
183 partition: str,
184 region: str,
185 account_id: str,
186) -> str | None:
187 """Normalize an exact SQS queue name or CloudFormation queue URL."""
188 if "://" not in physical_id:
189 return physical_id or None
191 dns_suffix = _SQS_DNS_SUFFIXES.get(partition)
192 parsed = urlsplit(physical_id)
193 if (
194 dns_suffix is None
195 or parsed.scheme != "https"
196 or parsed.hostname != f"sqs.{region}.{dns_suffix}"
197 or parsed.query
198 or parsed.fragment
199 ):
200 return None
201 path_parts = parsed.path.strip("/").split("/")
202 if len(path_parts) != 2 or path_parts[0] != account_id:
203 return None
204 return path_parts[1] or None
207def _tagged_arn_matches_protected_physical_id(
208 resource_type: str,
209 arn: str,
210 physical_id: str,
211 *,
212 expected_partition: str,
213 expected_region: str,
214 expected_account: str,
215) -> bool:
216 """Match a Tagging API ARN to one exact CloudFormation physical ID."""
217 if arn == physical_id:
218 return True
220 parts = arn.split(":", 5)
221 if len(parts) != 6 or parts[0] != "arn":
222 return False
223 partition, service, region, account_id, resource = (
224 parts[1],
225 parts[2],
226 parts[3],
227 parts[4],
228 parts[5],
229 )
231 if resource_type == "AWS::Lambda::Function":
232 return _lambda_arn_name(arn) == physical_id
233 if resource_type == "AWS::DynamoDB::Table":
234 return (
235 service == "dynamodb"
236 and resource.startswith("table/")
237 and "/" not in resource.removeprefix("table/")
238 and resource.removeprefix("table/") == physical_id
239 )
240 if resource_type == "AWS::S3::Bucket":
241 return (
242 service == "s3" and bool(resource) and "/" not in resource and resource == physical_id
243 )
244 if resource_type == "AWS::SQS::Queue":
245 if service != "sqs" or not region or not account_id or not resource:
246 return False
247 queue_name = _sqs_queue_name_from_physical_id(
248 physical_id,
249 partition=partition,
250 region=region,
251 account_id=account_id,
252 )
253 return queue_name == resource
254 if resource_type == "AWS::KMS::Key":
255 return (
256 service == "kms"
257 and resource.startswith("key/")
258 and "/" not in resource.removeprefix("key/")
259 and resource.removeprefix("key/") == physical_id
260 )
261 expected_ec2_category = {
262 "AWS::EC2::FlowLog": "flow_logs",
263 "AWS::EC2::NatGateway": "nat_gateways",
264 }.get(resource_type)
265 if expected_ec2_category is not None:
266 return bool(
267 expected_partition
268 and expected_region
269 and re.fullmatch(r"\d{12}", expected_account)
270 and _ec2_tagged_resource_identity(
271 arn,
272 expected_region,
273 expected_partition,
274 expected_account,
275 )
276 == (expected_ec2_category, physical_id)
277 )
278 return False
281def _tagged_resource_is_protected(
282 record: Any,
283 *,
284 protected_stack_ids: set[str],
285 protected_resource_ids: dict[str, set[str]],
286 exact_arns: set[str],
287 expected_partition: str,
288 expected_region: str,
289 expected_account: str,
290) -> bool:
291 """Return whether a tagged record has one exact protected identity."""
292 if not isinstance(record, Mapping):
293 return False
294 arn = str(record.get("arn") or "")
295 if not arn:
296 return False
297 if arn in exact_arns:
298 return True
300 tags = record.get("tags")
301 if isinstance(tags, Mapping):
302 stack_id = str(tags.get(_CLOUDFORMATION_STACK_ID_TAG) or "")
303 if stack_id in protected_stack_ids:
304 return True
306 return any(
307 _tagged_arn_matches_protected_physical_id(
308 resource_type,
309 arn,
310 physical_id,
311 expected_partition=expected_partition,
312 expected_region=expected_region,
313 expected_account=expected_account,
314 )
315 for resource_type, physical_ids in protected_resource_ids.items()
316 for physical_id in physical_ids
317 )
320def _matches_protected_physical_identity(
321 resource_type: str,
322 category: str,
323 candidate: Any,
324 physical_id: str,
325 *,
326 protected_backup_plan_ids: set[str] | None = None,
327) -> bool:
328 """Match one inventory record without any prefix or ownership-name fallback."""
329 if category == "kms_keys":
330 if not isinstance(candidate, dict):
331 return False
332 return physical_id in {
333 str(candidate.get("key_id") or ""),
334 str(candidate.get("arn") or ""),
335 }
336 if not isinstance(candidate, str):
337 return False
338 if resource_type == "AWS::Backup::BackupSelection":
339 plan_id, separator, selection_id = candidate.partition(":")
340 return bool(
341 separator
342 and selection_id == physical_id
343 and plan_id in (protected_backup_plan_ids or set())
344 )
345 if candidate == physical_id:
346 return True
347 backup_prefix = _BACKUP_ARN_RESOURCE_PREFIXES.get(resource_type)
348 if backup_prefix is not None:
349 return _backup_arn_physical_id(candidate, backup_prefix) == physical_id
350 iam_kind = _IAM_ARN_RESOURCE_KINDS.get(resource_type)
351 if iam_kind is not None:
352 return _iam_arn_name(candidate, iam_kind) == physical_id
353 if resource_type == "AWS::Lambda::Function":
354 return _lambda_arn_name(candidate) == physical_id
355 return False
358def _valid_kubernetes_dns_subdomain(value: str) -> bool:
359 return bool(
360 value
361 and len(value) <= 253
362 and all(_KUBERNETES_DNS_LABEL.fullmatch(label) for label in value.split("."))
363 )
366def _eks_pod_parent_cluster(
367 arn: str,
368 expected_region: str,
369 expected_partition: str,
370 expected_account: str,
371) -> str | None:
372 """Parse canonical in-scope EKS pod identities; malformed records stay visible."""
373 parts = arn.split(":", 5)
374 if (
375 len(parts) != 6
376 or parts[0] != "arn"
377 or parts[1] != expected_partition
378 or parts[2] != "eks"
379 or parts[3] != expected_region
380 or parts[4] != expected_account
381 ):
382 return None
383 resource_parts = parts[5].split("/")
384 if any(not component for component in resource_parts):
385 return None
386 if resource_parts[0] == "pod":
387 if len(resource_parts) != 5:
388 return None
389 _kind, cluster, namespace, pod_name, pod_uid = resource_parts
390 if (
391 not _EKS_CLUSTER_NAME.fullmatch(cluster)
392 or not _KUBERNETES_DNS_LABEL.fullmatch(namespace)
393 or not _valid_kubernetes_dns_subdomain(pod_name)
394 or not _KUBERNETES_POD_UID.fullmatch(pod_uid)
395 ):
396 return None
397 return cluster
398 if resource_parts[0] == "podidentityassociation":
399 if len(resource_parts) != 3:
400 return None
401 _kind, cluster, association_id = resource_parts
402 if not _EKS_CLUSTER_NAME.fullmatch(cluster) or not _EKS_ASSOCIATION_ID.fullmatch(
403 association_id
404 ):
405 return None
406 return cluster
407 return None
410def _ec2_tagged_resource_identity(
411 arn: str,
412 expected_region: str,
413 expected_partition: str,
414 expected_account: str,
415) -> tuple[str, str] | None:
416 """Map only canonical in-scope EC2 ARNs to authoritative live identities."""
417 parts = arn.split(":", 5)
418 if (
419 len(parts) != 6
420 or parts[0] != "arn"
421 or parts[1] != expected_partition
422 or parts[2] != "ec2"
423 or parts[3] != expected_region
424 or parts[4] != expected_account
425 ):
426 return None
427 resource_kind, separator, resource_id = parts[5].partition("/")
428 identity = _EC2_TAGGED_RESOURCE_IDENTITIES.get(resource_kind)
429 if not separator or not resource_id or "/" in resource_id or identity is None:
430 return None
431 category, resource_id_pattern = identity
432 if not resource_id_pattern.fullmatch(resource_id):
433 return None
434 return category, resource_id