fix(parameters): use shared cache key in get_parameters_by_name (#8614)

* fix(parameters): use shared cache key in get_parameters_by_name

* fix(parameters): include decrypt in get_parameters_by_name cache key

get() stores entries under (name, transform, is_nested, sdk_options) since #8619,
so the batch path needs the same shape to share cache. Including decrypt also
stops a SecureString fetched with decrypt=True from being served from cache to a
later call without it.

---------

Co-authored-by: Leandro Damascena <lcdama@amazon.pt>

Author: SumiRann1

aws_lambda_powertools/utilities/parameters/ssm.py

@@ -572,7 +572,13 @@ def _get_parameters_by_name_from_cache(self, batch: dict[str, dict]) -> dict[str
         """Fetch each parameter from batch that hasn't been expired"""
         cache = {}
         for name, options in batch.items():
-            cache_key = (name, options["transform"])
+            # Same key shape as get() so both paths share cache entries; decrypt changes the value,
+            # so a SecureString fetched with decrypt must never be served to a call without it
+            cache_key = self._build_cache_key(
+                name=name,
+                transform=options["transform"],
+                sdk_options={"decrypt": options["decrypt"]},
+            )
             if self.has_not_expired_in_cache(cache_key):
                 cache[name] = self.store[cache_key].value
 
@@ -688,7 +694,11 @@ def _transform_and_cache_get_parameters_response(
                     transform_errors.append(name)
                     continue
 
-            _cache_key = (name, transform)
+            _cache_key = self._build_cache_key(
+                name=name,
+                transform=transform,
+                sdk_options={"decrypt": options["decrypt"]},
+            )
             self.add_to_cache(key=_cache_key, value=value, max_age=options["max_age"])
 
             response[name] = value

tests/functional/parameters/_boto3/test_utilities_parameters.py

@@ -1620,7 +1620,7 @@ def get_parameters(self, **kwargs):
     # THEN the transform failure is reported and not cached
     assert ret["_errors"] == [param]
     assert param not in ret
-    assert (param, "json") not in provider.store
+    assert provider._build_cache_key(name=param, transform="json", sdk_options={"decrypt": False}) not in provider.store
 
     # WHEN fetching again after the parameter is corrected
     ret = provider.get_parameters_by_name(parameters=params, transform="json", raise_on_error=False)
@@ -1629,7 +1629,7 @@ def get_parameters(self, **kwargs):
     assert client.get_parameters_calls == 2
     assert ret["_errors"] == []
     assert ret[param] == corrected_value
-    assert (param, "json") in provider.store
+    assert provider._build_cache_key(name=param, transform="json", sdk_options={"decrypt": False}) in provider.store
 
 
 def test_ssm_provider_get_parameters_by_name_raise_on_transform_failure_with_decrypt(mock_name, config):
@@ -1699,7 +1699,9 @@ def get_parameters(self, **kwargs):
     assert ret["_errors"] == []
     assert ret[decrypt_param] is None
     assert ret[batch_param] is None
-    assert (batch_param, "json") in provider.store
+    assert (
+        provider._build_cache_key(name=batch_param, transform="json", sdk_options={"decrypt": False}) in provider.store
+    )
 
 
 def test_ssm_provider_get_parameters_by_name_auto_transform_infers_from_parameter_name(mock_name, mock_value, config):
@@ -3296,7 +3298,7 @@ def _get_parameters_by_name(self, parameters, raise_on_error=True, decrypt=False
 def test_get_parameters_by_name_cache(monkeypatch, mock_name, mock_value, config):
     # GIVEN we have a parameter to fetch but is already in cache
     params = {mock_name: {}}
-    cache_key = (mock_name, None)
+    cache_key = SSMProvider(boto_config=config)._build_cache_key(name=mock_name, sdk_options={"decrypt": False})
 
     class TestProvider(SSMProvider):
         def __init__(self, boto_config: Config = config, **kwargs):
@@ -3306,7 +3308,7 @@ def _get_parameters_by_name(self, *args, **kwargs) -> tuple[dict[str, Any], list
             raise RuntimeError("Should not be called if it's in cache")
 
     provider = TestProvider()
-    provider.add_to_cache(key=(mock_name, None), value=mock_value, max_age=10)
+    provider.add_to_cache(key=cache_key, value=mock_value, max_age=10)
 
     monkeypatch.setitem(parameters.base.DEFAULT_PROVIDERS, "ssm", provider)
 
@@ -4615,6 +4617,105 @@ def test_secrets_provider_get_multiple_with_json_transform_error(config):
         provider.get_multiple(secret_names, transform="binary", raise_on_transform_error=True)
 
 
+def test_ssm_provider_get_parameters_by_name_uses_cache_populated_by_get(mock_name, mock_value, config):
+    # GIVEN a parameter fetched and cached through get()
+    param = f"/a/{mock_name}"
+    get_response = {"Parameter": {"Name": param, "Value": mock_value, "Type": "String", "Version": 1}}
+
+    provider = parameters.SSMProvider(boto_config=config)
+    stubber = stub.Stubber(provider.client)
+    # only ONE API call is stubbed: a second call would raise UnStubbedResponseError
+    stubber.add_response("get_parameter", get_response, {"Name": param, "WithDecryption": False})
+    stubber.activate()
+
+    # WHEN the same parameter is requested through get_parameters_by_name
+    try:
+        provider.get(param)
+        ret = provider.get_parameters_by_name(parameters={param: {}})
+
+        # THEN it is served from cache, without another API call
+        assert ret == {param: mock_value}
+        stubber.assert_no_pending_responses()
+    finally:
+        stubber.deactivate()
+
+
+def test_ssm_provider_get_uses_cache_populated_by_get_parameters_by_name(mock_name, mock_value, config):
+    # GIVEN a parameter fetched and cached through get_parameters_by_name
+    param = f"/a/{mock_name}"
+    fail = "/does_not_exist"  # stub model doesn't support all-success yet
+    stub_response = build_get_parameters_stub(params={param: mock_value}, invalid_parameters=[fail])
+
+    provider = parameters.SSMProvider(boto_config=config)
+    stubber = stub.Stubber(provider.client)
+    # only ONE API call is stubbed: a second call would raise UnStubbedResponseError
+    stubber.add_response("get_parameters", stub_response, {"Names": [param, fail]})
+    stubber.activate()
+
+    # WHEN the same parameter is requested through get()
+    try:
+        provider.get_parameters_by_name(parameters={param: {}, fail: {}}, raise_on_error=False)
+        ret = provider.get(param)
+
+        # THEN it is served from cache, without another API call
+        assert ret == mock_value
+        stubber.assert_no_pending_responses()
+    finally:
+        stubber.deactivate()
+
+
+def test_ssm_provider_get_parameters_by_name_decrypt_does_not_share_cache(mock_name, mock_value, config):
+    # GIVEN a SecureString fetched through get_parameters_by_name with decrypt=True
+    param = f"/a/{mock_name}"
+    fail = "/does_not_exist"  # stub model doesn't support all-success yet
+    decrypted = build_get_parameters_stub(params={param: mock_value}, invalid_parameters=[fail])
+    encrypted = build_get_parameters_stub(params={param: "AQICAHg..."}, invalid_parameters=[fail])
+
+    provider = parameters.SSMProvider(boto_config=config)
+    stubber = stub.Stubber(provider.client)
+    stubber.add_response("get_parameters", decrypted, {"Names": [param, fail], "WithDecryption": True})
+    stubber.add_response("get_parameters", encrypted, {"Names": [param, fail]})
+    stubber.activate()
+
+    # WHEN the same parameter is requested again without decrypt
+    try:
+        first = provider.get_parameters_by_name(parameters={param: {}, fail: {}}, decrypt=True, raise_on_error=False)
+        second = provider.get_parameters_by_name(parameters={param: {}, fail: {}}, raise_on_error=False)
+
+        # THEN the decrypted value is not served from cache and the API is called again
+        assert first[param] == mock_value
+        assert second[param] == "AQICAHg..."
+        stubber.assert_no_pending_responses()
+    finally:
+        stubber.deactivate()
+
+
+def test_ssm_provider_get_with_decrypt_does_not_share_cache_with_batch(mock_name, mock_value, config):
+    # GIVEN a SecureString fetched through get() with decrypt=True
+    param = f"/a/{mock_name}"
+    fail = "/does_not_exist"  # stub model doesn't support all-success yet
+    get_response = {"Parameter": {"Name": param, "Value": mock_value, "Type": "SecureString", "Version": 1}}
+    encrypted = build_get_parameters_stub(params={param: "AQICAHg..."}, invalid_parameters=[fail])
+
+    provider = parameters.SSMProvider(boto_config=config)
+    stubber = stub.Stubber(provider.client)
+    stubber.add_response("get_parameter", get_response, {"Name": param, "WithDecryption": True})
+    stubber.add_response("get_parameters", encrypted, {"Names": [param, fail]})
+    stubber.activate()
+
+    # WHEN the same parameter is requested through get_parameters_by_name without decrypt
+    try:
+        single = provider.get(param, decrypt=True)
+        batch = provider.get_parameters_by_name(parameters={param: {}, fail: {}}, raise_on_error=False)
+
+        # THEN the decrypted value is not served from cache and the API is called again
+        assert single == mock_value
+        assert batch[param] == "AQICAHg..."
+        stubber.assert_no_pending_responses()
+    finally:
+        stubber.deactivate()
+
+
 @pytest.mark.parametrize("force_fetch", [False, True])
 @pytest.mark.parametrize("secret_value", [{"SecretString": "value"}, {"SecretBinary": b"value"}])
 def test_secrets_provider_get_multiple_excludes_prefix_matches(config, force_fetch, secret_value):