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

40.79

/localstack-core/localstack/services/cloudformation/engine/parameters.py

"""
TODO: ordering & grouping of parameters
TODO: design proper structure for parameters to facilitate validation etc.
TODO: clearer language around both parameters and "resolving"

Documentation extracted from AWS docs (https://docs.aws.amazon.com/AWSCloudFormation/latest/UserGuide/parameters-section-structure.html):
    The following requirements apply when using parameters:

        You can have a maximum of 200 parameters in an AWS CloudFormation template.
        Each parameter must be given a logical name (also called logical ID), which must be alphanumeric and unique among all logical names within the template.
        Each parameter must be assigned a parameter type that is supported by AWS CloudFormation. For more information, see Type.
        Each parameter must be assigned a value at runtime for AWS CloudFormation to successfully provision the stack. You can optionally specify a default value for AWS CloudFormation to use unless another value is provided.
        Parameters must be declared and referenced from within the same template. You can reference parameters from the Resources and Outputs sections of the template.

        When you create or update stacks and create change sets, AWS CloudFormation uses whatever values exist in Parameter Store at the time the operation is run. If a specified parameter doesn't exist in Parameter Store under the caller's AWS account, AWS CloudFormation returns a validation error.

        For stack updates, the Use existing value option in the console and the UsePreviousValue attribute for update-stack tell AWS CloudFormation to use the existing Systems Manager parameter key—not its value. AWS CloudFormation always fetches the latest values from Parameter Store when it updates stacks.

"""

import logging
from typing import Literal, TypedDict

from botocore.exceptions import ClientError

from localstack.aws.api.cloudformation import Parameter, ParameterDeclaration
from localstack.aws.connect import connect_to

LOG = logging.getLogger(__name__)


def extract_stack_parameter_declarations(template: dict) -> dict[str, ParameterDeclaration]:
    """
    Extract and build a dict of stack parameter declarations from a CloudFormation stack templatef

    :param template: the parsed CloudFormation stack template
    :return: a dictionary of declared parameters, mapping logical IDs to the corresponding parameter declaration
    """
    result = {}
    for param_key, param in template.get("Parameters", {}).items():
        result[param_key] = ParameterDeclaration(
            ParameterKey=param_key,
            DefaultValue=param.get("Default"),
            ParameterType=param.get("Type"),
            NoEcho=param.get("NoEcho", False),
            # TODO: test & implement rest here
            # ParameterConstraints=?,
            # Description=?
        )
    return result


class StackParameter(Parameter):
    # we need the type information downstream when actually using the resolved value
    # e.g. in case of lists so that we know that we should interpret the string as a comma-separated list.
    ParameterType: str


def resolve_parameters(
    account_id: str,
    region_name: str,
    parameter_declarations: dict[str, ParameterDeclaration],
    new_parameters: dict[str, Parameter],
    old_parameters: dict[str, Parameter],
) -> dict[str, StackParameter]:
    """
    Resolves stack parameters or raises an exception if any parameter can not be resolved.

    Assumptions:
        - There are no extra undeclared parameters given (validate before calling this method)

    TODO: is UsePreviousValue=False equivalent to not specifying it, in all situations?

    :param parameter_declarations: The parameter declaration from the (potentially new) template, i.e. the "Parameters" section
    :param new_parameters: The parameters to resolve
    :param old_parameters: The old parameters from the previous stack deployment, if available
    :return: a copy of new_parameters with resolved values
    """
    resolved_parameters = {}

    # populate values for every parameter declared in the template
    for pm in parameter_declarations.values():
        pm_key = pm["ParameterKey"]
        resolved_param = StackParameter(ParameterKey=pm_key, ParameterType=pm["ParameterType"])
        new_parameter = new_parameters.get(pm_key)
        old_parameter = old_parameters.get(pm_key)

        if new_parameter is None:
            # since no value has been specified for the deployment, we need to be able to resolve the default or fail
            default_value = pm["DefaultValue"]
            if default_value is None:
                LOG.error("New parameter without a default value: %s", pm_key)
                raise Exception(
                    f"Invalid. Parameter '{pm_key}' needs to have either param specified or Default."
                )  # TODO: test and verify

            resolved_param["ParameterValue"] = default_value
        else:
            if (
                new_parameter.get("UsePreviousValue", False)
                and new_parameter.get("ParameterValue") is not None
            ):
                raise Exception(
                    f"Can't set both 'UsePreviousValue' and a concrete value for parameter '{pm_key}'."
                )  # TODO: test and verify

            if new_parameter.get("UsePreviousValue", False):
                if old_parameter is None:
                    raise Exception(
                        f"Set 'UsePreviousValue' but stack has no previous value for parameter '{pm_key}'."
                    )  # TODO: test and verify

                resolved_param["ParameterValue"] = old_parameter["ParameterValue"]
            else:
                resolved_param["ParameterValue"] = new_parameter["ParameterValue"]

        resolved_param["NoEcho"] = pm.get("NoEcho", False)
        resolved_parameters[pm_key] = resolved_param

        # Note that SSM parameters always need to be resolved anew here
        # TODO: support more parameter types
        if pm["ParameterType"].startswith("AWS::SSM"):
            if pm["ParameterType"] in [
                "AWS::SSM::Parameter::Value<String>",
                "AWS::SSM::Parameter::Value<AWS::EC2::Image::Id>",
                "AWS::SSM::Parameter::Value<CommaDelimitedList>",
            ]:
                # TODO: error handling (e.g. no permission to lookup SSM parameter or SSM parameter doesn't exist)
                resolved_param["ResolvedValue"] = resolve_ssm_parameter(
                    account_id, region_name, resolved_param["ParameterValue"]
                )
            else:
                raise Exception(f"Unsupported stack parameter type: {pm['ParameterType']}")

    return resolved_parameters


# TODO: inject credentials / client factory for proper account/region lookup
def resolve_ssm_parameter(account_id: str, region_name: str, stack_parameter_value: str) -> str:
    """
    Resolve the SSM stack parameter from the SSM service with a name equal to the stack parameter value.
    """
    ssm_client = connect_to(aws_access_key_id=account_id, region_name=region_name).ssm
    try:
        return ssm_client.get_parameter(Name=stack_parameter_value)["Parameter"]["Value"]
    except ClientError:
        LOG.error("client error fetching parameter '%s'", stack_parameter_value)
        raise


def strip_parameter_type(in_param: StackParameter) -> Parameter:
    result = in_param.copy()
    result.pop("ParameterType", None)
    return result


def mask_no_echo(in_param: StackParameter) -> Parameter:
    result = in_param.copy()
    no_echo = result.pop("NoEcho", False)
    if no_echo:
        result["ParameterValue"] = "****"
    return result


def convert_stack_parameters_to_list(
    in_params: dict[str, StackParameter] | None,
) -> list[StackParameter]:
    if not in_params:
        return []
    return list(in_params.values())


def convert_stack_parameters_to_dict(in_params: list[Parameter] | None) -> dict[str, Parameter]:
    if not in_params:
        return {}
    return {p["ParameterKey"]: p for p in in_params}


class LegacyParameterProperties(TypedDict):
    Value: str
    ParameterType: str
    ParameterValue: str | None
    ResolvedValue: str | None


class LegacyParameter(TypedDict):
    LogicalResourceId: str
    Type: Literal["Parameter"]
    Properties: LegacyParameterProperties


# TODO: not actually parameter_type but the logical "ID"
def map_to_legacy_structure(parameter_name: str, new_parameter: StackParameter) -> LegacyParameter:
    """
    Helper util to convert a normal (resolved) stack parameter to a legacy parameter structure that can then be merged with stack resources.

    :param new_parameter: a resolved stack parameter
    :return: legacy parameter that can be merged with stack resources for uniform lookup based on logical ID
    """
    return LegacyParameter(
        LogicalResourceId=new_parameter["ParameterKey"],
        Type="Parameter",
        Properties=LegacyParameterProperties(
            ParameterType=new_parameter.get("ParameterType"),
            ParameterValue=new_parameter.get("ParameterValue"),
            ResolvedValue=new_parameter.get("ResolvedValue"),
            Value=new_parameter.get("ResolvedValue", new_parameter.get("ParameterValue")),
        ),
    )

1	"""
2	TODO: ordering & grouping of parameters
3	TODO: design proper structure for parameters to facilitate validation etc.
4	TODO: clearer language around both parameters and "resolving"
5
6	Documentation extracted from AWS docs (https://docs.aws.amazon.com/AWSCloudFormation/latest/UserGuide/parameters-section-structure.html):
7	The following requirements apply when using parameters:
8
9	You can have a maximum of 200 parameters in an AWS CloudFormation template.
10	Each parameter must be given a logical name (also called logical ID), which must be alphanumeric and unique among all logical names within the template.
11	Each parameter must be assigned a parameter type that is supported by AWS CloudFormation. For more information, see Type.
12	Each parameter must be assigned a value at runtime for AWS CloudFormation to successfully provision the stack. You can optionally specify a default value for AWS CloudFormation to use unless another value is provided.
13	Parameters must be declared and referenced from within the same template. You can reference parameters from the Resources and Outputs sections of the template.
14
15	When you create or update stacks and create change sets, AWS CloudFormation uses whatever values exist in Parameter Store at the time the operation is run. If a specified parameter doesn't exist in Parameter Store under the caller's AWS account, AWS CloudFormation returns a validation error.
16
17	For stack updates, the Use existing value option in the console and the UsePreviousValue attribute for update-stack tell AWS CloudFormation to use the existing Systems Manager parameter key—not its value. AWS CloudFormation always fetches the latest values from Parameter Store when it updates stacks.
18
19	"""
20
21	import logging	1✔
22	from typing import Literal, TypedDict	1✔
23
24	from botocore.exceptions import ClientError	1✔
25
26	from localstack.aws.api.cloudformation import Parameter, ParameterDeclaration	1✔
27	from localstack.aws.connect import connect_to	1✔
28
29	LOG = logging.getLogger(__name__)	1✔
30
31
32	def extract_stack_parameter_declarations(template: dict) -> dict[str, ParameterDeclaration]:	1✔
33	"""
34	Extract and build a dict of stack parameter declarations from a CloudFormation stack templatef
35
36	:param template: the parsed CloudFormation stack template
37	:return: a dictionary of declared parameters, mapping logical IDs to the corresponding parameter declaration
38	"""
39	result = {}	×
40	for param_key, param in template.get("Parameters", {}).items():	×
41	result[param_key] = ParameterDeclaration(	×
42	ParameterKey=param_key,
43	DefaultValue=param.get("Default"),
44	ParameterType=param.get("Type"),
45	NoEcho=param.get("NoEcho", False),
46	# TODO: test & implement rest here
47	# ParameterConstraints=?,
48	# Description=?
49	)
50	return result	×
51
52
53	class StackParameter(Parameter):	1✔
54	# we need the type information downstream when actually using the resolved value
55	# e.g. in case of lists so that we know that we should interpret the string as a comma-separated list.
56	ParameterType: str	1✔
57
58
59	def resolve_parameters(	1✔
60	account_id: str,
61	region_name: str,
62	parameter_declarations: dict[str, ParameterDeclaration],
63	new_parameters: dict[str, Parameter],
64	old_parameters: dict[str, Parameter],
65	) -> dict[str, StackParameter]:
66	"""
67	Resolves stack parameters or raises an exception if any parameter can not be resolved.
68
69	Assumptions:
70	- There are no extra undeclared parameters given (validate before calling this method)
71
72	TODO: is UsePreviousValue=False equivalent to not specifying it, in all situations?
73
74	:param parameter_declarations: The parameter declaration from the (potentially new) template, i.e. the "Parameters" section
75	:param new_parameters: The parameters to resolve
76	:param old_parameters: The old parameters from the previous stack deployment, if available
77	:return: a copy of new_parameters with resolved values
78	"""
79	resolved_parameters = {}	×
80
81	# populate values for every parameter declared in the template
82	for pm in parameter_declarations.values():	×
83	pm_key = pm["ParameterKey"]	×
84	resolved_param = StackParameter(ParameterKey=pm_key, ParameterType=pm["ParameterType"])	×
85	new_parameter = new_parameters.get(pm_key)	×
86	old_parameter = old_parameters.get(pm_key)	×
87
88	if new_parameter is None:	×
89	# since no value has been specified for the deployment, we need to be able to resolve the default or fail
90	default_value = pm["DefaultValue"]	×
91	if default_value is None:	×
92	LOG.error("New parameter without a default value: %s", pm_key)	×
93	raise Exception(	×
94	f"Invalid. Parameter '{pm_key}' needs to have either param specified or Default."
95	) # TODO: test and verify
96
97	resolved_param["ParameterValue"] = default_value	×
98	else:
99	if (	×
100	new_parameter.get("UsePreviousValue", False)
101	and new_parameter.get("ParameterValue") is not None
102	):
103	raise Exception(	×
104	f"Can't set both 'UsePreviousValue' and a concrete value for parameter '{pm_key}'."
105	) # TODO: test and verify
106
107	if new_parameter.get("UsePreviousValue", False):	×
108	if old_parameter is None:	×
109	raise Exception(	×
110	f"Set 'UsePreviousValue' but stack has no previous value for parameter '{pm_key}'."
111	) # TODO: test and verify
112
113	resolved_param["ParameterValue"] = old_parameter["ParameterValue"]	×
114	else:
115	resolved_param["ParameterValue"] = new_parameter["ParameterValue"]	×
116
117	resolved_param["NoEcho"] = pm.get("NoEcho", False)	×
118	resolved_parameters[pm_key] = resolved_param	×
119
120	# Note that SSM parameters always need to be resolved anew here
121	# TODO: support more parameter types
122	if pm["ParameterType"].startswith("AWS::SSM"):	×
123	if pm["ParameterType"] in [	×
124	"AWS::SSM::Parameter::Value<String>",
125	"AWS::SSM::Parameter::Value<AWS::EC2::Image::Id>",
126	"AWS::SSM::Parameter::Value<CommaDelimitedList>",
127	]:
128	# TODO: error handling (e.g. no permission to lookup SSM parameter or SSM parameter doesn't exist)
129	resolved_param["ResolvedValue"] = resolve_ssm_parameter(	×
130	account_id, region_name, resolved_param["ParameterValue"]
131	)
132	else:
133	raise Exception(f"Unsupported stack parameter type: {pm['ParameterType']}")	×
134
135	return resolved_parameters	×
136
137
138	# TODO: inject credentials / client factory for proper account/region lookup
139	def resolve_ssm_parameter(account_id: str, region_name: str, stack_parameter_value: str) -> str:	1✔
140	"""
141	Resolve the SSM stack parameter from the SSM service with a name equal to the stack parameter value.
142	"""
143	ssm_client = connect_to(aws_access_key_id=account_id, region_name=region_name).ssm	1✔
144	try:	1✔
145	return ssm_client.get_parameter(Name=stack_parameter_value)["Parameter"]["Value"]	1✔
146	except ClientError:	1✔
147	LOG.error("client error fetching parameter '%s'", stack_parameter_value)	1✔
148	raise	1✔
149
150
151	def strip_parameter_type(in_param: StackParameter) -> Parameter:	1✔
152	result = in_param.copy()	×
153	result.pop("ParameterType", None)	×
154	return result	×
155
156
157	def mask_no_echo(in_param: StackParameter) -> Parameter:	1✔
158	result = in_param.copy()	×
159	no_echo = result.pop("NoEcho", False)	×
160	if no_echo:	×
161	result["ParameterValue"] = "****"	×
162	return result	×
163
164
165	def convert_stack_parameters_to_list(	1✔
166	in_params: dict[str, StackParameter] \| None,
167	) -> list[StackParameter]:
168	if not in_params:	×
169	return []	×
170	return list(in_params.values())	×
171
172
173	def convert_stack_parameters_to_dict(in_params: list[Parameter] \| None) -> dict[str, Parameter]:	1✔
174	if not in_params:	×
175	return {}	×
176	return {p["ParameterKey"]: p for p in in_params}	×
177
178
179	class LegacyParameterProperties(TypedDict):	1✔
180	Value: str	1✔
181	ParameterType: str	1✔
182	ParameterValue: str \| None	1✔
183	ResolvedValue: str \| None	1✔
184
185
186	class LegacyParameter(TypedDict):	1✔
187	LogicalResourceId: str	1✔
188	Type: Literal["Parameter"]	1✔
189	Properties: LegacyParameterProperties	1✔
190
191
192	# TODO: not actually parameter_type but the logical "ID"
193	def map_to_legacy_structure(parameter_name: str, new_parameter: StackParameter) -> LegacyParameter:	1✔
194	"""
195	Helper util to convert a normal (resolved) stack parameter to a legacy parameter structure that can then be merged with stack resources.
196
197	:param new_parameter: a resolved stack parameter
198	:return: legacy parameter that can be merged with stack resources for uniform lookup based on logical ID
199	"""
200	return LegacyParameter(	×
201	LogicalResourceId=new_parameter["ParameterKey"],
202	Type="Parameter",
203	Properties=LegacyParameterProperties(
204	ParameterType=new_parameter.get("ParameterType"),
205	ParameterValue=new_parameter.get("ParameterValue"),
206	ResolvedValue=new_parameter.get("ResolvedValue"),
207	Value=new_parameter.get("ResolvedValue", new_parameter.get("ParameterValue")),
208	),
209	)

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