22709357475

Committed 05 Mar 2026 08:35AM UTC coverage: 59.732% (-27.2%) from 86.974%

Build # 22709357475

Build Type

Pull #13880

github

Committed by

web-flow

Commit Message

Merge 28fcab93c into 710618057

Pull Request Pull Request #13880: Firehose: Replace TaggingService

Coverage Stats

12 of 12 new or added lines in 2 files covered. (100.0%)

20464 existing lines in 510 files now uncovered.

45290 of 75822 relevant lines covered (59.73%)

0.6 hits per line

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

30.51

/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✔
UNCOV 27	return DynamicReference(dynamic_ref_match[1], dynamic_ref_match[2])	×
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
UNCOV 40	if reference.service_name == "ssm":	×
UNCOV 41	ssm_client = connect_to(aws_access_key_id=account_id, region_name=region_name).ssm	×
UNCOV 42	try:	×
UNCOV 43	return ssm_client.get_parameter(Name=reference.reference_key)["Parameter"]["Value"]	×
44	except ClientError as e:	×
45	LOG.error("client error accessing SSM parameter '%s': %s", reference.reference_key, e)	×
46	raise	×
UNCOV 47	elif reference.service_name == "ssm-secure":	×
UNCOV 48	ssm_client = connect_to(aws_access_key_id=account_id, region_name=region_name).ssm	×
UNCOV 49	try:	×
UNCOV 50	return ssm_client.get_parameter(Name=reference.reference_key, WithDecryption=True)[	×
51	"Parameter"
52	]["Value"]
53	except ClientError as e:	×
54	LOG.error("client error accessing SSM parameter '%s': %s", reference.reference_key, e)	×
55	raise	×
UNCOV 56	elif reference.service_name == "secretsmanager":	×
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
UNCOV 64	secret_id = reference.reference_key	×
UNCOV 65	[json_key, version_stage, version_id] = [None, None, None]	×
UNCOV 66	if "SecretString" in reference.reference_key:	×
UNCOV 67	parts = reference.reference_key.split(":SecretString:")	×
UNCOV 68	secret_id = parts[0]	×
69	# json-key, version-stage and version-id are optional.
UNCOV 70	[json_key, version_stage, version_id] = f"{parts[1]}::".split(":")[:3]	×
71
UNCOV 72	kwargs = {} # optional args for get_secret_value	×
UNCOV 73	if version_id:	×
74	kwargs["VersionId"] = version_id	×
UNCOV 75	if version_stage:	×
76	kwargs["VersionStage"] = version_stage	×
77
UNCOV 78	secretsmanager_client = connect_to(	×
79	aws_access_key_id=account_id, region_name=region_name
80	).secretsmanager
UNCOV 81	try:	×
UNCOV 82	secret_value = secretsmanager_client.get_secret_value(SecretId=secret_id, **kwargs)[	×
83	"SecretString"
84	]
85	except ClientError:	×
86	LOG.error("client error while trying to access key '%s': %s", secret_id)	×
87	raise	×
88
UNCOV 89	if json_key:	×
90	json_secret = json.loads(secret_value)	×
91	if json_key not in json_secret:	×
92	raise RuntimeError(	×
93	f"JSON value for {reference.service_name}.{reference.reference_key} not present"
94	)
95	return str(json_secret[json_key])	×
96	else:
UNCOV 97	return str(secret_value)	×
98
99	LOG.warning(	×
100	"Unsupported service for dynamic parameter: service_name=%s", reference.service_name
101	)
102	return None	×

localstack / localstack / 22709357475

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