feat(parameters) - environment default variables

Author: leandrodamascena

aws_lambda_powertools/shared/constants.py

@@ -22,6 +22,9 @@
 
 IDEMPOTENCY_DISABLED_ENV: str = "POWERTOOLS_IDEMPOTENCY_DISABLED"
 
+PARAMETERS_DEFAULT_DECRYPT: str = "POWERTOOLS_PARAMETERS_SSM_DECRYPT"
+PARAMETERS_MAX_AGE: str = "POWERTOOLS_PARAMETERS_MAX_AGE"
+
 LOGGER_LAMBDA_CONTEXT_KEYS = [
     "function_arn",
     "function_memory_size",

aws_lambda_powertools/utilities/parameters/ssm.py

@@ -3,13 +3,18 @@
 """
 from __future__ import annotations
 
+import os
 from typing import TYPE_CHECKING, Any, Dict, List, Optional, Tuple, Union, overload
 
 import boto3
 from botocore.config import Config
 from typing_extensions import Literal
 
-from aws_lambda_powertools.shared.functions import slice_dictionary
+from aws_lambda_powertools.shared import constants
+from aws_lambda_powertools.shared.functions import (
+    resolve_truthy_env_var_choice,
+    slice_dictionary,
+)
 
 from .base import DEFAULT_MAX_AGE_SECS, DEFAULT_PROVIDERS, BaseProvider, transform_value
 from .exceptions import GetParameterError
@@ -112,7 +117,7 @@ def get(  # type: ignore[override]
         name: str,
         max_age: int = DEFAULT_MAX_AGE_SECS,
         transform: TransformOptions = None,
-        decrypt: bool = False,
+        decrypt: Optional[bool] = None,
         force_fetch: bool = False,
         **sdk_options,
     ) -> Optional[Union[str, dict, bytes]]:
@@ -145,6 +150,11 @@ def get(  # type: ignore[override]
             When the parameter provider fails to transform a parameter value.
         """
 
+        # Resolving if will use the value passed by parameter or the environment
+        decrypt = resolve_truthy_env_var_choice(
+            env=os.getenv(constants.PARAMETERS_DEFAULT_DECRYPT, "false"), choice=decrypt
+        )
+
         # Add to `decrypt` sdk_options to we can have an explicit option for this
         sdk_options["decrypt"] = decrypt
 
@@ -212,7 +222,7 @@ def get_parameters_by_name(
         self,
         parameters: Dict[str, Dict],
         transform: TransformOptions = None,
-        decrypt: bool = False,
+        decrypt: Optional[bool] = None,
         max_age: int = DEFAULT_MAX_AGE_SECS,
         raise_on_error: bool = True,
     ) -> Dict[str, str] | Dict[str, bytes] | Dict[str, dict]:
@@ -259,6 +269,12 @@ def get_parameters_by_name(
 
             When "_errors" reserved key is in parameters to be fetched from SSM.
         """
+
+        # Resolving if will use the default value (False), the value passed by parameter or the environment variable
+        decrypt = resolve_truthy_env_var_choice(
+            env=os.getenv(constants.PARAMETERS_DEFAULT_DECRYPT, "false"), choice=decrypt
+        )
+
         # Init potential batch/decrypt batch responses and errors
         batch_ret: Dict[str, Any] = {}
         decrypt_ret: Dict[str, Any] = {}
@@ -487,7 +503,7 @@ def _raise_if_errors_key_is_present(parameters: Dict, reserved_parameter: str, r
 def get_parameter(
     name: str,
     transform: Optional[str] = None,
-    decrypt: bool = False,
+    decrypt: Optional[bool] = None,
     force_fetch: bool = False,
     max_age: int = DEFAULT_MAX_AGE_SECS,
     **sdk_options,
@@ -543,6 +559,11 @@ def get_parameter(
     if "ssm" not in DEFAULT_PROVIDERS:
         DEFAULT_PROVIDERS["ssm"] = SSMProvider()
 
+    # Resolving if will use the default value (False), the value passed by parameter or the environment variable
+    decrypt = resolve_truthy_env_var_choice(
+        env=os.getenv(constants.PARAMETERS_DEFAULT_DECRYPT, "false"), choice=decrypt
+    )
+
     # Add to `decrypt` sdk_options to we can have an explicit option for this
     sdk_options["decrypt"] = decrypt
 
@@ -555,7 +576,7 @@ def get_parameters(
     path: str,
     transform: Optional[str] = None,
     recursive: bool = True,
-    decrypt: bool = False,
+    decrypt: Optional[bool] = None,
     force_fetch: bool = False,
     max_age: int = DEFAULT_MAX_AGE_SECS,
     raise_on_transform_error: bool = False,
@@ -617,6 +638,11 @@ def get_parameters(
     if "ssm" not in DEFAULT_PROVIDERS:
         DEFAULT_PROVIDERS["ssm"] = SSMProvider()
 
+    # Resolving if will use the default value (False), the value passed by parameter or the environment variable
+    decrypt = resolve_truthy_env_var_choice(
+        env=os.getenv(constants.PARAMETERS_DEFAULT_DECRYPT, "false"), choice=decrypt
+    )
+
     sdk_options["recursive"] = recursive
     sdk_options["decrypt"] = decrypt
 
@@ -634,7 +660,7 @@ def get_parameters(
 def get_parameters_by_name(
     parameters: Dict[str, Dict],
     transform: None = None,
-    decrypt: bool = False,
+    decrypt: Optional[bool] = None,
     max_age: int = DEFAULT_MAX_AGE_SECS,
     raise_on_error: bool = True,
 ) -> Dict[str, str]:
@@ -645,7 +671,7 @@ def get_parameters_by_name(
 def get_parameters_by_name(
     parameters: Dict[str, Dict],
     transform: Literal["binary"],
-    decrypt: bool = False,
+    decrypt: Optional[bool] = None,
     max_age: int = DEFAULT_MAX_AGE_SECS,
     raise_on_error: bool = True,
 ) -> Dict[str, bytes]:
@@ -656,7 +682,7 @@ def get_parameters_by_name(
 def get_parameters_by_name(
     parameters: Dict[str, Dict],
     transform: Literal["json"],
-    decrypt: bool = False,
+    decrypt: Optional[bool] = None,
     max_age: int = DEFAULT_MAX_AGE_SECS,
     raise_on_error: bool = True,
 ) -> Dict[str, Dict[str, Any]]:
@@ -667,7 +693,7 @@ def get_parameters_by_name(
 def get_parameters_by_name(
     parameters: Dict[str, Dict],
     transform: Literal["auto"],
-    decrypt: bool = False,
+    decrypt: Optional[bool] = None,
     max_age: int = DEFAULT_MAX_AGE_SECS,
     raise_on_error: bool = True,
 ) -> Union[Dict[str, str], Dict[str, dict]]:
@@ -677,7 +703,7 @@ def get_parameters_by_name(
 def get_parameters_by_name(
     parameters: Dict[str, Any],
     transform: TransformOptions = None,
-    decrypt: bool = False,
+    decrypt: Optional[bool] = None,
     max_age: int = DEFAULT_MAX_AGE_SECS,
     raise_on_error: bool = True,
 ) -> Union[Dict[str, str], Dict[str, bytes], Dict[str, dict]]:
@@ -732,6 +758,11 @@ def get_parameters_by_name(
     # NOTE: Decided against using multi-thread due to single-thread outperforming in 128M and 1G + timeout risk
     # see: https://github.com/awslabs/aws-lambda-powertools-python/issues/1040#issuecomment-1299954613
 
+    # Resolving if will use the default value (False), the value passed by parameter or the environment variable
+    decrypt = resolve_truthy_env_var_choice(
+        env=os.getenv(constants.PARAMETERS_DEFAULT_DECRYPT, "false"), choice=decrypt
+    )
+
     # Only create the provider if this function is called at least once
     if "ssm" not in DEFAULT_PROVIDERS:
         DEFAULT_PROVIDERS["ssm"] = SSMProvider()

docs/index.md

@@ -706,6 +706,7 @@ Core utilities such as Tracing, Logging, Metrics, and Event Handler will be avai
 | **POWERTOOLS_LOGGER_LOG_EVENT**           | Logs incoming event                                                                    | [Logging](./core/logger)                                                            | `false`               |
 | **POWERTOOLS_LOGGER_SAMPLE_RATE**         | Debug log sampling                                                                     | [Logging](./core/logger)                                                            | `0`                   |
 | **POWERTOOLS_LOG_DEDUPLICATION_DISABLED** | Disables log deduplication filter protection to use Pytest Live Log feature            | [Logging](./core/logger)                                                            | `false`               |
+| **POWERTOOLS_PARAMETERS_SSM_DECRYPT**     | Sets whether to decrypt or not values retrieved from AWS SSM Parameters Store          | [Parameters](.//utilities/parameters/#ssmprovider)                                  | `false`               |
 | **POWERTOOLS_DEV**                        | Increases verbosity across utilities                                                   | Multiple; see [POWERTOOLS_DEV effect below](#increasing-verbosity-across-utilities) | `false`               |
 | **LOG_LEVEL**                             | Sets logging level                                                                     | [Logging](./core/logger)                                                            | `INFO`                |
 

docs/utilities/parameters.md

@@ -179,6 +179,9 @@ The AWS Systems Manager Parameter Store provider supports two additional argumen
 
 You can create `SecureString` parameters, which are parameters that have a plaintext parameter name and an encrypted parameter value. If you don't use the `decrypt` argument, you will get an encrypted value. Read [here](https://docs.aws.amazon.com/kms/latest/developerguide/services-parameter-store.html) about best practices using KMS to secure your parameters.
 
+???+ tip
+	If you want to always decrypt parameters, you can set the `POWERTOOLS_PARAMETERS_SSM_DECRYPT=true` environment variable. **This will override the default value of `false` but can be overridden by the `decrypt` parameter**.
+
 === "builtin_provider_ssm_with_decrypt.py"
     ```python hl_lines="6 10 16"
     --8<-- "examples/parameters/src/builtin_provider_ssm_with_decrypt.py"

tests/functional/test_utilities_parameters.py

@@ -547,6 +547,44 @@ def test_ssm_provider_get_with_custom_client(mock_name, mock_value, mock_version
         stubber.deactivate()
 
 
+def test_ssm_provider_get_with_decrypt_environment_variable(monkeypatch, mock_name, mock_value, mock_version, config):
+    """
+    Test SSMProvider.get() with decrypt value replaced by environment variable
+    """
+
+    # Setting environment variable to override the default value
+    monkeypatch.setenv("POWERTOOLS_PARAMETERS_SSM_DECRYPT", "true")
+
+    # Create a new provider
+    provider = parameters.SSMProvider(config=config)
+
+    # Stub the boto3 client
+    stubber = stub.Stubber(provider.client)
+    response = {
+        "Parameter": {
+            "Name": mock_name,
+            "Type": "String",
+            "Value": mock_value,
+            "Version": mock_version,
+            "Selector": f"{mock_name}:{mock_version}",
+            "SourceResult": "string",
+            "LastModifiedDate": datetime(2015, 1, 1),
+            "ARN": f"arn:aws:ssm:us-east-2:111122223333:parameter/{mock_name}",
+        }
+    }
+    expected_params = {"Name": mock_name, "WithDecryption": True}
+    stubber.add_response("get_parameter", response, expected_params)
+    stubber.activate()
+
+    try:
+        value = provider.get(mock_name)
+
+        assert value == mock_value
+        stubber.assert_no_pending_responses()
+    finally:
+        stubber.deactivate()
+
+
 def test_ssm_provider_get_default_config(monkeypatch, mock_name, mock_value, mock_version):
     """
     Test SSMProvider.get() without specifying the config