diff --git a/aws_lambda_powertools/utilities/parameters/base.py b/aws_lambda_powertools/utilities/parameters/base.py index 5a6706776ce..05067cb46b8 100644 --- a/aws_lambda_powertools/utilities/parameters/base.py +++ b/aws_lambda_powertools/utilities/parameters/base.py @@ -334,9 +334,6 @@ def transform_value( err_msg = "Unable to transform value using '{transform}' transform: {exc}" - if isinstance(value, bytes): - value = value.decode("utf-8") - if isinstance(value, dict): # NOTE: We must handle partial failures when receiving multiple values # where one of the keys might fail during transform, e.g. `{"a": "valid", "b": "{"}` @@ -353,14 +350,14 @@ def transform_value( transformed_values[dict_key] = None return transformed_values - if transform == "auto": - # key="a.json", value='{"a": "b"}', or key="a.binary", value="b64_encoded" - transform_method = get_transform_method(value=key, transform=transform) - else: - # value='{"key": "value"} - transform_method = get_transform_method(value=value, transform=transform) + # Auto transforms inspect the parameter name; explicit transforms ignore it. + transform_method = get_transform_method(value=key, transform=transform) try: + # Identity transforms must preserve raw bytes. Decode failures belong to + # the same error policy as JSON/base64 failures. + if isinstance(value, bytes) and transform_method is not TRANSFORM_METHOD_MAPPING[None]: + value = value.decode("utf-8") return transform_method(value) except Exception as exc: if raise_on_transform_error: diff --git a/tests/functional/parameters/_boto3/test_utilities_parameters.py b/tests/functional/parameters/_boto3/test_utilities_parameters.py index 8442e4c9113..1c0b054fb47 100644 --- a/tests/functional/parameters/_boto3/test_utilities_parameters.py +++ b/tests/functional/parameters/_boto3/test_utilities_parameters.py @@ -3162,6 +3162,39 @@ def _get_multiple(self, path: str, **kwargs) -> dict[str, str]: assert value == mock_value +@pytest.mark.parametrize( + "transform,key", + [("binary", "secret"), ("json", "secret"), ("auto", "secret.binary"), ("auto", "secret.json")], +) +def test_transform_value_invalid_utf8_obeys_error_policy(transform, key): + value = b"\xff\xfe" + + with pytest.raises(parameters.TransformParameterError) as excinfo: + parameters.base.transform_value(value, transform=transform, key=key) + + assert isinstance(excinfo.value.__cause__, UnicodeDecodeError) + assert parameters.base.transform_value(value, transform=transform, key=key, raise_on_transform_error=False) is None + + +@pytest.mark.parametrize("value", [b"\xff\xfe", b"valid UTF-8"]) +@pytest.mark.parametrize("key", ["secret", "secret.bin"]) +def test_secret_provider_auto_preserves_unmatched_binary(value, key, config): + provider = parameters.SecretsProvider(boto_config=config) + with stub.Stubber(provider.client) as stubber: + stubber.add_response("get_secret_value", {"SecretBinary": value}, {"SecretId": key}) + assert provider.get(key, transform="auto") == value + stubber.assert_no_pending_responses() + + +def test_secret_provider_invalid_utf8_raises_transform_error(config): + provider = parameters.SecretsProvider(boto_config=config) + with stub.Stubber(provider.client) as stubber: + stubber.add_response("get_secret_value", {"SecretBinary": b"\xff\xfe"}, {"SecretId": "secret"}) + with pytest.raises(parameters.TransformParameterError): + provider.get("secret", transform="binary") + stubber.assert_no_pending_responses() + + def test_appconf_get_app_config_different_environment_and_application(monkeypatch, mock_name): # GIVEN a provider that reports which environment/application it was built for class TestProvider(BaseProvider):