18083953951

Committed 26 Sep 2025 07:54PM UTC coverage: 86.89% (+0.01%) from 86.88%

Build # 18083953951

Build Type

push

github

Committed by

web-flow

Commit Message

fix sqs dev endpoint to show invisible fifo messages correctly (#13196)

Coverage Stats

5 of 5 new or added lines in 1 file covered. (100.0%)

59 existing lines in 3 files now uncovered.

67757 of 77980 relevant lines covered (86.89%)

0.87 hits per line

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

71.19

/localstack-core/localstack/services/cloudformation/engine/v2/resolving.py

import json
import logging
import re
from dataclasses import dataclass
from typing import Any

from botocore.exceptions import ClientError

from localstack.aws.connect import connect_to

LOG = logging.getLogger(__name__)

# CloudFormation allows using dynamic references in `Fn::Sub` expressions, so we must make sure
# we don't capture the parameter usage by excluding ${} characters
REGEX_DYNAMIC_REF = re.compile(r"{{resolve:([^:]+):([^${}]+)}}")


@dataclass
class DynamicReference:
    service_name: str
    reference_key: str


def extract_dynamic_reference(value: Any) -> DynamicReference | None:
    if isinstance(value, str):
        if dynamic_ref_match := REGEX_DYNAMIC_REF.search(value):
            return DynamicReference(dynamic_ref_match[1], dynamic_ref_match[2])
    return None


def perform_dynamic_reference_lookup(
    reference: DynamicReference, account_id: str, region_name: str
) -> str | None:
    # basic dynamic reference support
    # see: https://docs.aws.amazon.com/AWSCloudFormation/latest/UserGuide/dynamic-references.html
    # technically there are more restrictions for each of these services but checking each of these
    # isn't really necessary for the current level of emulation

    # only these 3 services are supported for dynamic references right now
    if reference.service_name == "ssm":
        ssm_client = connect_to(aws_access_key_id=account_id, region_name=region_name).ssm
        try:
            return ssm_client.get_parameter(Name=reference.reference_key)["Parameter"]["Value"]
        except ClientError as e:
            LOG.error("client error accessing SSM parameter '%s': %s", reference.reference_key, e)
            raise
    elif reference.service_name == "ssm-secure":
        ssm_client = connect_to(aws_access_key_id=account_id, region_name=region_name).ssm
        try:
            return ssm_client.get_parameter(Name=reference.reference_key, WithDecryption=True)[
                "Parameter"
            ]["Value"]
        except ClientError as e:
            LOG.error("client error accessing SSM parameter '%s': %s", reference.reference_key, e)
            raise
    elif reference.service_name == "secretsmanager":
        # reference key needs to be parsed further
        # because {{resolve:secretsmanager:secret-id:secret-string:json-key:version-stage:version-id}}
        # we match for "secret-id:secret-string:json-key:version-stage:version-id"
        # where
        #   secret-id can either be the secret name or the full ARN of the secret
        #   secret-string *must* be SecretString
        #   all other values are optional
        secret_id = reference.reference_key
        [json_key, version_stage, version_id] = [None, None, None]
        if "SecretString" in reference.reference_key:
            parts = reference.reference_key.split(":SecretString:")
            secret_id = parts[0]
            # json-key, version-stage and version-id are optional.
            [json_key, version_stage, version_id] = f"{parts[1]}::".split(":")[:3]

        kwargs = {}  # optional args for get_secret_value
        if version_id:
            kwargs["VersionId"] = version_id
        if version_stage:
            kwargs["VersionStage"] = version_stage

        secretsmanager_client = connect_to(
            aws_access_key_id=account_id, region_name=region_name
        ).secretsmanager
        try:
            secret_value = secretsmanager_client.get_secret_value(SecretId=secret_id, **kwargs)[
                "SecretString"
            ]
        except ClientError:
            LOG.error("client error while trying to access key '%s': %s", secret_id)
            raise

        if json_key:
            json_secret = json.loads(secret_value)
            if json_key not in json_secret:
                raise RuntimeError(
                    f"JSON value for {reference.service_name}.{reference.reference_key} not present"
                )
            return str(json_secret[json_key])
        else:
            return str(secret_value)

    LOG.warning(
        "Unsupported service for dynamic parameter: service_name=%s", reference.service_name
    )
    return None

1	import json	1✔
2	import logging	1✔
3	import re	1✔
4	from dataclasses import dataclass	1✔
5	from typing import Any	1✔
6
7	from botocore.exceptions import ClientError	1✔
8
9	from localstack.aws.connect import connect_to	1✔
10
11	LOG = logging.getLogger(__name__)	1✔
12
13	# CloudFormation allows using dynamic references in `Fn::Sub` expressions, so we must make sure
14	# we don't capture the parameter usage by excluding ${} characters
15	REGEX_DYNAMIC_REF = re.compile(r"{{resolve:([^:]+):([^${}]+)}}")	1✔
16
17
18	@dataclass	1✔
19	class DynamicReference:	1✔
20	service_name: str	1✔
21	reference_key: str	1✔
22
23
24	def extract_dynamic_reference(value: Any) -> DynamicReference \| None:	1✔
25	if isinstance(value, str):	1✔
26	if dynamic_ref_match := REGEX_DYNAMIC_REF.search(value):	1✔
27	return DynamicReference(dynamic_ref_match[1], dynamic_ref_match[2])	1✔
28	return None	1✔
29
30
31	def perform_dynamic_reference_lookup(	1✔
32	reference: DynamicReference, account_id: str, region_name: str
33	) -> str \| None:
34	# basic dynamic reference support
35	# see: https://docs.aws.amazon.com/AWSCloudFormation/latest/UserGuide/dynamic-references.html
36	# technically there are more restrictions for each of these services but checking each of these
37	# isn't really necessary for the current level of emulation
38
39	# only these 3 services are supported for dynamic references right now
40	if reference.service_name == "ssm":	1✔
41	ssm_client = connect_to(aws_access_key_id=account_id, region_name=region_name).ssm	1✔
42	try:	1✔
43	return ssm_client.get_parameter(Name=reference.reference_key)["Parameter"]["Value"]	1✔
44	except ClientError as e:	×
UNCOV 45	LOG.error("client error accessing SSM parameter '%s': %s", reference.reference_key, e)	×
UNCOV 46	raise	×
47	elif reference.service_name == "ssm-secure":	1✔
48	ssm_client = connect_to(aws_access_key_id=account_id, region_name=region_name).ssm	1✔
49	try:	1✔
50	return ssm_client.get_parameter(Name=reference.reference_key, WithDecryption=True)[	1✔
51	"Parameter"
52	]["Value"]
53	except ClientError as e:	×
UNCOV 54	LOG.error("client error accessing SSM parameter '%s': %s", reference.reference_key, e)	×
UNCOV 55	raise	×
56	elif reference.service_name == "secretsmanager":	1✔
57	# reference key needs to be parsed further
58	# because {{resolve:secretsmanager:secret-id:secret-string:json-key:version-stage:version-id}}
59	# we match for "secret-id:secret-string:json-key:version-stage:version-id"
60	# where
61	# secret-id can either be the secret name or the full ARN of the secret
62	# secret-string must be SecretString
63	# all other values are optional
64	secret_id = reference.reference_key	1✔
65	[json_key, version_stage, version_id] = [None, None, None]	1✔
66	if "SecretString" in reference.reference_key:	1✔
67	parts = reference.reference_key.split(":SecretString:")	1✔
68	secret_id = parts[0]	1✔
69	# json-key, version-stage and version-id are optional.
70	[json_key, version_stage, version_id] = f"{parts[1]}::".split(":")[:3]	1✔
71
72	kwargs = {} # optional args for get_secret_value	1✔
73	if version_id:	1✔
74	kwargs["VersionId"] = version_id	×
75	if version_stage:	1✔
UNCOV 76	kwargs["VersionStage"] = version_stage	×
77
78	secretsmanager_client = connect_to(	1✔
79	aws_access_key_id=account_id, region_name=region_name
80	).secretsmanager
81	try:	1✔
82	secret_value = secretsmanager_client.get_secret_value(SecretId=secret_id, **kwargs)[	1✔
83	"SecretString"
84	]
85	except ClientError:	×
UNCOV 86	LOG.error("client error while trying to access key '%s': %s", secret_id)	×
UNCOV 87	raise	×
88
89	if json_key:	1✔
90	json_secret = json.loads(secret_value)	×
UNCOV 91	if json_key not in json_secret:	×
UNCOV 92	raise RuntimeError(	×
93	f"JSON value for {reference.service_name}.{reference.reference_key} not present"
94	)
UNCOV 95	return str(json_secret[json_key])	×
96	else:
97	return str(secret_value)	1✔
98
UNCOV 99	LOG.warning(	×
100	"Unsupported service for dynamic parameter: service_name=%s", reference.service_name
101	)
UNCOV 102	return None	×

localstack / localstack / 18083953951

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