diff --git a/aws_lambda_powertools/utilities/parameters/__init__.py b/aws_lambda_powertools/utilities/parameters/__init__.py index 9fcaa4fa701..9f8827ed9b6 100644 --- a/aws_lambda_powertools/utilities/parameters/__init__.py +++ b/aws_lambda_powertools/utilities/parameters/__init__.py @@ -8,8 +8,8 @@ from .base import BaseProvider, clear_caches from .dynamodb import DynamoDBProvider from .exceptions import GetParameterError, TransformParameterError -from .secrets import SecretsProvider, get_secret -from .ssm import SSMProvider, get_parameter, get_parameters, get_parameters_by_name +from .secrets import SecretsProvider, get_secret, set_secret +from .ssm import SSMProvider, get_parameter, get_parameters, get_parameters_by_name, set_parameter __all__ = [ "AppConfigProvider", @@ -21,8 +21,10 @@ "TransformParameterError", "get_app_config", "get_parameter", + "set_parameter", "get_parameters", "get_parameters_by_name", "get_secret", + "set_secret", "clear_caches", ] diff --git a/aws_lambda_powertools/utilities/parameters/base.py b/aws_lambda_powertools/utilities/parameters/base.py index 5ce06589613..2317ebc82d9 100644 --- a/aws_lambda_powertools/utilities/parameters/base.py +++ b/aws_lambda_powertools/utilities/parameters/base.py @@ -154,6 +154,12 @@ def _get(self, name: str, **sdk_options) -> Union[str, bytes, Dict[str, Any]]: """ raise NotImplementedError() + def set(self, name: str, value: Any, *, overwrite: bool = False, **kwargs): + """ + Set parameter value from the underlying parameter store + """ + raise NotImplementedError() + def get_multiple( self, path: str, diff --git a/aws_lambda_powertools/utilities/parameters/exceptions.py b/aws_lambda_powertools/utilities/parameters/exceptions.py index 1287568b463..6a9554bf142 100644 --- a/aws_lambda_powertools/utilities/parameters/exceptions.py +++ b/aws_lambda_powertools/utilities/parameters/exceptions.py @@ -9,3 +9,11 @@ class GetParameterError(Exception): class TransformParameterError(Exception): """When a provider fails to transform a parameter value""" + + +class SetParameterError(Exception): + """When a provider raises an exception on writing a SSM parameter""" + + +class SetSecretError(Exception): + """When a provider raises an exception on writing a secret""" diff --git a/aws_lambda_powertools/utilities/parameters/secrets.py b/aws_lambda_powertools/utilities/parameters/secrets.py index 9b09367cadb..0494c64985a 100644 --- a/aws_lambda_powertools/utilities/parameters/secrets.py +++ b/aws_lambda_powertools/utilities/parameters/secrets.py @@ -2,21 +2,27 @@ AWS Secrets Manager parameter retrieval and caching utility """ +from __future__ import annotations + +import json +import logging import os from typing import TYPE_CHECKING, Any, Dict, Literal, Optional, Union, overload import boto3 from botocore.config import Config -from aws_lambda_powertools.utilities.parameters.types import TransformOptions - if TYPE_CHECKING: from mypy_boto3_secretsmanager import SecretsManagerClient from aws_lambda_powertools.shared import constants from aws_lambda_powertools.shared.functions import resolve_max_age +from aws_lambda_powertools.shared.json_encoder import Encoder +from aws_lambda_powertools.utilities.parameters.base import DEFAULT_MAX_AGE_SECS, DEFAULT_PROVIDERS, BaseProvider +from aws_lambda_powertools.utilities.parameters.exceptions import SetSecretError +from aws_lambda_powertools.utilities.parameters.types import SetSecretResponse, TransformOptions -from .base import DEFAULT_MAX_AGE_SECS, DEFAULT_PROVIDERS, BaseProvider +logger = logging.getLogger(__name__) class SecretsProvider(BaseProvider): @@ -117,6 +123,134 @@ def _get_multiple(self, path: str, **sdk_options) -> Dict[str, str]: """ raise NotImplementedError() + def _create_secret(self, name: str, **sdk_options): + """ + Create a secret with the given name. + + Parameters: + ---------- + name: str + The name of the secret. + **sdk_options: + Additional options to be passed to the create_secret method. + + Raises: + SetSecretError: If there is an error setting the secret. + """ + try: + sdk_options["Name"] = name + return self.client.create_secret(**sdk_options) + except Exception as exc: + raise SetSecretError(f"Error setting secret - {str(exc)}") from exc + + def _update_secret(self, name: str, **sdk_options): + """ + Update a secret with the given name. + + Parameters: + ---------- + name: str + The name of the secret. + **sdk_options: + Additional options to be passed to the create_secret method. + """ + sdk_options["SecretId"] = name + return self.client.put_secret_value(**sdk_options) + + def set( + self, + name: str, + value: Union[str, dict, bytes], + *, # force keyword arguments + client_request_token: Optional[str] = None, + **sdk_options, + ) -> SetSecretResponse: + """ + Modify the details of a secret or create a new secret if it doesn't already exist. + + We aim to minimize API calls by assuming that the secret already exists and needs updating. + If it doesn't exist, we attempt to create a new one. Refer to the following workflow for a better understanding: + + + ┌────────────────────────┐ ┌─────────────────┐ + ┌───────▶│Resource NotFound error?│────▶│Create Secret API│─────┐ + │ └────────────────────────┘ └─────────────────┘ │ + │ │ + │ │ + │ ▼ + ┌─────────────────┐ ┌─────────────────────┐ + │Update Secret API│────────────────────────────────────────────▶│ Return or Exception │ + └─────────────────┘ └─────────────────────┘ + + Parameters + ---------- + name: str + The ARN or name of the secret to add a new version to or create a new one. + value: str, dict or bytes + Specifies text data that you want to encrypt and store in this new version of the secret. + client_request_token: str, optional + This value helps ensure idempotency. It's recommended that you generate + a UUID-type value to ensure uniqueness within the specified secret. + This value becomes the VersionId of the new version. This field is + auto-populated if not provided, but no idempotency will be enforced this way. + sdk_options: dict, optional + Dictionary of options that will be passed to the Secrets Manager update_secret API call + + Raises + ------ + SetSecretError + When attempting to update or create a secret fails. + + Returns: + ------- + SetSecretResponse: + The dict returned by boto3. + + Example + ------- + **Sets a secret*** + + >>> from aws_lambda_powertools.utilities import parameters + >>> + >>> parameters.set_secret(name="llamas-are-awesome", value="supers3cr3tllam@passw0rd") + + **Sets a secret and includes an client_request_token** + + >>> from aws_lambda_powertools.utilities import parameters + >>> import uuid + >>> + >>> parameters.set_secret( + name="my-secret", + value='{"password": "supers3cr3tllam@passw0rd"}', + client_request_token=str(uuid.uuid4()) + ) + + URLs: + ------- + https://boto3.amazonaws.com/v1/documentation/api/latest/reference/services/secretsmanager/client/put_secret_value.html + https://boto3.amazonaws.com/v1/documentation/api/latest/reference/services/secretsmanager/client/create_secret.html + """ + + if isinstance(value, dict): + value = json.dumps(value, cls=Encoder) + + if isinstance(value, bytes): + sdk_options["SecretBinary"] = value + else: + sdk_options["SecretString"] = value + + if client_request_token: + sdk_options["ClientRequestToken"] = client_request_token + + try: + logger.debug(f"Attempting to update secret {name}") + return self._update_secret(name=name, **sdk_options) + except self.client.exceptions.ResourceNotFoundException: + logger.debug(f"Secret {name} doesn't exist, creating a new one") + return self._create_secret(name=name, **sdk_options) + except Exception as exc: + raise SetSecretError(f"Error setting secret - {str(exc)}") from exc + @overload def get_secret( @@ -224,3 +358,87 @@ def get_secret( force_fetch=force_fetch, **sdk_options, ) + + +def set_secret( + name: str, + value: Union[str, bytes], + *, # force keyword arguments + client_request_token: Optional[str] = None, + **sdk_options, +) -> SetSecretResponse: + """ + Modify the details of a secret or create a new secret if it doesn't already exist. + + We aim to minimize API calls by assuming that the secret already exists and needs updating. + If it doesn't exist, we attempt to create a new one. Refer to the following workflow for a better understanding: + + + ┌────────────────────────┐ ┌─────────────────┐ + ┌───────▶│Resource NotFound error?│────▶│Create Secret API│─────┐ + │ └────────────────────────┘ └─────────────────┘ │ + │ │ + │ │ + │ ▼ + ┌─────────────────┐ ┌─────────────────────┐ + │Update Secret API│────────────────────────────────────────────▶│ Return or Exception │ + └─────────────────┘ └─────────────────────┘ + + Parameters + ---------- + name: str + The ARN or name of the secret to add a new version to or create a new one. + value: str, dict or bytes + Specifies text data that you want to encrypt and store in this new version of the secret. + client_request_token: str, optional + This value helps ensure idempotency. It's recommended that you generate + a UUID-type value to ensure uniqueness within the specified secret. + This value becomes the VersionId of the new version. This field is + auto-populated if not provided, but no idempotency will be enforced this way. + sdk_options: dict, optional + Dictionary of options that will be passed to the Secrets Manager update_secret API call + + Raises + ------ + SetSecretError + When attempting to update or create a secret fails. + + Returns: + ------- + SetSecretResponse: + The dict returned by boto3. + + Example + ------- + **Sets a secret*** + + >>> from aws_lambda_powertools.utilities import parameters + >>> + >>> parameters.set_secret(name="llamas-are-awesome", value="supers3cr3tllam@passw0rd") + + **Sets a secret and includes an client_request_token** + + >>> from aws_lambda_powertools.utilities import parameters + >>> + >>> parameters.set_secret( + name="my-secret", + value='{"password": "supers3cr3tllam@passw0rd"}', + client_request_token="61f2af5f-5f75-44b1-a29f-0cc37af55b11" + ) + + URLs: + ------- + https://boto3.amazonaws.com/v1/documentation/api/latest/reference/services/secretsmanager/client/put_secret_value.html + https://boto3.amazonaws.com/v1/documentation/api/latest/reference/services/secretsmanager/client/create_secret.html + """ + + # Only create the provider if this function is called at least once + if "secrets" not in DEFAULT_PROVIDERS: + DEFAULT_PROVIDERS["secrets"] = SecretsProvider() + + return DEFAULT_PROVIDERS["secrets"].set( + name=name, + value=value, + client_request_token=client_request_token, + **sdk_options, + ) diff --git a/aws_lambda_powertools/utilities/parameters/ssm.py b/aws_lambda_powertools/utilities/parameters/ssm.py index 1be07e7c9f0..76553bda0fe 100644 --- a/aws_lambda_powertools/utilities/parameters/ssm.py +++ b/aws_lambda_powertools/utilities/parameters/ssm.py @@ -4,6 +4,7 @@ from __future__ import annotations +import logging import os from typing import TYPE_CHECKING, Any, Dict, List, Optional, Tuple, Union, overload @@ -17,15 +18,24 @@ slice_dictionary, ) from aws_lambda_powertools.shared.types import Literal - -from .base import DEFAULT_MAX_AGE_SECS, DEFAULT_PROVIDERS, BaseProvider, transform_value -from .exceptions import GetParameterError -from .types import TransformOptions +from aws_lambda_powertools.utilities.parameters.base import ( + DEFAULT_MAX_AGE_SECS, + DEFAULT_PROVIDERS, + BaseProvider, + transform_value, +) +from aws_lambda_powertools.utilities.parameters.exceptions import GetParameterError, SetParameterError +from aws_lambda_powertools.utilities.parameters.types import PutParameterResponse, TransformOptions if TYPE_CHECKING: from mypy_boto3_ssm import SSMClient from mypy_boto3_ssm.type_defs import GetParametersResultTypeDef +SSM_PARAMETER_TYPES = Literal["String", "StringList", "SecureString"] +SSM_PARAMETER_TIER = Literal["Standard", "Advanced", "Intelligent-Tiering"] + +logger = logging.getLogger(__name__) + class SSMProvider(BaseProvider): """ @@ -169,6 +179,126 @@ def get( # type: ignore[override] return super().get(name, max_age, transform, force_fetch, **sdk_options) + @overload + def set( + self, + name: str, + value: list[str], + *, + overwrite: bool = False, + description: str = "", + parameter_type: Literal["StringList"] = "StringList", + tier: Literal["Standard", "Advanced", "Intelligent-Tiering"] = "Standard", + kms_key_id: str | None = "None", + **sdk_options, + ): ... + + @overload + def set( + self, + name: str, + value: str, + *, + overwrite: bool = False, + description: str = "", + parameter_type: Literal["SecureString"] = "SecureString", + tier: Literal["Standard", "Advanced", "Intelligent-Tiering"] = "Standard", + kms_key_id: str, + **sdk_options, + ): ... + + @overload + def set( + self, + name: str, + value: str, + *, + overwrite: bool = False, + description: str = "", + parameter_type: Literal["String"] = "String", + tier: Literal["Standard", "Advanced", "Intelligent-Tiering"] = "Standard", + kms_key_id: str | None = None, + **sdk_options, + ): ... + + def set( + self, + name: str, + value: str | list[str], + *, + overwrite: bool = False, + description: str = "", + parameter_type: SSM_PARAMETER_TYPES = "String", + tier: SSM_PARAMETER_TIER = "Standard", + kms_key_id: str | None = None, + **sdk_options, + ) -> PutParameterResponse: + """ + Sets a parameter in AWS Systems Manager Parameter Store. + + Parameters + ---------- + name: str + The fully qualified name includes the complete hierarchy of the parameter name and name. + value: str + The parameter value + overwrite: bool, optional + If the parameter value should be overwritten, False by default + description: str, optional + The description of the parameter + parameter_type: str, optional + Type of the parameter. Allowed values are String, StringList, and SecureString + tier: str, optional + The parameter tier to use. Allowed values are Standard, Advanced, and Intelligent-Tiering + kms_key_id: str, optional + The KMS key id to use to encrypt the parameter + sdk_options: dict, optional + Dictionary of options that will be passed to the Parameter Store get_parameter API call + + Raises + ------ + SetParameterError + When the parameter provider fails to retrieve a parameter value for + a given name. + + URLs: + ------- + https://boto3.amazonaws.com/v1/documentation/api/latest/reference/services/ssm/client/put_parameter.html + + Example + ------- + **Sets a parameter value from Systems Manager Parameter Store** + + >>> from aws_lambda_powertools.utilities import parameters + >>> + >>> response = parameters.set_parameter(name="/my/example/parameter", value="More Powertools") + >>> + >>> print(response) + 123 + + Returns + ------- + PutParameterResponse + The dict returned by boto3. + """ + opts = { + "Name": name, + "Value": value, + "Overwrite": overwrite, + "Type": parameter_type, + "Tier": tier, + "Description": description, + **sdk_options, + } + + if kms_key_id: + opts["KeyId"] = kms_key_id + + try: + return self.client.put_parameter(**opts) + except Exception as exc: + raise SetParameterError(f"Error setting parameter - {str(exc)}") from exc + def _get(self, name: str, decrypt: bool = False, **sdk_options) -> str: """ Retrieve a parameter value from AWS Systems Manager Parameter Store @@ -811,6 +941,81 @@ def get_parameters( ) +def set_parameter( + name: str, + value: str, + *, # force keyword arguments + overwrite: bool = False, + description: str = "", + parameter_type: SSM_PARAMETER_TYPES = "String", + tier: SSM_PARAMETER_TIER = "Standard", + kms_key_id: str | None = None, + **sdk_options, +) -> PutParameterResponse: + """ + Sets a parameter in AWS Systems Manager Parameter Store. + + Parameters + ---------- + name: str + The fully qualified name includes the complete hierarchy of the parameter name and name. + value: str + The parameter value + overwrite: bool, optional + If the parameter value should be overwritten, False by default + description: str, optional + The description of the parameter + parameter_type: str, optional + Type of the parameter. Allowed values are String, StringList, and SecureString + tier: str, optional + The parameter tier to use. Allowed values are Standard, Advanced, and Intelligent-Tiering + kms_key_id: str, optional + The KMS key id to use to encrypt the parameter + sdk_options: dict, optional + Dictionary of options that will be passed to the Parameter Store get_parameter API call + + Raises + ------ + SetParameterError + When attempting to set a parameter fails. + + URLs: + ------- + https://boto3.amazonaws.com/v1/documentation/api/latest/reference/services/ssm/client/put_parameter.html + + Example + ------- + **Sets a parameter value from Systems Manager Parameter Store** + + >>> from aws_lambda_powertools.utilities import parameters + >>> + >>> response = parameters.set_parameter(name="/my/example/parameter", value="More Powertools") + >>> + >>> print(response) + 123 + + Returns + ------- + PutParameterResponse + The dict returned by boto3. + """ + + # Only create the provider if this function is called at least once + if "ssm" not in DEFAULT_PROVIDERS: + DEFAULT_PROVIDERS["ssm"] = SSMProvider() + + return DEFAULT_PROVIDERS["ssm"].set( + name, + value, + parameter_type=parameter_type, + overwrite=overwrite, + tier=tier, + description=description, + kms_key_id=kms_key_id, + **sdk_options, + ) + + @overload def get_parameters_by_name( parameters: Dict[str, Dict], diff --git a/aws_lambda_powertools/utilities/parameters/types.py b/aws_lambda_powertools/utilities/parameters/types.py index faa06cee89e..c087a3764f4 100644 --- a/aws_lambda_powertools/utilities/parameters/types.py +++ b/aws_lambda_powertools/utilities/parameters/types.py @@ -1,3 +1,20 @@ -from aws_lambda_powertools.shared.types import Literal +from typing import Any, Optional + +from aws_lambda_powertools.shared.types import Dict, List, Literal, TypedDict TransformOptions = Literal["json", "binary", "auto", None] + + +class PutParameterResponse(TypedDict): + Version: int + Tier: str + ResponseMetadata: dict + + +class SetSecretResponse(TypedDict): + ARN: str + Name: str + VersionId: str + VersionStages: Optional[List[str]] + ReplicationStatus: Optional[List[Dict[str, Any]]] + ResponseMetadata: dict diff --git a/docs/utilities/parameters.md b/docs/utilities/parameters.md index d2d80230c77..92c0c53ce86 100644 --- a/docs/utilities/parameters.md +++ b/docs/utilities/parameters.md @@ -27,16 +27,18 @@ This utility requires additional permissions to work as expected. ???+ note Different parameter providers require different permissions. -| Provider | Function/Method | IAM Permission | -| --------- | ---------------------------------------------------------------------- | ------------------------------------------------------------------------------------ | -| SSM | **`get_parameter`**, **`SSMProvider.get`** | **`ssm:GetParameter`** | -| SSM | **`get_parameters`**, **`SSMProvider.get_multiple`** | **`ssm:GetParametersByPath`** | -| SSM | **`get_parameters_by_name`**, **`SSMProvider.get_parameters_by_name`** | **`ssm:GetParameter`** and **`ssm:GetParameters`** | -| SSM | If using **`decrypt=True`** | You must add an additional permission **`kms:Decrypt`** | -| Secrets | **`get_secret`**, **`SecretsProvider.get`** | **`secretsmanager:GetSecretValue`** | -| DynamoDB | **`DynamoDBProvider.get`** | **`dynamodb:GetItem`** | -| DynamoDB | **`DynamoDBProvider.get_multiple`** | **`dynamodb:Query`** | -| AppConfig | **`get_app_config`**, **`AppConfigProvider.get_app_config`** | **`appconfig:GetLatestConfiguration`** and **`appconfig:StartConfigurationSession`** | +| Provider | Function/Method | IAM Permission | +| --------- | ---------------------------------------------------------------------- | ----------------------------------------------------------------------------------------------- | +| SSM | **`get_parameter`**, **`SSMProvider.get`** | **`ssm:GetParameter`** | +| SSM | **`get_parameters`**, **`SSMProvider.get_multiple`** | **`ssm:GetParametersByPath`** | +| SSM | **`get_parameters_by_name`**, **`SSMProvider.get_parameters_by_name`** | **`ssm:GetParameter`** and **`ssm:GetParameters`** | +| SSM | **`set_parameter`**, **`SSMProvider.set_parameter`** | **`ssm:PutParameter`** | +| SSM | If using **`decrypt=True`** | You must add an additional permission **`kms:Decrypt`** | +| Secrets | **`get_secret`**, **`SecretsProvider.get`** | **`secretsmanager:GetSecretValue`** | +| Secrets | **`set_secret`**, **`SecretsProvider.set`** | **`secretsmanager:PutSecretValue`** and **`secretsmanager:CreateSecret`** (if creating secrets) | +| DynamoDB | **`DynamoDBProvider.get`** | **`dynamodb:GetItem`** | +| DynamoDB | **`DynamoDBProvider.get_multiple`** | **`dynamodb:Query`** | +| AppConfig | **`get_app_config`**, **`AppConfigProvider.get_app_config`** | **`appconfig:GetLatestConfiguration`** and **`appconfig:StartConfigurationSession`** | ### Fetching parameters @@ -84,6 +86,22 @@ For multiple parameters, you can use either: --8<-- "examples/parameters/src/get_parameter_by_name_error_handling.py" ``` +### Setting parameters + +You can set a parameter using the `set_parameter` high-level function. This will create a new parameter if it doesn't exist. + +=== "getting_started_set_single_ssm_parameter.py" + ```python hl_lines="8" + --8<-- "examples/parameters/src/getting_started_set_single_ssm_parameter.py" + ``` + +=== "getting_started_set_ssm_parameter_overwrite.py" + Sometimes you may be setting a parameter that you will have to update later on. Use the `overwrite` option to overwrite any existing value. If you do not set this option, the parameter value will not be overwritten and an exception will be raised. + + ```python hl_lines="8 12" + --8<-- "examples/parameters/src/getting_started_set_ssm_parameter_overwrite.py" + ``` + ### Fetching secrets You can fetch secrets stored in Secrets Manager using `get_secret`. @@ -93,6 +111,18 @@ You can fetch secrets stored in Secrets Manager using `get_secret`. --8<-- "examples/parameters/src/getting_started_secret.py" ``` +### Setting secrets + +You can set secrets stored in Secrets Manager using `set_secret`. + +???+ note + We strive to minimize API calls by attempting to update existing secrets as our primary approach. If a secret doesn't exist, we proceed to create a new one. + +=== "getting_started_secret.py" + ```python hl_lines="4 25" + --8<-- "examples/parameters/src/getting_started_setting_secret.py" + ``` + ### Fetching app configurations You can fetch application configurations in AWS AppConfig using `get_app_config`. diff --git a/examples/parameters/src/getting_started_set_single_ssm_parameter.py b/examples/parameters/src/getting_started_set_single_ssm_parameter.py new file mode 100644 index 00000000000..4718d99105f --- /dev/null +++ b/examples/parameters/src/getting_started_set_single_ssm_parameter.py @@ -0,0 +1,12 @@ +from aws_lambda_powertools.utilities import parameters +from aws_lambda_powertools.utilities.typing import LambdaContext + + +def lambda_handler(event: dict, context: LambdaContext) -> dict: + try: + # Set a single parameter, returns the version ID of the parameter + parameter_version = parameters.set_parameter(name="/mySuper/Parameter", value="PowerToolsIsAwesome") + + return {"mySuperParameterVersion": parameter_version, "statusCode": 200} + except parameters.exceptions.SetParameterError as error: + return {"comments": None, "message": str(error), "statusCode": 400} diff --git a/examples/parameters/src/getting_started_set_ssm_parameter_overwrite.py b/examples/parameters/src/getting_started_set_ssm_parameter_overwrite.py new file mode 100644 index 00000000000..a80cf2d9818 --- /dev/null +++ b/examples/parameters/src/getting_started_set_ssm_parameter_overwrite.py @@ -0,0 +1,17 @@ +from aws_lambda_powertools.utilities import parameters +from aws_lambda_powertools.utilities.typing import LambdaContext + + +def lambda_handler(event: dict, context: LambdaContext) -> dict: + try: + # Set a single parameter, but overwrite if it already exists. + # Overwrite is False by default, so we explicitly set it to True + updating_parameter = parameters.set_parameter( + name="/mySuper/Parameter", + value="PowerToolsIsAwesome", + overwrite=True, + ) + + return {"mySuperParameterVersion": updating_parameter, "statusCode": 200} + except parameters.exceptions.SetParameterError as error: + return {"comments": None, "message": str(error), "statusCode": 400} diff --git a/examples/parameters/src/getting_started_setting_secret.py b/examples/parameters/src/getting_started_setting_secret.py new file mode 100644 index 00000000000..50412380fdf --- /dev/null +++ b/examples/parameters/src/getting_started_setting_secret.py @@ -0,0 +1,30 @@ +from typing import Any + +from aws_lambda_powertools import Logger +from aws_lambda_powertools.utilities import parameters +from aws_lambda_powertools.utilities.typing import LambdaContext + +logger = Logger(serialize_stacktrace=True) + + +def access_token(client_id: str, client_secret: str, audience: str) -> str: + # example function that returns a JWT Access Token + # add your own logic here + return f"{client_id}.{client_secret}.{audience}" + + +def lambda_handler(event: dict, context: LambdaContext): + try: + client_id: Any = parameters.get_parameter("/aws-powertools/client_id") + client_secret: Any = parameters.get_parameter("/aws-powertools/client_secret") + audience: Any = parameters.get_parameter("/aws-powertools/audience") + + jwt_token = access_token(client_id=client_id, client_secret=client_secret, audience=audience) + + # set-secret will create a new secret if it doesn't exist and return the version id + update_secret_version_id = parameters.set_secret(name="/aws-powertools/jwt_token", value=jwt_token) + + return {"access_token": "updated", "statusCode": 200, "update_secret_version_id": update_secret_version_id} + except parameters.exceptions.SetSecretError as error: + logger.exception(error) + return {"access_token": "updated", "statusCode": 400} diff --git a/tests/functional/test_utilities_parameters.py b/tests/functional/test_utilities_parameters.py index 5ff043f7ed3..334b3d37ea5 100644 --- a/tests/functional/test_utilities_parameters.py +++ b/tests/functional/test_utilities_parameters.py @@ -4,6 +4,7 @@ import json import random import string +import uuid from datetime import datetime, timedelta from io import BytesIO from typing import Any, Dict, List, Tuple @@ -511,6 +512,373 @@ def test_ssm_provider_get(mock_name, mock_value, mock_version, config): stubber.deactivate() +def test_set_parameter(monkeypatch, mock_name, mock_value): + """ + Test set_parameter() + """ + + class TestProvider(BaseProvider): + def set(self, name: str, value: Any, *, overwrite: bool = False, **kwargs) -> str: + assert name == mock_name + return mock_value + + def _get(self, name: str, **kwargs) -> str: + raise NotImplementedError() + + def _get_multiple(self, path: str, **kwargs) -> Dict[str, str]: + raise NotImplementedError() + + monkeypatch.setitem(parameters.base.DEFAULT_PROVIDERS, "ssm", TestProvider()) + + value = parameters.set_parameter(name=mock_name, value=mock_value) + + assert value == mock_value + + +def test_ssm_provider_set_parameter(mock_name, mock_value, mock_version, config): + """ + Test SSMProvider.set_parameter() with a non-cached value + """ + # GIVEN a SSMProvider instance with default values + provider = parameters.SSMProvider(config=config) + + # WHEN setting a parameter + stubber = stub.Stubber(provider.client) + response = {"Version": mock_version, "Tier": "Standard"} + expected_params = { + "Name": mock_name, + "Value": mock_value, + "Type": "String", + "Overwrite": False, + "Description": "", + "Tier": "Standard", + } + stubber.add_response("put_parameter", response, expected_params) + stubber.activate() + + # THEN it should return values + try: + assert provider.set(name=mock_name, value=mock_value) == response + stubber.assert_no_pending_responses() + finally: + stubber.deactivate() + + +def test_ssm_provider_set_parameter_default_config(monkeypatch, mock_name, mock_value, mock_version): + """ + Test SSMProvider._set() without specifying the config + """ + monkeypatch.setenv("AWS_DEFAULT_REGION", "us-east-2") + + # GIVEN a SSMProvider instance with default values + provider = parameters.SSMProvider() + + # WHEN setting a parameter + stubber = stub.Stubber(provider.client) + response = {"Version": mock_version, "Tier": "Advanced"} + expected_params = { + "Name": mock_name, + "Value": mock_value, + "Type": "String", + "Overwrite": False, + "Tier": "Standard", + "Description": "", + } + stubber.add_response("put_parameter", response, expected_params) + stubber.activate() + + # THEN it should return values + try: + assert provider.set(name=mock_name, value=mock_value) == response + stubber.assert_no_pending_responses() + finally: + stubber.deactivate() + + +def test_ssm_provider_set_parameter_with_custom_options(monkeypatch, mock_name, mock_value, mock_version): + """ + Test SSMProvider._set() with custom options + """ + + monkeypatch.setenv("AWS_DEFAULT_REGION", "us-east-2") + + # GIVEN a SSMProvider instance + provider = parameters.SSMProvider() + + # WHEN using custom parameters + stubber = stub.Stubber(provider.client) + response = {"Version": mock_version, "Tier": "Advanced"} + expected_params = { + "Name": mock_name, + "Value": mock_value, + "Type": "SecureString", + "Overwrite": True, + "Tier": "Advanced", + "Description": "Parameter", + "KeyId": "arn:aws:kms:us-west-2:111122223333:key/1234abcd-12ab-34cd-56ef-1234567890ab", + } + stubber.add_response("put_parameter", response, expected_params) + stubber.activate() + + # THEN it should return values + try: + version = provider.set( + name=mock_name, + value=mock_value, + tier="Advanced", + parameter_type="SecureString", + overwrite=True, + description="Parameter", + kms_key_id="arn:aws:kms:us-west-2:111122223333:key/1234abcd-12ab-34cd-56ef-1234567890ab", + ) + + assert version == response + stubber.assert_no_pending_responses() + finally: + stubber.deactivate() + + +def test_ssm_provider_set_parameter_raise_on_failure(mock_name, mock_value, mock_version, config): + """ + Test SSMProvider.set_parameter() with failure + """ + # GIVEN a SSMProvider instance + provider = parameters.SSMProvider(config=config) + + # Stub the boto3 client + stubber = stub.Stubber(provider.client) + response = {"Version": mock_version, "Tier": "Standard"} + expected_params = { + "Name": mock_name, + "Value": mock_value, + "Type": "String", + "Overwrite": False, + "Description": "", + "Tier": "NoTier", + } + stubber.add_response("put_parameter", response, expected_params) + stubber.activate() + + # WHEN cannot set a Parameter with tier=NoTier + # THEN raise SetParameterError + with pytest.raises(parameters.exceptions.SetParameterError, match="Error setting parameter*"): + try: + provider.set(name=mock_name, value=mock_value) + stubber.assert_no_pending_responses() + finally: + stubber.deactivate() + + +def test_set_secret(monkeypatch, mock_name, mock_value): + """ + Test set_secret() + """ + + # GIVEN a mock implementation of BaseProvider set method + class TestProvider(BaseProvider): + def set(self, name: str, value: Any, *, overwrite: bool = False, **kwargs) -> str: + assert name == mock_name + return mock_value + + def _get(self, name: str, **kwargs) -> str: + raise NotImplementedError() + + def _get_multiple(self, path: str, **kwargs) -> Dict[str, str]: + raise NotImplementedError() + + monkeypatch.setitem(parameters.base.DEFAULT_PROVIDERS, "secrets", TestProvider()) + + # WHEN set_secret function is called + value = parameters.set_secret(name=mock_name, value=mock_value) + + # THEN it should return the mock_value + assert value == mock_value + + +def test_secret_provider_update_secret_with_plain_text_value(mock_name, mock_value, config): + """ + Test SecretsProvider.set() with a plain text value + """ + # GIVEN a SecretsProvider instance + provider = parameters.SecretsProvider(config=config) + + client_request_token = str(uuid.uuid4()) + + # WHEN setting a secret with a plain text value + stubber = stub.Stubber(provider.client) + response = {"Name": mock_name, "ARN": f"arn:aws:secretsmanager:us-east-1:132456789012:secret/{mock_name}"} + expected_params = { + "SecretId": mock_name, + "SecretString": mock_value, + "ClientRequestToken": client_request_token, + } + stubber.add_response("put_secret_value", response, expected_params) + stubber.activate() + + # THEN it should call put_secret_value with the plain text value and the client request token + try: + assert response == provider.set(name=mock_name, value=mock_value, client_request_token=client_request_token) + stubber.assert_no_pending_responses() + finally: + stubber.deactivate() + + +def test_secret_provider_update_secret_with_binary_value(mock_name, config): + """ + Test SecretsProvider.set() with a binary value + """ + + mock_value = b"value_to_test" + + # GIVEN a SecretsProvider instance + provider = parameters.SecretsProvider(config=config) + + # WHEN setting a secret with a binary value + stubber = stub.Stubber(provider.client) + response = {"Name": mock_name, "ARN": f"arn:aws:secretsmanager:us-east-1:132456789012:secret/{mock_name}"} + expected_params = { + "SecretId": mock_name, + "SecretBinary": mock_value, + } + stubber.add_response("put_secret_value", response, expected_params) + stubber.activate() + + # THEN it should call put_secret_value with the binary value + try: + assert response == provider.set(name=mock_name, value=mock_value) + stubber.assert_no_pending_responses() + finally: + stubber.deactivate() + + +def test_secret_provider_update_secret_with_dict_value(mock_name, config): + """ + Test SecretsProvider.set() with a dict value + """ + + mock_value = {"key": "powertools"} + + # GIVEN a SecretsProvider instance + provider = parameters.SecretsProvider(config=config) + + # WHEN setting a secret with a dictionary value + stubber = stub.Stubber(provider.client) + response = {"Name": mock_name, "ARN": f"arn:aws:secretsmanager:us-east-1:132456789012:secret/{mock_name}"} + expected_params = { + "SecretId": mock_name, + "SecretString": json.dumps(mock_value), + } + stubber.add_response("put_secret_value", response, expected_params) + stubber.activate() + + # THEN it should encode the dictionary as JSON and call put_secret_value with the encoded value + try: + assert response == provider.set(name=mock_name, value=mock_value) + stubber.assert_no_pending_responses() + finally: + stubber.deactivate() + + +def test_secret_provider_update_secret_with_raise_on_failure(mock_name, mock_value, config): + """ + Test SecretsProvider.set() with raise on failure + """ + # GIVEN a SecretsProvider instance + provider = parameters.SecretsProvider(config=config) + + # Stub the boto3 client + stubber = stub.Stubber(provider.client) + response = {"Name": mock_name, "ARN": f"arn:aws:secretsmanager:us-east-1:132456789012:secret/{mock_name}"} + expected_params = { + "SecretName": mock_name, + "SecretString": mock_value, + } + stubber.add_response("put_secret_value", response, expected_params) + stubber.activate() + + # WHEN cannot update a Secret with wrong parameter + # THEN raise SetSecretError + with pytest.raises(parameters.exceptions.SetSecretError, match="Error setting secret*"): + try: + assert response == provider.set(name=mock_name, value=mock_value) + stubber.assert_no_pending_responses() + finally: + stubber.deactivate() + + +def test_secret_provider_create_secret(mocker, mock_name, mock_value, config): + """ + Test Test SecretsProvider.set() forcing a new secret creation + """ + # GIVEN a SecretsProvider instance + provider = parameters.SecretsProvider(config=config) + + # WHEN the put_secret_value method raises a ResourceNotFoundException + mock_update_secret = mocker.patch.object(provider, "_update_secret") + mock_update_secret.side_effect = provider.client.exceptions.ResourceNotFoundException( + {"Error": {"Code": "ResourceNotFoundException"}}, + "put_secret_value", + ) + + # WHEN setting values for a new secret + client_request_token = str(uuid.uuid4()) + # Stub the boto3 client + stubber = stub.Stubber(provider.client) + response = {"Name": mock_name, "ARN": f"arn:aws:secretsmanager:us-east-1:132456789012:secret/{mock_name}"} + expected_params = { + "Name": mock_name, + "SecretString": mock_value, + "ClientRequestToken": client_request_token, + } + + # THEN it should call create_secret + stubber.add_response("create_secret", response, expected_params) + stubber.activate() + + try: + assert response == provider.set(name=mock_name, value=mock_value, client_request_token=client_request_token) + stubber.assert_no_pending_responses() + finally: + stubber.deactivate() + + +def test_secret_provider_create_secret_raise_on_error(mocker, mock_name, mock_value, config): + """ + Test Test SecretsProvider.set() forcing a new secret creation + """ + # GIVEN a SecretsProvider instance + provider = parameters.SecretsProvider(config=config) + + # WHEN the put_secret_value method raises a ResourceNotFoundException + mock_update_secret = mocker.patch.object(provider, "_update_secret") + mock_update_secret.side_effect = provider.client.exceptions.ResourceNotFoundException( + {"Error": {"Code": "ResourceNotFoundException"}}, + "put_secret_value", + ) + + # WHEN setting values for a new secret with wrong parameters + client_request_token = str(uuid.uuid4()) + # Stub the boto3 client + stubber = stub.Stubber(provider.client) + response = {"Name": mock_name, "ARN": f"arn:aws:secretsmanager:us-east-1:132456789012:secret/{mock_name}"} + expected_params = { + "NameSecret": mock_name, + "SecretString": mock_value, + "ClientRequestToken": client_request_token, + } + stubber.add_response("create_secret", response, expected_params) + stubber.activate() + + # WHEN cannot update a Secret with wrong parameter + # THEN raise SetSecretError + with pytest.raises(parameters.exceptions.SetSecretError, match="Error setting secret*"): + try: + assert response == provider.set(name=mock_name, value=mock_value) + stubber.assert_no_pending_responses() + finally: + stubber.deactivate() + + def test_ssm_provider_get_with_custom_client(mock_name, mock_value, mock_version, config): """ Test SSMProvider.get() with a non-cached value