Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
18 changes: 15 additions & 3 deletions moto/ssm/models.py
Original file line number Diff line number Diff line change
Expand Up @@ -1871,6 +1871,15 @@ def _raise_errors(self) -> None:
f"{count} validation error{plural} detected: {errors}"
)

def _strip_parameter_arn(self, name: str) -> str:
"""Reduce a full parameter ARN to the bare parameter name.

Names that are not ARNs are returned unchanged.
"""
if name.startswith(self.ssm_prefix):
return name.replace(self.ssm_prefix, "")
return name

def get_parameters(self, names: list[str]) -> dict[str, Parameter]:
result = {}

Expand All @@ -1882,7 +1891,11 @@ def get_parameters(self, names: list[str]) -> dict[str, Parameter]:
)

for name in set(names):
if name.split(":")[0] in self._parameters:
# Names may be full ARNs, as when reading a parameter shared from
# another account. Strip the prefix for the lookup, but keep the
# caller's original string as the key, so that the response layer
# reports InvalidParameters using what was actually requested.
if self._strip_parameter_arn(name).split(":")[0] in self._parameters:
try:
param = self.get_parameter(name)

Expand Down Expand Up @@ -2050,8 +2063,7 @@ def _match_filters(
return True

def get_parameter(self, name: str) -> Parameter | None:
if name.startswith(self.ssm_prefix):
name = name.replace(self.ssm_prefix, "")
name = self._strip_parameter_arn(name)

name_parts = name.split(":")
name_prefix = name_parts[0]
Expand Down
43 changes: 43 additions & 0 deletions tests/test_ssm/test_ssm.py
Original file line number Diff line number Diff line change
Expand Up @@ -2334,6 +2334,49 @@ def test_get_parameters_should_only_return_unique_requests():
assert len(response["Parameters"]) == 1


@pytest.mark.parametrize(
"as_arn",
[False, True],
ids=["by-name", "by-arn"],
)
@mock_aws
def test_get_parameters_by_arn(as_arn):
# A parameter shared from another account is read by its full ARN.
# get_parameter already supported this; get_parameters did not.
client = boto3.client("ssm", region_name=SSM_REGION)
client.put_parameter(Name="/Parameter1", Value="Value1", Type="String")
client.put_parameter(Name="/Parameter2", Value="Value2", Type="String")

def ref(name: str) -> str:
if as_arn:
return f"arn:aws:ssm:{SSM_REGION}:{ACCOUNT_ID}:parameter{name}"
return name

response = client.get_parameters(Names=[ref("/Parameter1"), ref("/Parameter2")])

assert response["InvalidParameters"] == []
assert {p["Name"]: p["Value"] for p in response["Parameters"]} == {
"/Parameter1": "Value1",
"/Parameter2": "Value2",
}


@mock_aws
def test_get_parameters_by_arn_reports_unknown_arn_as_invalid():
# An ARN that resolves to nothing must come back in InvalidParameters
# verbatim -- as requested -- not as the stripped name.
client = boto3.client("ssm", region_name=SSM_REGION)
client.put_parameter(Name="/Parameter1", Value="Value1", Type="String")

known = f"arn:aws:ssm:{SSM_REGION}:{ACCOUNT_ID}:parameter/Parameter1"
unknown = f"arn:aws:ssm:{SSM_REGION}:{ACCOUNT_ID}:parameter/DoesNotExist"

response = client.get_parameters(Names=[known, unknown])

assert [p["Name"] for p in response["Parameters"]] == ["/Parameter1"]
assert response["InvalidParameters"] == [unknown]


@mock_aws
def test_get_parameter_history_should_throw_exception_when_MaxResults_is_too_large():
client = boto3.client("ssm", region_name=SSM_REGION)
Expand Down