ea4d3a2c-ec60-4c10-9aae-c0c88ca77069

Committed 13 Apr 2025 11:57PM UTC coverage: 86.431% (-0.1%) from 86.564%

Build # ea4d3a2c-ec60-4c10-9aae-c0c88ca77069

Build Type

push

circleci

Committed by

web-flow

Commit Message

add AVP to list of CFN composite quirks (#12517)

Coverage Stats

63584 of 73566 relevant lines covered (86.43%)

0.86 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

81.1

/localstack-core/localstack/utils/aws/arns.py

import logging
import re
from functools import cache
from typing import Optional, TypedDict

from botocore.utils import ArnParser, InvalidArnException

from localstack.aws.accounts import DEFAULT_AWS_ACCOUNT_ID
from localstack.aws.connect import connect_to
from localstack.utils.strings import long_uid

LOG = logging.getLogger(__name__)

#
# Partition Utilities
#

DEFAULT_PARTITION = "aws"
REGION_PREFIX_TO_PARTITION = {
    # (region prefix, aws partition)
    "cn-": "aws-cn",
    "us-gov-": "aws-us-gov",
    "us-iso-": "aws-iso",
    "us-isob-": "aws-iso-b",
}
PARTITION_NAMES = list(REGION_PREFIX_TO_PARTITION.values()) + [DEFAULT_PARTITION]
ARN_PARTITION_REGEX = r"^arn:(" + "|".join(sorted(PARTITION_NAMES)) + ")"


def get_partition(region: Optional[str]) -> str:
    if not region:
        return DEFAULT_PARTITION
    if region in PARTITION_NAMES:
        return region
    for prefix in REGION_PREFIX_TO_PARTITION:
        if region.startswith(prefix):
            return REGION_PREFIX_TO_PARTITION[prefix]
    return DEFAULT_PARTITION


#
# ARN parsing utilities
#


class ArnData(TypedDict):
    partition: str
    service: str
    region: str
    account: str
    resource: str


_arn_parser = ArnParser()


def parse_arn(arn: str) -> ArnData:
    """
    Uses a botocore ArnParser to parse an arn.

    :param arn: the arn string to parse
    :returns: a dictionary containing the ARN components
    :raises InvalidArnException: if the arn is invalid
    """
    return _arn_parser.parse_arn(arn)


def extract_account_id_from_arn(arn: str) -> Optional[str]:
    try:
        return parse_arn(arn).get("account")
    except InvalidArnException:
        return None


def extract_region_from_arn(arn: str) -> Optional[str]:
    try:
        return parse_arn(arn).get("region")
    except InvalidArnException:
        return None


def extract_service_from_arn(arn: str) -> Optional[str]:
    try:
        return parse_arn(arn).get("service")
    except InvalidArnException:
        return None


def extract_resource_from_arn(arn: str) -> Optional[str]:
    try:
        return parse_arn(arn).get("resource")
    except InvalidArnException:
        return None


#
# Generic ARN builder
#


def _resource_arn(name: str, pattern: str, account_id: str, region_name: str) -> str:
    if ":" in name:
        return name
    if len(pattern.split("%s")) == 4:
        return pattern % (get_partition(region_name), account_id, name)
    return pattern % (get_partition(region_name), region_name, account_id, name)


#
# ARN builders for specific resource types
#

#
# IAM
#


def iam_role_arn(role_name: str, account_id: str, region_name: str) -> str:
    if not role_name:
        return role_name
    if re.match(f"{ARN_PARTITION_REGEX}:iam::", role_name):
        return role_name
    return "arn:%s:iam::%s:role/%s" % (get_partition(region_name), account_id, role_name)


def iam_resource_arn(resource: str, account_id: str, role: str = None) -> str:
    if not role:
        role = f"role-{resource}"
    # Only used in tests, so we can hardcode the region for now
    return iam_role_arn(role_name=role, account_id=account_id, region_name="us-east-1")


#
# Secretsmanager
#


def secretsmanager_secret_arn(
    secret_id: str, account_id: str, region_name: str, random_suffix: str = None
) -> str:
    if ":" in (secret_id or ""):
        return secret_id
    pattern = "arn:%s:secretsmanager:%s:%s:secret:%s"
    arn = _resource_arn(secret_id, pattern, account_id=account_id, region_name=region_name)
    if random_suffix:
        arn += f"-{random_suffix}"
    return arn


#
# Cloudformation
#


def cloudformation_stack_arn(
    stack_name: str, stack_id: str, account_id: str, region_name: str
) -> str:
    pattern = "arn:%s:cloudformation:%s:%s:stack/%s/{stack_id}".format(stack_id=stack_id)
    return _resource_arn(stack_name, pattern, account_id=account_id, region_name=region_name)


def cloudformation_change_set_arn(
    change_set_name: str, change_set_id: str, account_id: str, region_name: str
) -> str:
    pattern = "arn:%s:cloudformation:%s:%s:changeSet/%s/{cs_id}".format(cs_id=change_set_id)
    return _resource_arn(change_set_name, pattern, account_id=account_id, region_name=region_name)


#
# DynamoDB
#


def dynamodb_table_arn(table_name: str, account_id: str, region_name: str) -> str:
    table_name = table_name.split(":table/")[-1]
    pattern = "arn:%s:dynamodb:%s:%s:table/%s"
    return _resource_arn(table_name, pattern, account_id=account_id, region_name=region_name)


def dynamodb_stream_arn(
    table_name: str, latest_stream_label: str, account_id: str, region_name: str
) -> str:
    return "arn:%s:dynamodb:%s:%s:table/%s/stream/%s" % (
        get_partition(region_name),
        region_name,
        account_id,
        table_name,
        latest_stream_label,
    )


#
# Cloudwatch
#


def cloudwatch_alarm_arn(alarm_name: str, account_id: str, region_name: str) -> str:
    # format pattern directly as alarm_name can include ":" and this is not supported by the helper _resource_arn
    return (
        f"arn:{get_partition(region_name)}:cloudwatch:{region_name}:{account_id}:alarm:{alarm_name}"
    )


def cloudwatch_dashboard_arn(dashboard_name: str, account_id: str, region_name: str) -> str:
    pattern = "arn:%s:cloudwatch::%s:dashboard/%s"
    return _resource_arn(dashboard_name, pattern, account_id=account_id, region_name=region_name)


#
# Logs
#


def log_group_arn(group_name: str, account_id: str, region_name: str) -> str:
    pattern = "arn:%s:logs:%s:%s:log-group:%s"
    return _resource_arn(group_name, pattern, account_id=account_id, region_name=region_name)


#
# Events
#


def events_archive_arn(archive_name: str, account_id: str, region_name: str) -> str:
    pattern = "arn:%s:events:%s:%s:archive/%s"
    return _resource_arn(archive_name, pattern, account_id=account_id, region_name=region_name)


def event_bus_arn(bus_name: str, account_id: str, region_name: str) -> str:
    pattern = "arn:%s:events:%s:%s:event-bus/%s"
    return _resource_arn(bus_name, pattern, account_id=account_id, region_name=region_name)


def events_replay_arn(replay_name: str, account_id: str, region_name: str) -> str:
    pattern = "arn:%s:events:%s:%s:replay/%s"
    return _resource_arn(replay_name, pattern, account_id=account_id, region_name=region_name)


def events_rule_arn(
    rule_name: str, account_id: str, region_name: str, event_bus_name: str = "default"
) -> str:
    pattern = "arn:%s:events:%s:%s:rule/%s"
    if event_bus_name != "default":
        rule_name = f"{event_bus_name}/{rule_name}"
    return _resource_arn(rule_name, pattern, account_id=account_id, region_name=region_name)


def events_connection_arn(
    connection_name: str, connection_id: str, account_id: str, region_name: str
) -> str:
    name = f"{connection_name}/{connection_id}"
    pattern = "arn:%s:events:%s:%s:connection/%s"
    return _resource_arn(name, pattern, account_id=account_id, region_name=region_name)


def events_api_destination_arn(
    api_destination_name: str, api_destination_id: str, account_id: str, region_name: str
) -> str:
    name = f"{api_destination_name}/{api_destination_id}"
    pattern = "arn:%s:events:%s:%s:api-destination/%s"
    return _resource_arn(name, pattern, account_id=account_id, region_name=region_name)


#
# Lambda
#


def lambda_function_arn(function_name: str, account_id: str, region_name: str) -> str:
    return lambda_function_or_layer_arn(
        "function", function_name, version=None, account_id=account_id, region_name=region_name
    )


def lambda_layer_arn(layer_name: str, account_id: str, region_name: str) -> str:
    return lambda_function_or_layer_arn(
        "layer", layer_name, version=None, account_id=account_id, region_name=region_name
    )


def lambda_code_signing_arn(code_signing_id: str, account_id: str, region_name: str) -> str:
    pattern = "arn:%s:lambda:%s:%s:code-signing-config:%s"
    return _resource_arn(code_signing_id, pattern, account_id=account_id, region_name=region_name)


def lambda_event_source_mapping_arn(uuid: str, account_id: str, region_name: str) -> str:
    pattern = "arn:%s:lambda:%s:%s:event-source-mapping:%s"
    return _resource_arn(uuid, pattern, account_id=account_id, region_name=region_name)


def lambda_function_or_layer_arn(
    type: str,
    entity_name: str,
    version: Optional[str],
    account_id: str,
    region_name: str,
) -> str:
    pattern = "arn:([a-z-]+):lambda:.*:.*:(function|layer):.*"
    if re.match(pattern, entity_name):
        return entity_name
    if ":" in entity_name:
        client = connect_to(aws_access_key_id=account_id, region_name=region_name).lambda_
        entity_name, _, alias = entity_name.rpartition(":")
        try:
            alias_response = client.get_alias(FunctionName=entity_name, Name=alias)
            version = alias_response["FunctionVersion"]

        except Exception as e:
            msg = f"Alias {alias} of {entity_name} not found"
            LOG.info("%s: %s", msg, e)
            raise Exception(msg)

    result = (
        f"arn:{get_partition(region_name)}:lambda:{region_name}:{account_id}:{type}:{entity_name}"
    )
    if version:
        result = f"{result}:{version}"
    return result


#
# Stepfunctions
#


def stepfunctions_state_machine_arn(name: str, account_id: str, region_name: str) -> str:
    pattern = "arn:%s:states:%s:%s:stateMachine:%s"
    return _resource_arn(name, pattern, account_id=account_id, region_name=region_name)


def stepfunctions_standard_execution_arn(state_machine_arn: str, execution_name: str) -> str:
    arn_data: ArnData = parse_arn(state_machine_arn)
    standard_execution_arn = ":".join(
        [
            "arn",
            arn_data["partition"],
            arn_data["service"],
            arn_data["region"],
            arn_data["account"],
            "execution",
            "".join(arn_data["resource"].split(":")[1:]),
            execution_name,
        ]
    )
    return standard_execution_arn


def stepfunctions_express_execution_arn(state_machine_arn: str, execution_name: str) -> str:
    arn_data: ArnData = parse_arn(state_machine_arn)
    express_execution_arn = ":".join(
        [
            "arn",
            arn_data["partition"],
            arn_data["service"],
            arn_data["region"],
            arn_data["account"],
            "express",
            "".join(arn_data["resource"].split(":")[1:]),
            execution_name,
            long_uid(),
        ]
    )
    return express_execution_arn


def stepfunctions_activity_arn(name: str, account_id: str, region_name: str) -> str:
    pattern = "arn:%s:states:%s:%s:activity:%s"
    return _resource_arn(name, pattern, account_id=account_id, region_name=region_name)


#
# Cognito
#


def cognito_user_pool_arn(user_pool_id: str, account_id: str, region_name: str) -> str:
    pattern = "arn:%s:cognito-idp:%s:%s:userpool/%s"
    return _resource_arn(user_pool_id, pattern, account_id=account_id, region_name=region_name)


#
# Kinesis
#


def kinesis_stream_arn(stream_name: str, account_id: str, region_name: str) -> str:
    pattern = "arn:%s:kinesis:%s:%s:stream/%s"
    return _resource_arn(stream_name, pattern, account_id, region_name)


#
# Elasticsearch
#


def elasticsearch_domain_arn(domain_name: str, account_id: str, region_name: str) -> str:
    pattern = "arn:%s:es:%s:%s:domain/%s"
    return _resource_arn(domain_name, pattern, account_id=account_id, region_name=region_name)


#
# Firehose
#


def firehose_stream_arn(stream_name: str, account_id: str, region_name: str) -> str:
    pattern = "arn:%s:firehose:%s:%s:deliverystream/%s"
    return _resource_arn(stream_name, pattern, account_id=account_id, region_name=region_name)


#
# KMS
#


def kms_key_arn(key_id: str, account_id: str, region_name: str) -> str:
    pattern = "arn:%s:kms:%s:%s:key/%s"
    return _resource_arn(key_id, pattern, account_id=account_id, region_name=region_name)


def kms_alias_arn(alias_name: str, account_id: str, region_name: str):
    if not alias_name.startswith("alias/"):
        alias_name = "alias/" + alias_name
    pattern = "arn:%s:kms:%s:%s:%s"
    return _resource_arn(alias_name, pattern, account_id=account_id, region_name=region_name)


#
# SSM
#


def ssm_parameter_arn(param_name: str, account_id: str, region_name: str) -> str:
    pattern = "arn:%s:ssm:%s:%s:parameter/%s"
    param_name = param_name.lstrip("/")
    return _resource_arn(param_name, pattern, account_id=account_id, region_name=region_name)


#
# S3
#


def s3_bucket_arn(bucket_name_or_arn: str, region="us-east-1") -> str:
    bucket_name = s3_bucket_name(bucket_name_or_arn)
    return f"arn:{get_partition(region)}:s3:::{bucket_name}"


#
# SQS
#


def sqs_queue_arn(queue_name: str, account_id: str, region_name: str) -> str:
    queue_name = queue_name.split("/")[-1]
    return "arn:%s:sqs:%s:%s:%s" % (get_partition(region_name), region_name, account_id, queue_name)


#
# APIGW
#


def apigateway_restapi_arn(api_id: str, account_id: str, region_name: str) -> str:
    return "arn:%s:apigateway:%s:%s:/restapis/%s" % (
        get_partition(region_name),
        region_name,
        account_id,
        api_id,
    )


def apigateway_invocations_arn(lambda_uri: str, region_name: str) -> str:
    return "arn:%s:apigateway:%s:lambda:path/2015-03-31/functions/%s/invocations" % (
        get_partition(region_name),
        region_name,
        lambda_uri,
    )


#
# SNS
#


def sns_topic_arn(topic_name: str, account_id: str, region_name: str) -> str:
    return f"arn:{get_partition(region_name)}:sns:{region_name}:{account_id}:{topic_name}"


#
# ECR
#


def ecr_repository_arn(name: str, account_id: str, region_name: str) -> str:
    pattern = "arn:%s:ecr:%s:%s:repository/%s"
    return _resource_arn(name, pattern, account_id=account_id, region_name=region_name)


#
# Route53
#


def route53_resolver_firewall_rule_group_arn(id: str, account_id: str, region_name: str) -> str:
    pattern = "arn:%s:route53resolver:%s:%s:firewall-rule-group/%s"
    return _resource_arn(id, pattern, account_id=account_id, region_name=region_name)


def route53_resolver_firewall_domain_list_arn(id: str, account_id: str, region_name: str) -> str:
    pattern = "arn:%s:route53resolver:%s:%s:firewall-domain-list/%s"
    return _resource_arn(id, pattern, account_id=account_id, region_name=region_name)


def route53_resolver_firewall_rule_group_associations_arn(
    id: str, account_id: str, region_name: str
) -> str:
    pattern = "arn:%s:route53resolver:%s:%s:firewall-rule-group-association/%s"
    return _resource_arn(id, pattern, account_id=account_id, region_name=region_name)


def route53_resolver_query_log_config_arn(id: str, account_id: str, region_name: str) -> str:
    pattern = "arn:%s:route53resolver:%s:%s:resolver-query-log-config/%s"
    return _resource_arn(id, pattern, account_id=account_id, region_name=region_name)


#
# SES
#


def ses_identity_arn(email: str, account_id: str, region_name: str) -> str:
    pattern = "arn:%s:ses:%s:%s:identity/%s"
    return _resource_arn(email, pattern, account_id=account_id, region_name=region_name)


#
# Other ARN related helpers
#


def opensearch_domain_name(domain_arn: str) -> str:
    return domain_arn.rpartition("/")[2]


def firehose_name(firehose_arn: str) -> str:
    return firehose_arn.split("/")[-1]


def kinesis_stream_name(kinesis_arn: str) -> str:
    return kinesis_arn.split(":stream/")[-1]


def lambda_function_name(name_or_arn: str) -> str:
    if ":" in name_or_arn:
        arn = parse_arn(name_or_arn)
        if arn["service"] != "lambda":
            raise ValueError("arn is not a lambda arn %s" % name_or_arn)

        return parse_arn(name_or_arn)["resource"].split(":")[1]
    else:
        return name_or_arn


@cache
def sqs_queue_url_for_arn(queue_arn: str) -> str:
    """
    Return the SQS queue URL for the given queue ARN.
    """
    if "://" in queue_arn:
        return queue_arn

    try:
        arn = parse_arn(queue_arn)
        account_id = arn["account"]
        region_name = arn["region"]
        queue_name = arn["resource"]
    except InvalidArnException:
        account_id = DEFAULT_AWS_ACCOUNT_ID
        region_name = None
        queue_name = queue_arn

    sqs_client = connect_to(region_name=region_name).sqs
    result = sqs_client.get_queue_url(QueueName=queue_name, QueueOwnerAWSAccountId=account_id)[
        "QueueUrl"
    ]
    return result


def sqs_queue_name(queue_arn: str) -> str:
    if ":" in queue_arn:
        return parse_arn(queue_arn)["resource"]
    else:
        return queue_arn


def s3_bucket_name(bucket_name_or_arn: str) -> str:
    return bucket_name_or_arn.split(":::")[-1]


def is_arn(possible_arn: str) -> bool:
    try:
        parse_arn(possible_arn)
        return True
    except InvalidArnException:
        return False

1	import logging	1✔
2	import re	1✔
3	from functools import cache	1✔
4	from typing import Optional, TypedDict	1✔
5
6	from botocore.utils import ArnParser, InvalidArnException	1✔
7
8	from localstack.aws.accounts import DEFAULT_AWS_ACCOUNT_ID	1✔
9	from localstack.aws.connect import connect_to	1✔
10	from localstack.utils.strings import long_uid	1✔
11
12	LOG = logging.getLogger(__name__)	1✔
13
14	#
15	# Partition Utilities
16	#
17
18	DEFAULT_PARTITION = "aws"	1✔
19	REGION_PREFIX_TO_PARTITION = {	1✔
20	# (region prefix, aws partition)
21	"cn-": "aws-cn",
22	"us-gov-": "aws-us-gov",
23	"us-iso-": "aws-iso",
24	"us-isob-": "aws-iso-b",
25	}
26	PARTITION_NAMES = list(REGION_PREFIX_TO_PARTITION.values()) + [DEFAULT_PARTITION]	1✔
27	ARN_PARTITION_REGEX = r"^arn:(" + "\|".join(sorted(PARTITION_NAMES)) + ")"	1✔
28
29
30	def get_partition(region: Optional[str]) -> str:	1✔
31	if not region:	1✔
32	return DEFAULT_PARTITION	1✔
33	if region in PARTITION_NAMES:	1✔
34	return region	×
35	for prefix in REGION_PREFIX_TO_PARTITION:	1✔
36	if region.startswith(prefix):	1✔
37	return REGION_PREFIX_TO_PARTITION[prefix]	×
38	return DEFAULT_PARTITION	1✔
39
40
41	#
42	# ARN parsing utilities
43	#
44
45
46	class ArnData(TypedDict):	1✔
47	partition: str	1✔
48	service: str	1✔
49	region: str	1✔
50	account: str	1✔
51	resource: str	1✔
52
53
54	_arn_parser = ArnParser()	1✔
55
56
57	def parse_arn(arn: str) -> ArnData:	1✔
58	"""
59	Uses a botocore ArnParser to parse an arn.
60
61	:param arn: the arn string to parse
62	:returns: a dictionary containing the ARN components
63	:raises InvalidArnException: if the arn is invalid
64	"""
65	return _arn_parser.parse_arn(arn)	1✔
66
67
68	def extract_account_id_from_arn(arn: str) -> Optional[str]:	1✔
69	try:	1✔
70	return parse_arn(arn).get("account")	1✔
71	except InvalidArnException:	1✔
72	return None	1✔
73
74
75	def extract_region_from_arn(arn: str) -> Optional[str]:	1✔
76	try:	1✔
77	return parse_arn(arn).get("region")	1✔
78	except InvalidArnException:	1✔
79	return None	1✔
80
81
82	def extract_service_from_arn(arn: str) -> Optional[str]:	1✔
83	try:	1✔
84	return parse_arn(arn).get("service")	1✔
85	except InvalidArnException:	×
86	return None	×
87
88
89	def extract_resource_from_arn(arn: str) -> Optional[str]:	1✔
90	try:	1✔
91	return parse_arn(arn).get("resource")	1✔
92	except InvalidArnException:	×
93	return None	×
94
95
96	#
97	# Generic ARN builder
98	#
99
100
101	def _resource_arn(name: str, pattern: str, account_id: str, region_name: str) -> str:	1✔
102	if ":" in name:	1✔
103	return name	1✔
104	if len(pattern.split("%s")) == 4:	1✔
105	return pattern % (get_partition(region_name), account_id, name)	1✔
106	return pattern % (get_partition(region_name), region_name, account_id, name)	1✔
107
108
109	#
110	# ARN builders for specific resource types
111	#
112
113	#
114	# IAM
115	#
116
117
118	def iam_role_arn(role_name: str, account_id: str, region_name: str) -> str:	1✔
119	if not role_name:	1✔
120	return role_name	×
121	if re.match(f"{ARN_PARTITION_REGEX}:iam::", role_name):	1✔
122	return role_name	×
123	return "arn:%s:iam::%s:role/%s" % (get_partition(region_name), account_id, role_name)	1✔
124
125
126	def iam_resource_arn(resource: str, account_id: str, role: str = None) -> str:	1✔
127	if not role:	1✔
128	role = f"role-{resource}"	1✔
129	# Only used in tests, so we can hardcode the region for now
130	return iam_role_arn(role_name=role, account_id=account_id, region_name="us-east-1")	1✔
131
132
133	#
134	# Secretsmanager
135	#
136
137
138	def secretsmanager_secret_arn(	1✔
139	secret_id: str, account_id: str, region_name: str, random_suffix: str = None
140	) -> str:
141	if ":" in (secret_id or ""):	×
142	return secret_id	×
143	pattern = "arn:%s:secretsmanager:%s:%s:secret:%s"	×
144	arn = _resource_arn(secret_id, pattern, account_id=account_id, region_name=region_name)	×
145	if random_suffix:	×
146	arn += f"-{random_suffix}"	×
147	return arn	×
148
149
150	#
151	# Cloudformation
152	#
153
154
155	def cloudformation_stack_arn(	1✔
156	stack_name: str, stack_id: str, account_id: str, region_name: str
157	) -> str:
158	pattern = "arn:%s:cloudformation:%s:%s:stack/%s/{stack_id}".format(stack_id=stack_id)	1✔
159	return _resource_arn(stack_name, pattern, account_id=account_id, region_name=region_name)	1✔
160
161
162	def cloudformation_change_set_arn(	1✔
163	change_set_name: str, change_set_id: str, account_id: str, region_name: str
164	) -> str:
165	pattern = "arn:%s:cloudformation:%s:%s:changeSet/%s/{cs_id}".format(cs_id=change_set_id)	1✔
166	return _resource_arn(change_set_name, pattern, account_id=account_id, region_name=region_name)	1✔
167
168
169	#
170	# DynamoDB
171	#
172
173
174	def dynamodb_table_arn(table_name: str, account_id: str, region_name: str) -> str:	1✔
175	table_name = table_name.split(":table/")[-1]	1✔
176	pattern = "arn:%s:dynamodb:%s:%s:table/%s"	1✔
177	return _resource_arn(table_name, pattern, account_id=account_id, region_name=region_name)	1✔
178
179
180	def dynamodb_stream_arn(	1✔
181	table_name: str, latest_stream_label: str, account_id: str, region_name: str
182	) -> str:
183	return "arn:%s:dynamodb:%s:%s:table/%s/stream/%s" % (	1✔
184	get_partition(region_name),
185	region_name,
186	account_id,
187	table_name,
188	latest_stream_label,
189	)
190
191
192	#
193	# Cloudwatch
194	#
195
196
197	def cloudwatch_alarm_arn(alarm_name: str, account_id: str, region_name: str) -> str:	1✔
198	# format pattern directly as alarm_name can include ":" and this is not supported by the helper _resource_arn
199	return (	1✔
200	f"arn:{get_partition(region_name)}:cloudwatch:{region_name}:{account_id}:alarm:{alarm_name}"
201	)
202
203
204	def cloudwatch_dashboard_arn(dashboard_name: str, account_id: str, region_name: str) -> str:	1✔
205	pattern = "arn:%s:cloudwatch::%s:dashboard/%s"	1✔
206	return _resource_arn(dashboard_name, pattern, account_id=account_id, region_name=region_name)	1✔
207
208
209	#
210	# Logs
211	#
212
213
214	def log_group_arn(group_name: str, account_id: str, region_name: str) -> str:	1✔
215	pattern = "arn:%s:logs:%s:%s:log-group:%s"	1✔
216	return _resource_arn(group_name, pattern, account_id=account_id, region_name=region_name)	1✔
217
218
219	#
220	# Events
221	#
222
223
224	def events_archive_arn(archive_name: str, account_id: str, region_name: str) -> str:	1✔
225	pattern = "arn:%s:events:%s:%s:archive/%s"	1✔
226	return _resource_arn(archive_name, pattern, account_id=account_id, region_name=region_name)	1✔
227
228
229	def event_bus_arn(bus_name: str, account_id: str, region_name: str) -> str:	1✔
230	pattern = "arn:%s:events:%s:%s:event-bus/%s"	1✔
231	return _resource_arn(bus_name, pattern, account_id=account_id, region_name=region_name)	1✔
232
233
234	def events_replay_arn(replay_name: str, account_id: str, region_name: str) -> str:	1✔
235	pattern = "arn:%s:events:%s:%s:replay/%s"	1✔
236	return _resource_arn(replay_name, pattern, account_id=account_id, region_name=region_name)	1✔
237
238
239	def events_rule_arn(	1✔
240	rule_name: str, account_id: str, region_name: str, event_bus_name: str = "default"
241	) -> str:
242	pattern = "arn:%s:events:%s:%s:rule/%s"	1✔
243	if event_bus_name != "default":	1✔
244	rule_name = f"{event_bus_name}/{rule_name}"	1✔
245	return _resource_arn(rule_name, pattern, account_id=account_id, region_name=region_name)	1✔
246
247
248	def events_connection_arn(	1✔
249	connection_name: str, connection_id: str, account_id: str, region_name: str
250	) -> str:
251	name = f"{connection_name}/{connection_id}"	1✔
252	pattern = "arn:%s:events:%s:%s:connection/%s"	1✔
253	return _resource_arn(name, pattern, account_id=account_id, region_name=region_name)	1✔
254
255
256	def events_api_destination_arn(	1✔
257	api_destination_name: str, api_destination_id: str, account_id: str, region_name: str
258	) -> str:
259	name = f"{api_destination_name}/{api_destination_id}"	1✔
260	pattern = "arn:%s:events:%s:%s:api-destination/%s"	1✔
261	return _resource_arn(name, pattern, account_id=account_id, region_name=region_name)	1✔
262
263
264	#
265	# Lambda
266	#
267
268
269	def lambda_function_arn(function_name: str, account_id: str, region_name: str) -> str:	1✔
270	return lambda_function_or_layer_arn(	1✔
271	"function", function_name, version=None, account_id=account_id, region_name=region_name
272	)
273
274
275	def lambda_layer_arn(layer_name: str, account_id: str, region_name: str) -> str:	1✔
276	return lambda_function_or_layer_arn(	×
277	"layer", layer_name, version=None, account_id=account_id, region_name=region_name
278	)
279
280
281	def lambda_code_signing_arn(code_signing_id: str, account_id: str, region_name: str) -> str:	1✔
282	pattern = "arn:%s:lambda:%s:%s:code-signing-config:%s"	×
283	return _resource_arn(code_signing_id, pattern, account_id=account_id, region_name=region_name)	×
284
285
286	def lambda_event_source_mapping_arn(uuid: str, account_id: str, region_name: str) -> str:	1✔
287	pattern = "arn:%s:lambda:%s:%s:event-source-mapping:%s"	1✔
288	return _resource_arn(uuid, pattern, account_id=account_id, region_name=region_name)	1✔
289
290
291	def lambda_function_or_layer_arn(	1✔
292	type: str,
293	entity_name: str,
294	version: Optional[str],
295	account_id: str,
296	region_name: str,
297	) -> str:
298	pattern = "arn:([a-z-]+):lambda:.:.:(function\|layer):.*"	1✔
299	if re.match(pattern, entity_name):	1✔
300	return entity_name	×
301	if ":" in entity_name:	1✔
302	client = connect_to(aws_access_key_id=account_id, region_name=region_name).lambda_	×
303	entity_name, _, alias = entity_name.rpartition(":")	×
304	try:	×
305	alias_response = client.get_alias(FunctionName=entity_name, Name=alias)	×
306	version = alias_response["FunctionVersion"]	×
307
308	except Exception as e:	×
309	msg = f"Alias {alias} of {entity_name} not found"	×
310	LOG.info("%s: %s", msg, e)	×
311	raise Exception(msg)	×
312
313	result = (	1✔
314	f"arn:{get_partition(region_name)}:lambda:{region_name}:{account_id}:{type}:{entity_name}"
315	)
316	if version:	1✔
317	result = f"{result}:{version}"	×
318	return result	1✔
319
320
321	#
322	# Stepfunctions
323	#
324
325
326	def stepfunctions_state_machine_arn(name: str, account_id: str, region_name: str) -> str:	1✔
327	pattern = "arn:%s:states:%s:%s:stateMachine:%s"	1✔
328	return _resource_arn(name, pattern, account_id=account_id, region_name=region_name)	1✔
329
330
331	def stepfunctions_standard_execution_arn(state_machine_arn: str, execution_name: str) -> str:	1✔
332	arn_data: ArnData = parse_arn(state_machine_arn)	1✔
333	standard_execution_arn = ":".join(	1✔
334	[
335	"arn",
336	arn_data["partition"],
337	arn_data["service"],
338	arn_data["region"],
339	arn_data["account"],
340	"execution",
341	"".join(arn_data["resource"].split(":")[1:]),
342	execution_name,
343	]
344	)
345	return standard_execution_arn	1✔
346
347
348	def stepfunctions_express_execution_arn(state_machine_arn: str, execution_name: str) -> str:	1✔
349	arn_data: ArnData = parse_arn(state_machine_arn)	1✔
350	express_execution_arn = ":".join(	1✔
351	[
352	"arn",
353	arn_data["partition"],
354	arn_data["service"],
355	arn_data["region"],
356	arn_data["account"],
357	"express",
358	"".join(arn_data["resource"].split(":")[1:]),
359	execution_name,
360	long_uid(),
361	]
362	)
363	return express_execution_arn	1✔
364
365
366	def stepfunctions_activity_arn(name: str, account_id: str, region_name: str) -> str:	1✔
367	pattern = "arn:%s:states:%s:%s:activity:%s"	1✔
368	return _resource_arn(name, pattern, account_id=account_id, region_name=region_name)	1✔
369
370
371	#
372	# Cognito
373	#
374
375
376	def cognito_user_pool_arn(user_pool_id: str, account_id: str, region_name: str) -> str:	1✔
377	pattern = "arn:%s:cognito-idp:%s:%s:userpool/%s"	×
378	return _resource_arn(user_pool_id, pattern, account_id=account_id, region_name=region_name)	×
379
380
381	#
382	# Kinesis
383	#
384
385
386	def kinesis_stream_arn(stream_name: str, account_id: str, region_name: str) -> str:	1✔
387	pattern = "arn:%s:kinesis:%s:%s:stream/%s"	1✔
388	return _resource_arn(stream_name, pattern, account_id, region_name)	1✔
389
390
391	#
392	# Elasticsearch
393	#
394
395
396	def elasticsearch_domain_arn(domain_name: str, account_id: str, region_name: str) -> str:	1✔
397	pattern = "arn:%s:es:%s:%s:domain/%s"	×
398	return _resource_arn(domain_name, pattern, account_id=account_id, region_name=region_name)	×
399
400
401	#
402	# Firehose
403	#
404
405
406	def firehose_stream_arn(stream_name: str, account_id: str, region_name: str) -> str:	1✔
407	pattern = "arn:%s:firehose:%s:%s:deliverystream/%s"	1✔
408	return _resource_arn(stream_name, pattern, account_id=account_id, region_name=region_name)	1✔
409
410
411	#
412	# KMS
413	#
414
415
416	def kms_key_arn(key_id: str, account_id: str, region_name: str) -> str:	1✔
417	pattern = "arn:%s:kms:%s:%s:key/%s"	1✔
418	return _resource_arn(key_id, pattern, account_id=account_id, region_name=region_name)	1✔
419
420
421	def kms_alias_arn(alias_name: str, account_id: str, region_name: str):	1✔
422	if not alias_name.startswith("alias/"):	1✔
423	alias_name = "alias/" + alias_name	×
424	pattern = "arn:%s:kms:%s:%s:%s"	1✔
425	return _resource_arn(alias_name, pattern, account_id=account_id, region_name=region_name)	1✔
426
427
428	#
429	# SSM
430	#
431
432
433	def ssm_parameter_arn(param_name: str, account_id: str, region_name: str) -> str:	1✔
434	pattern = "arn:%s:ssm:%s:%s:parameter/%s"	×
435	param_name = param_name.lstrip("/")	×
436	return _resource_arn(param_name, pattern, account_id=account_id, region_name=region_name)	×
437
438
439	#
440	# S3
441	#
442
443
444	def s3_bucket_arn(bucket_name_or_arn: str, region="us-east-1") -> str:	1✔
445	bucket_name = s3_bucket_name(bucket_name_or_arn)	1✔
446	return f"arn:{get_partition(region)}:s3:::{bucket_name}"	1✔
447
448
449	#
450	# SQS
451	#
452
453
454	def sqs_queue_arn(queue_name: str, account_id: str, region_name: str) -> str:	1✔
455	queue_name = queue_name.split("/")[-1]	1✔
456	return "arn:%s:sqs:%s:%s:%s" % (get_partition(region_name), region_name, account_id, queue_name)	1✔
457
458
459	#
460	# APIGW
461	#
462
463
464	def apigateway_restapi_arn(api_id: str, account_id: str, region_name: str) -> str:	1✔
465	return "arn:%s:apigateway:%s:%s:/restapis/%s" % (	1✔
466	get_partition(region_name),
467	region_name,
468	account_id,
469	api_id,
470	)
471
472
473	def apigateway_invocations_arn(lambda_uri: str, region_name: str) -> str:	1✔
474	return "arn:%s:apigateway:%s:lambda:path/2015-03-31/functions/%s/invocations" % (	1✔
475	get_partition(region_name),
476	region_name,
477	lambda_uri,
478	)
479
480
481	#
482	# SNS
483	#
484
485
486	def sns_topic_arn(topic_name: str, account_id: str, region_name: str) -> str:	1✔
487	return f"arn:{get_partition(region_name)}:sns:{region_name}:{account_id}:{topic_name}"	1✔
488
489
490	#
491	# ECR
492	#
493
494
495	def ecr_repository_arn(name: str, account_id: str, region_name: str) -> str:	1✔
496	pattern = "arn:%s:ecr:%s:%s:repository/%s"	1✔
497	return _resource_arn(name, pattern, account_id=account_id, region_name=region_name)	1✔
498
499
500	#
501	# Route53
502	#
503
504
505	def route53_resolver_firewall_rule_group_arn(id: str, account_id: str, region_name: str) -> str:	1✔
506	pattern = "arn:%s:route53resolver:%s:%s:firewall-rule-group/%s"	1✔
507	return _resource_arn(id, pattern, account_id=account_id, region_name=region_name)	1✔
508
509
510	def route53_resolver_firewall_domain_list_arn(id: str, account_id: str, region_name: str) -> str:	1✔
511	pattern = "arn:%s:route53resolver:%s:%s:firewall-domain-list/%s"	1✔
512	return _resource_arn(id, pattern, account_id=account_id, region_name=region_name)	1✔
513
514
515	def route53_resolver_firewall_rule_group_associations_arn(	1✔
516	id: str, account_id: str, region_name: str
517	) -> str:
518	pattern = "arn:%s:route53resolver:%s:%s:firewall-rule-group-association/%s"	×
519	return _resource_arn(id, pattern, account_id=account_id, region_name=region_name)	×
520
521
522	def route53_resolver_query_log_config_arn(id: str, account_id: str, region_name: str) -> str:	1✔
523	pattern = "arn:%s:route53resolver:%s:%s:resolver-query-log-config/%s"	1✔
524	return _resource_arn(id, pattern, account_id=account_id, region_name=region_name)	1✔
525
526
527	#
528	# SES
529	#
530
531
532	def ses_identity_arn(email: str, account_id: str, region_name: str) -> str:	1✔
533	pattern = "arn:%s:ses:%s:%s:identity/%s"	×
534	return _resource_arn(email, pattern, account_id=account_id, region_name=region_name)	×
535
536
537	#
538	# Other ARN related helpers
539	#
540
541
542	def opensearch_domain_name(domain_arn: str) -> str:	1✔
543	return domain_arn.rpartition("/")[2]	×
544
545
546	def firehose_name(firehose_arn: str) -> str:	1✔
547	return firehose_arn.split("/")[-1]	1✔
548
549
550	def kinesis_stream_name(kinesis_arn: str) -> str:	1✔
551	return kinesis_arn.split(":stream/")[-1]	1✔
552
553
554	def lambda_function_name(name_or_arn: str) -> str:	1✔
555	if ":" in name_or_arn:	1✔
556	arn = parse_arn(name_or_arn)	1✔
557	if arn["service"] != "lambda":	1✔
558	raise ValueError("arn is not a lambda arn %s" % name_or_arn)	1✔
559
560	return parse_arn(name_or_arn)["resource"].split(":")[1]	1✔
561	else:
562	return name_or_arn	1✔
563
564
565	@cache	1✔
566	def sqs_queue_url_for_arn(queue_arn: str) -> str:	1✔
567	"""
568	Return the SQS queue URL for the given queue ARN.
569	"""
570	if "://" in queue_arn:	1✔
571	return queue_arn	×
572
573	try:	1✔
574	arn = parse_arn(queue_arn)	1✔
575	account_id = arn["account"]	1✔
576	region_name = arn["region"]	1✔
577	queue_name = arn["resource"]	1✔
578	except InvalidArnException:	×
579	account_id = DEFAULT_AWS_ACCOUNT_ID	×
580	region_name = None	×
581	queue_name = queue_arn	×
582
583	sqs_client = connect_to(region_name=region_name).sqs	1✔
584	result = sqs_client.get_queue_url(QueueName=queue_name, QueueOwnerAWSAccountId=account_id)[	1✔
585	"QueueUrl"
586	]
587	return result	1✔
588
589
590	def sqs_queue_name(queue_arn: str) -> str:	1✔
591	if ":" in queue_arn:	1✔
592	return parse_arn(queue_arn)["resource"]	1✔
593	else:
594	return queue_arn	×
595
596
597	def s3_bucket_name(bucket_name_or_arn: str) -> str:	1✔
598	return bucket_name_or_arn.split(":::")[-1]	1✔
599
600
601	def is_arn(possible_arn: str) -> bool:	1✔
602	try:	1✔
603	parse_arn(possible_arn)	1✔
604	return True	1✔
605	except InvalidArnException:	1✔
606	return False	1✔

localstack / localstack / ea4d3a2c-ec60-4c10-9aae-c0c88ca77069

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous