20565403496

Committed 29 Dec 2025 05:11AM UTC coverage: 84.103% (-2.8%) from 86.921%

Build # 20565403496

Build Type

Pull #13567

github

Committed by

web-flow

Commit Message

Merge 4816837a5 into 2417384aa

Pull Request Pull Request #13567: Update ASF APIs

Run Details

67166 of 79862 relevant lines covered (84.1%)

0.84 hits per line

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

34.78

/localstack-core/localstack/services/sns/v2/utils.py

import base64
import json
from uuid import uuid4

from botocore.utils import InvalidArnException

from localstack.aws.api.sns import InvalidParameterException
from localstack.services.sns.constants import E164_REGEX, VALID_SUBSCRIPTION_ATTR_NAME
from localstack.services.sns.v2.models import SnsStore, SnsSubscription
from localstack.utils.aws.arns import ArnData, parse_arn
from localstack.utils.strings import short_uid, to_bytes, to_str


def parse_and_validate_topic_arn(topic_arn: str | None) -> ArnData:
    return _parse_and_validate_arn(topic_arn, "Topic")


def parse_and_validate_platform_application_arn(platform_application_arn: str | None) -> ArnData:
    return _parse_and_validate_arn(platform_application_arn, "PlatformApplication")


def _parse_and_validate_arn(arn: str | None, resource_type: str) -> ArnData:
    arn = arn or ""
    try:
        return parse_arn(arn)
    except InvalidArnException:
        count = len(arn.split(":"))
        raise InvalidParameterException(
            f"Invalid parameter: {resource_type}Arn Reason: An ARN must have at least 6 elements, not {count}"
        )


def is_valid_e164_number(number: str) -> bool:
    return E164_REGEX.match(number) is not None


def validate_subscription_attribute(
    attribute_name: str,
    attribute_value: str,
    topic_arn: str,
    endpoint: str,
    is_subscribe_call: bool = False,
) -> None:
    """
    Validate the subscription attribute to be set. See:
    https://docs.aws.amazon.com/sns/latest/api/API_SetSubscriptionAttributes.html
    :param attribute_name: the subscription attribute name, must be in VALID_SUBSCRIPTION_ATTR_NAME
    :param attribute_value: the subscription attribute value
    :param topic_arn: the topic_arn of the subscription, needed to know if it is FIFO
    :param endpoint: the subscription endpoint (like an SQS queue ARN)
    :param is_subscribe_call: the error message is different if called from Subscribe or SetSubscriptionAttributes
    :raises InvalidParameterException
    :return:
    """
    error_prefix = (
        "Invalid parameter: Attributes Reason: " if is_subscribe_call else "Invalid parameter: "
    )
    if attribute_name not in VALID_SUBSCRIPTION_ATTR_NAME:
        raise InvalidParameterException(f"{error_prefix}AttributeName")

    if attribute_name == "FilterPolicy":
        try:
            json.loads(attribute_value or "{}")
        except json.JSONDecodeError:
            raise InvalidParameterException(f"{error_prefix}FilterPolicy: failed to parse JSON.")
    elif attribute_name == "FilterPolicyScope":
        if attribute_value not in ("MessageAttributes", "MessageBody"):
            raise InvalidParameterException(
                f"{error_prefix}FilterPolicyScope: Invalid value [{attribute_value}]. "
                f"Please use either MessageBody or MessageAttributes"
            )
    elif attribute_name == "RawMessageDelivery":
        # TODO: only for SQS and https(s) subs, + firehose
        if attribute_value.lower() not in ("true", "false"):
            raise InvalidParameterException(
                f"{error_prefix}RawMessageDelivery: Invalid value [{attribute_value}]. "
                f"Must be true or false."
            )

    elif attribute_name == "RedrivePolicy":
        try:
            dlq_target_arn = json.loads(attribute_value).get("deadLetterTargetArn", "")
        except json.JSONDecodeError:
            raise InvalidParameterException(f"{error_prefix}RedrivePolicy: failed to parse JSON.")
        try:
            parsed_arn = parse_arn(dlq_target_arn)
        except InvalidArnException:
            raise InvalidParameterException(
                f"{error_prefix}RedrivePolicy: deadLetterTargetArn is an invalid arn"
            )

        if topic_arn.endswith(".fifo"):
            if endpoint.endswith(".fifo") and (
                not parsed_arn["resource"].endswith(".fifo") or "sqs" not in parsed_arn["service"]
            ):
                raise InvalidParameterException(
                    f"{error_prefix}RedrivePolicy: must use a FIFO queue as DLQ for a FIFO Subscription to a FIFO Topic."
                )


def create_subscription_arn(topic_arn: str) -> str:
    # This is the format of a Subscription ARN
    # arn:aws:sns:us-west-2:123456789012:my-topic:8a21d249-4329-4871-acc6-7be709c6ea7f
    return f"{topic_arn}:{uuid4()}"


def create_platform_endpoint_arn(
    platform_application_arn: str,
) -> str:
    # This is the format of an Endpoint Arn
    # arn:aws:sns:us-west-2:1234567890:endpoint/GCM/MyApplication/12345678-abcd-9012-efgh-345678901234
    return f"{platform_application_arn.replace('app', 'endpoint', 1)}/{uuid4()}"


def encode_subscription_token_with_region(region: str) -> str:
    """
    Create a 64 characters Subscription Token with the region encoded
    :param region:
    :return: a subscription token with the region encoded
    """
    return ((region.encode() + b"/").hex() + short_uid() * 8)[:64]


def get_next_page_token_from_arn(resource_arn: str) -> str:
    return to_str(base64.b64encode(to_bytes(resource_arn)))


def get_region_from_subscription_token(token: str) -> str:
    """
    Try to decode and return the region from a subscription token
    :param token:
    :return: the region if able to decode it
    :raises: InvalidParameterException if the token is invalid
    """
    try:
        region = token.split("2f", maxsplit=1)[0]
        return bytes.fromhex(region).decode("utf-8")
    except (IndexError, ValueError, TypeError, UnicodeDecodeError):
        raise InvalidParameterException("Invalid parameter: Token")


def get_topic_subscriptions(store: SnsStore, topic_arn: str) -> list[SnsSubscription]:
    # TODO: delete this once the legacy v1 implementation has been removed
    if hasattr(store, "topic_subscriptions"):
        sub_arns = store.topic_subscriptions.get(topic_arn, [])
    else:
        sub_arns: list[str] = store.topics[topic_arn].get("subscriptions", [])

    subscriptions = [store.subscriptions[k] for k in sub_arns if k in store.subscriptions]
    return subscriptions

1	import base64	1✔
2	import json	1✔
3	from uuid import uuid4	1✔
4
5	from botocore.utils import InvalidArnException	1✔
6
7	from localstack.aws.api.sns import InvalidParameterException	1✔
8	from localstack.services.sns.constants import E164_REGEX, VALID_SUBSCRIPTION_ATTR_NAME	1✔
9	from localstack.services.sns.v2.models import SnsStore, SnsSubscription	1✔
10	from localstack.utils.aws.arns import ArnData, parse_arn	1✔
11	from localstack.utils.strings import short_uid, to_bytes, to_str	1✔
12
13
14	def parse_and_validate_topic_arn(topic_arn: str \| None) -> ArnData:	1✔
15	return _parse_and_validate_arn(topic_arn, "Topic")	×
16
17
18	def parse_and_validate_platform_application_arn(platform_application_arn: str \| None) -> ArnData:	1✔
19	return _parse_and_validate_arn(platform_application_arn, "PlatformApplication")	×
20
21
22	def _parse_and_validate_arn(arn: str \| None, resource_type: str) -> ArnData:	1✔
23	arn = arn or ""	×
24	try:	×
25	return parse_arn(arn)	×
26	except InvalidArnException:	×
27	count = len(arn.split(":"))	×
28	raise InvalidParameterException(	×
29	f"Invalid parameter: {resource_type}Arn Reason: An ARN must have at least 6 elements, not {count}"
30	)
31
32
33	def is_valid_e164_number(number: str) -> bool:	1✔
34	return E164_REGEX.match(number) is not None	×
35
36
37	def validate_subscription_attribute(	1✔
38	attribute_name: str,
39	attribute_value: str,
40	topic_arn: str,
41	endpoint: str,
42	is_subscribe_call: bool = False,
43	) -> None:
44	"""
45	Validate the subscription attribute to be set. See:
46	https://docs.aws.amazon.com/sns/latest/api/API_SetSubscriptionAttributes.html
47	:param attribute_name: the subscription attribute name, must be in VALID_SUBSCRIPTION_ATTR_NAME
48	:param attribute_value: the subscription attribute value
49	:param topic_arn: the topic_arn of the subscription, needed to know if it is FIFO
50	:param endpoint: the subscription endpoint (like an SQS queue ARN)
51	:param is_subscribe_call: the error message is different if called from Subscribe or SetSubscriptionAttributes
52	:raises InvalidParameterException
53	:return:
54	"""
55	error_prefix = (	×
56	"Invalid parameter: Attributes Reason: " if is_subscribe_call else "Invalid parameter: "
57	)
58	if attribute_name not in VALID_SUBSCRIPTION_ATTR_NAME:	×
59	raise InvalidParameterException(f"{error_prefix}AttributeName")	×
60
61	if attribute_name == "FilterPolicy":	×
62	try:	×
63	json.loads(attribute_value or "{}")	×
64	except json.JSONDecodeError:	×
65	raise InvalidParameterException(f"{error_prefix}FilterPolicy: failed to parse JSON.")	×
66	elif attribute_name == "FilterPolicyScope":	×
67	if attribute_value not in ("MessageAttributes", "MessageBody"):	×
68	raise InvalidParameterException(	×
69	f"{error_prefix}FilterPolicyScope: Invalid value [{attribute_value}]. "
70	f"Please use either MessageBody or MessageAttributes"
71	)
72	elif attribute_name == "RawMessageDelivery":	×
73	# TODO: only for SQS and https(s) subs, + firehose
74	if attribute_value.lower() not in ("true", "false"):	×
75	raise InvalidParameterException(	×
76	f"{error_prefix}RawMessageDelivery: Invalid value [{attribute_value}]. "
77	f"Must be true or false."
78	)
79
80	elif attribute_name == "RedrivePolicy":	×
81	try:	×
82	dlq_target_arn = json.loads(attribute_value).get("deadLetterTargetArn", "")	×
83	except json.JSONDecodeError:	×
84	raise InvalidParameterException(f"{error_prefix}RedrivePolicy: failed to parse JSON.")	×
85	try:	×
86	parsed_arn = parse_arn(dlq_target_arn)	×
87	except InvalidArnException:	×
88	raise InvalidParameterException(	×
89	f"{error_prefix}RedrivePolicy: deadLetterTargetArn is an invalid arn"
90	)
91
92	if topic_arn.endswith(".fifo"):	×
93	if endpoint.endswith(".fifo") and (	×
94	not parsed_arn["resource"].endswith(".fifo") or "sqs" not in parsed_arn["service"]
95	):
96	raise InvalidParameterException(	×
97	f"{error_prefix}RedrivePolicy: must use a FIFO queue as DLQ for a FIFO Subscription to a FIFO Topic."
98	)
99
100
101	def create_subscription_arn(topic_arn: str) -> str:	1✔
102	# This is the format of a Subscription ARN
103	# arn:aws:sns:us-west-2:123456789012:my-topic:8a21d249-4329-4871-acc6-7be709c6ea7f
104	return f"{topic_arn}:{uuid4()}"	×
105
106
107	def create_platform_endpoint_arn(	1✔
108	platform_application_arn: str,
109	) -> str:
110	# This is the format of an Endpoint Arn
111	# arn:aws:sns:us-west-2:1234567890:endpoint/GCM/MyApplication/12345678-abcd-9012-efgh-345678901234
112	return f"{platform_application_arn.replace('app', 'endpoint', 1)}/{uuid4()}"	×
113
114
115	def encode_subscription_token_with_region(region: str) -> str:	1✔
116	"""
117	Create a 64 characters Subscription Token with the region encoded
118	:param region:
119	:return: a subscription token with the region encoded
120	"""
121	return ((region.encode() + b"/").hex() + short_uid() * 8)[:64]	×
122
123
124	def get_next_page_token_from_arn(resource_arn: str) -> str:	1✔
125	return to_str(base64.b64encode(to_bytes(resource_arn)))	×
126
127
128	def get_region_from_subscription_token(token: str) -> str:	1✔
129	"""
130	Try to decode and return the region from a subscription token
131	:param token:
132	:return: the region if able to decode it
133	:raises: InvalidParameterException if the token is invalid
134	"""
135	try:	×
136	region = token.split("2f", maxsplit=1)[0]	×
137	return bytes.fromhex(region).decode("utf-8")	×
138	except (IndexError, ValueError, TypeError, UnicodeDecodeError):	×
139	raise InvalidParameterException("Invalid parameter: Token")	×
140
141
142	def get_topic_subscriptions(store: SnsStore, topic_arn: str) -> list[SnsSubscription]:	1✔
143	# TODO: delete this once the legacy v1 implementation has been removed
144	if hasattr(store, "topic_subscriptions"):	1✔
145	sub_arns = store.topic_subscriptions.get(topic_arn, [])	1✔
146	else:
147	sub_arns: list[str] = store.topics[topic_arn].get("subscriptions", [])	×
148
149	subscriptions = [store.subscriptions[k] for k in sub_arns if k in store.subscriptions]	1✔
150	return subscriptions	1✔

localstack / localstack / 20565403496

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