diff --git a/aws_lambda_powertools/utilities/parameters/__init__.py b/aws_lambda_powertools/utilities/parameters/__init__.py index 71f57ab7c08..d0628ca6e67 100644 --- a/aws_lambda_powertools/utilities/parameters/__init__.py +++ b/aws_lambda_powertools/utilities/parameters/__init__.py @@ -6,7 +6,7 @@ from .base import BaseProvider, clear_caches from .dynamodb import DynamoDBProvider from .exceptions import GetParameterError, TransformParameterError -from .secrets import SecretsProvider, get_secret, set_secret +from .secrets import SecretsProvider, get_secret, get_secrets_by_name, set_secret from .ssm import SSMProvider, get_parameter, get_parameters, get_parameters_by_name, set_parameter __all__ = [ @@ -23,6 +23,7 @@ "get_parameters", "get_parameters_by_name", "get_secret", + "get_secrets_by_name", "set_secret", "clear_caches", ] diff --git a/aws_lambda_powertools/utilities/parameters/exceptions.py b/aws_lambda_powertools/utilities/parameters/exceptions.py index 6a9554bf142..05945c837ed 100644 --- a/aws_lambda_powertools/utilities/parameters/exceptions.py +++ b/aws_lambda_powertools/utilities/parameters/exceptions.py @@ -7,6 +7,10 @@ class GetParameterError(Exception): """When a provider raises an exception on parameter retrieval""" +class GetSecretError(Exception): + """When a provider raises an exception on secret retrieval""" + + class TransformParameterError(Exception): """When a provider fails to transform a parameter value""" diff --git a/aws_lambda_powertools/utilities/parameters/secrets.py b/aws_lambda_powertools/utilities/parameters/secrets.py index 7fd35ce1d5f..359c84a80a7 100644 --- a/aws_lambda_powertools/utilities/parameters/secrets.py +++ b/aws_lambda_powertools/utilities/parameters/secrets.py @@ -8,16 +8,20 @@ import logging import os import warnings -from typing import TYPE_CHECKING, Literal, overload +from typing import TYPE_CHECKING, Any, Literal, overload import boto3 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 BaseProvider +from aws_lambda_powertools.utilities.parameters.base import BaseProvider, transform_value from aws_lambda_powertools.utilities.parameters.constants import DEFAULT_MAX_AGE_SECS, DEFAULT_PROVIDERS -from aws_lambda_powertools.utilities.parameters.exceptions import SetSecretError +from aws_lambda_powertools.utilities.parameters.exceptions import ( + GetSecretError, + SetSecretError, + TransformParameterError, +) from aws_lambda_powertools.warnings import PowertoolsDeprecationWarning if TYPE_CHECKING: @@ -126,11 +130,159 @@ def _get(self, name: str, **sdk_options) -> str | bytes: return secret_value["SecretBinary"] - def _get_multiple(self, path: str, **sdk_options) -> dict[str, str]: + def _get_multiple(self, names: list[str], **sdk_options) -> dict[str, Any]: # type: ignore[override] """ - Retrieving multiple parameter values is not supported with AWS Secrets Manager + Retrieve multiple secrets using AWS Secrets Manager batch_get_secret_value API + + Parameters + ---------- + names: list[str] + List of secret names to retrieve + sdk_options: dict, optional + Additional options passed to batch_get_secret_value API call + + Returns + ------- + dict[str, str] + Dictionary mapping secret names to their values + + Raises + ------ + GetParameterError + When the parameter provider fails to retrieve secrets """ - raise NotImplementedError() + + # Merge filters: combine names with any additional filters from sdk_options + filters = sdk_options.get("Filters", []) + name_filter = {"Key": "name", "Values": names} + + # Add name filter to existing filters + filters.append(name_filter) + sdk_options["Filters"] = filters + + # Remove SecretIdList if present to avoid conflicts + sdk_options.pop("SecretIdList", None) + + secrets: dict[str, Any] = {} + next_token = None + + # Handle pagination automatically + while True: + if next_token: + sdk_options["NextToken"] = next_token + elif "NextToken" in sdk_options: + # Remove NextToken from first call if it was passed + sdk_options.pop("NextToken") # pragma: no cover + + try: + response = self.client.batch_get_secret_value(**sdk_options) + except Exception as exc: + raise GetSecretError(f"Failed to retrieve secrets: {str(exc)}") from exc + + # Process successful secrets + for secret in response.get("SecretValues", []): + secret_name = secret["Name"] + + # Extract secret value (SecretString or SecretBinary) + if "SecretString" in secret: + secrets[secret_name] = secret["SecretString"] + elif "SecretBinary" in secret: + secrets[secret_name] = secret["SecretBinary"] + + # Check if there are more results + next_token = response.get("NextToken") + if not next_token: + break + + # If no secrets were found, raise an error + if not secrets: + raise GetSecretError(f"No secrets found matching the provided names: {names}") + + return secrets + + def get_multiple( # type: ignore[override] + self, + names: list[str], + max_age: int | None = None, + transform: TransformOptions = None, + raise_on_transform_error: bool = False, + force_fetch: bool = False, + **sdk_options, + ) -> dict[str, Any]: + """ + Retrieve multiple secrets by name from AWS Secrets Manager + + Parameters + ---------- + names: list[str] + List of secret names to retrieve + max_age: int, optional + Maximum age of the cached value + transform: str, optional + Optional transformation of the parameter value. Supported values + are "json" for JSON strings and "binary" for base 64 encoded values. + raise_on_transform_error: bool, optional + Raises an exception if any transform fails, otherwise this will + return a None value for each transform that failed + force_fetch: bool, optional + Force update even before a cached item has expired, defaults to False + sdk_options: dict, optional + Arguments that will be passed directly to the underlying API call + + Returns + ------- + dict[str, str | bytes | dict] + Dictionary mapping secret names to their values + + Raises + ------ + GetParameterError + When the parameter provider fails to retrieve secrets + TransformParameterError + When the parameter provider fails to transform a secret value + """ + if not names: + raise GetSecretError("You must provide at least one secret name") + + # Create a unique cache key for this batch of secrets + # Use sorted names to ensure consistent caching regardless of order + cache_key_name = "|".join(sorted(names)) + key = self._build_cache_key(name=cache_key_name, transform=transform, is_nested=True) + + # If max_age is not set, resolve it from the environment variable, defaulting to DEFAULT_MAX_AGE_SECS + max_age = resolve_max_age(env=os.getenv(constants.PARAMETERS_MAX_AGE_ENV, DEFAULT_MAX_AGE_SECS), choice=max_age) + + if not force_fetch and self.has_not_expired_in_cache(key): + cached_values = self.fetch_from_cache(key) + # Return only the requested secrets from cache (in case cache has more) + return {name: cached_values[name] for name in names if name in cached_values} + + try: + values = self._get_multiple(names, **sdk_options) + except Exception as exc: + raise GetSecretError(str(exc)) from exc + + if transform: + # Transform each secret value + transformed_values = {} + for name, value in values.items(): + try: + transformed_values[name] = transform_value( + key=name, + value=value, + transform=transform, + raise_on_transform_error=raise_on_transform_error, + ) + except TransformParameterError: + if raise_on_transform_error: + raise + transformed_values[name] = None # pragma: no cover + values = transformed_values + + # Cache the results + self.add_to_cache(key=key, value=values, max_age=max_age) + + return values def _create_secret(self, name: str, **sdk_options) -> CreateSecretResponseTypeDef: """ @@ -369,6 +521,85 @@ def get_secret( ) +def get_secrets_by_name( + names: list[str], + transform: TransformOptions = None, + force_fetch: bool = False, + max_age: int | None = None, + **sdk_options, +) -> dict[str, str | bytes | dict]: + """ + Retrieve multiple secrets by name from AWS Secrets Manager + + Parameters + ---------- + names: list[str] + List of secret names to retrieve + transform: str, optional + Transforms the content from a JSON object ('json') or base64 binary string ('binary') + force_fetch: bool, optional + Force update even before a cached item has expired, defaults to False + max_age: int, optional + Maximum age of the cached value + sdk_options: dict, optional + Dictionary of options that will be passed to the batch_get_secret_value call + + Raises + ------ + GetParameterError + When the parameter provider fails to retrieve secrets + TransformParameterError + When the parameter provider fails to transform a secret value + + Returns + ------- + dict[str, str | bytes | dict] + Dictionary mapping secret names to their values + + Example + ------- + **Retrieves multiple secrets** + + >>> from aws_lambda_powertools.utilities.parameters import get_secrets_by_name + >>> + >>> secrets = get_secrets_by_name(["db-password", "api-key", "jwt-secret"]) + >>> print(secrets["db-password"]) + + **Retrieves multiple secrets with JSON transformation** + + >>> from aws_lambda_powertools.utilities.parameters import get_secrets_by_name + >>> + >>> secrets = get_secrets_by_name(["config", "settings"], transform="json") + >>> print(secrets["config"]["database_url"]) + + **Retrieves multiple secrets with additional filters** + + >>> from aws_lambda_powertools.utilities.parameters import get_secrets_by_name + >>> + >>> secrets = get_secrets_by_name( + ... names=["app-secret"], + ... Filters=[{"Key": "primary-region", "Values": ["us-east-1"]}] + ... ) + """ + if not names: + raise GetSecretError("You must provide at least one secret name") + + # If max_age is not set, resolve it from the environment variable, defaulting to DEFAULT_MAX_AGE_SECS + max_age = resolve_max_age(env=os.getenv(constants.PARAMETERS_MAX_AGE_ENV, DEFAULT_MAX_AGE_SECS), choice=max_age) + + # 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"].get_multiple( + names=names, + max_age=max_age, + transform=transform, + force_fetch=force_fetch, + **sdk_options, + ) + + def set_secret( name: str, value: str | bytes, diff --git a/docs/utilities/parameters.md b/docs/utilities/parameters.md index e20d42611ae..42487cbcee3 100644 --- a/docs/utilities/parameters.md +++ b/docs/utilities/parameters.md @@ -36,6 +36,7 @@ This utility requires additional permissions to work as expected. | 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) | +| Secrets | **`get_secrets_by_name`**, **`SecretsProvider.get_multiple`** | **`secretsmanager:BatchGetSecretValue`**, **`secretsmanager:GetSecretValue`** and **`secretsmanager:ListSecrets`** | | DynamoDB | **`DynamoDBProvider.get`** | **`dynamodb:GetItem`** | | DynamoDB | **`DynamoDBProvider.get_multiple`** | **`dynamodb:Query`** | | AppConfig | **`get_app_config`**, **`AppConfigProvider.get_app_config`** | **`appconfig:GetLatestConfiguration`** and **`appconfig:StartConfigurationSession`** | @@ -111,6 +112,30 @@ You can fetch secrets stored in Secrets Manager using `get_secret`. --8<-- "examples/parameters/src/getting_started_secret.py" ``` +### Fetching multiple secrets + +You can fetch multiple secrets from Secrets Manager in a single API call using `get_secrets_by_name`. This reduces the number of API calls and improves performance when you need to retrieve several secrets at once. + +???+ info "Batch retrieval benefits" + - **Performance**: Retrieve up to 20 secrets in one API call + - **Cost optimization**: Fewer API calls reduce AWS costs + - **Error resilience**: Partial failures don't break the entire operation + - **Advanced filtering**: Use additional filters beyond secret names + +=== "getting_started_batch_secrets.py" + ```python hl_lines="1 7" + --8<-- "examples/parameters/src/getting_started_batch_secrets.py" + ``` + +#### Advanced filtering + +You can combine secret name filtering with additional AWS Secrets Manager filters for more precise results: + +=== "batch_secrets_with_filters.py" + ```python hl_lines="2 10-16" + --8<-- "examples/parameters/src/batch_secrets_with_filters.py" + ``` + ### Setting secrets You can set secrets stored in Secrets Manager using `set_secret`. @@ -251,6 +276,11 @@ You can create `SecureString` parameters, which are parameters that have a plain --8<-- "examples/parameters/src/builtin_provider_secret.py" ``` +=== "batch_secrets_provider.py" + ```python hl_lines="2-12" + --8<-- "examples/parameters/src/batch_secrets_provider.py" + ``` + #### DynamoDBProvider The DynamoDB Provider does not have any high-level functions, as it needs to know the name of the DynamoDB table containing the parameters. @@ -445,7 +475,9 @@ Here is the mapping between this utility's functions and methods and the underly | SSM Parameter Store | `SSMProvider.get` | `ssm` | [get_parameter](https://boto3.amazonaws.com/v1/documentation/api/latest/reference/services/ssm.html#SSM.Client.get_parameter){target="_blank"} | | SSM Parameter Store | `SSMProvider.get_multiple` | `ssm` | [get_parameters_by_path](https://boto3.amazonaws.com/v1/documentation/api/latest/reference/services/ssm.html#SSM.Client.get_parameters_by_path){target="_blank"} | | Secrets Manager | `get_secret` | `secretsmanager` | [get_secret_value](https://boto3.amazonaws.com/v1/documentation/api/latest/reference/services/secretsmanager.html#SecretsManager.Client.get_secret_value){target="_blank"} | +| Secrets Manager | `get_secrets_by_name` | `secretsmanager` | [batch_get_secret_value](https://boto3.amazonaws.com/v1/documentation/api/latest/reference/services/secretsmanager.html#SecretsManager.Client.batch_get_secret_value){target="_blank"} | | Secrets Manager | `SecretsProvider.get` | `secretsmanager` | [get_secret_value](https://boto3.amazonaws.com/v1/documentation/api/latest/reference/services/secretsmanager.html#SecretsManager.Client.get_secret_value){target="_blank"} | +| Secrets Manager | `SecretsProvider.get_multiple` | `secretsmanager` | [batch_get_secret_value](https://boto3.amazonaws.com/v1/documentation/api/latest/reference/services/secretsmanager.html#SecretsManager.Client.batch_get_secret_value){target="_blank"} | | DynamoDB | `DynamoDBProvider.get` | `dynamodb` | ([Table resource](https://boto3.amazonaws.com/v1/documentation/api/latest/reference/services/dynamodb.html#table){target="_blank"}) | | DynamoDB | `DynamoDBProvider.get_multiple` | `dynamodb` | ([Table resource](https://boto3.amazonaws.com/v1/documentation/api/latest/reference/services/dynamodb.html#table){target="_blank"}) | | App Config | `get_app_config` | `appconfigdata` | [start_configuration_session](https://boto3.amazonaws.com/v1/documentation/api/latest/reference/services/appconfigdata.html#AppConfigData.Client.start_configuration_session){target="_blank"} and [get_latest_configuration](https://boto3.amazonaws.com/v1/documentation/api/latest/reference/services/appconfigdata.html#AppConfigData.Client.get_latest_configuration){target="_blank"} | diff --git a/examples/parameters/src/batch_secrets_provider.py b/examples/parameters/src/batch_secrets_provider.py new file mode 100644 index 00000000000..9ed5cc51bfc --- /dev/null +++ b/examples/parameters/src/batch_secrets_provider.py @@ -0,0 +1,31 @@ +from aws_lambda_powertools import Logger +from aws_lambda_powertools.utilities.parameters import SecretsProvider +from aws_lambda_powertools.utilities.typing import LambdaContext + +logger = Logger() +# Create provider instance for more control +secrets_provider = SecretsProvider() + + +def lambda_handler(event, context: LambdaContext): + # Retrieve secrets with custom settings + secrets = secrets_provider.get_multiple( + names=["service/auth-token", "service/encryption-key"], + max_age=600, # Cache for 10 minutes + transform="json", # Parse JSON secrets + raise_on_transform_error=False, # Don't fail on transform errors + ) + + # Handle potential transform failures + auth_token = secrets.get("service/auth-token") + encryption_key = secrets.get("service/encryption-key") + + if auth_token is None: + logger.info("Warning: auth-token failed to parse as JSON") + if encryption_key is None: + logger.info("Warning: encryption-key failed to parse as JSON") + + return { + "statusCode": 200, + "body": f"Retrieved {len([s for s in secrets.values() if s is not None])} valid secrets", + } diff --git a/examples/parameters/src/batch_secrets_with_filters.py b/examples/parameters/src/batch_secrets_with_filters.py new file mode 100644 index 00000000000..611d7590d5b --- /dev/null +++ b/examples/parameters/src/batch_secrets_with_filters.py @@ -0,0 +1,25 @@ +from aws_lambda_powertools import Logger +from aws_lambda_powertools.utilities import parameters +from aws_lambda_powertools.utilities.typing import LambdaContext + +logger = Logger() + + +def lambda_handler(event, context: LambdaContext): + # Retrieve secrets with additional filtering + production_secrets = parameters.get_secrets_by_name( + names=["app-secret", "db-secret"], + Filters=[ + {"Key": "primary-region", "Values": ["us-east-1"]}, + {"Key": "tag-value", "Values": ["production"]}, + ], + ) + + # Only secrets matching ALL filters will be returned + for name, _ in production_secrets.items(): + logger.info(f"Found production secret: {name}") + + return { + "statusCode": 200, + "body": f"Retrieved {len(production_secrets)} production secrets", + } diff --git a/examples/parameters/src/getting_started_batch_secrets.py b/examples/parameters/src/getting_started_batch_secrets.py new file mode 100644 index 00000000000..d574a9468c3 --- /dev/null +++ b/examples/parameters/src/getting_started_batch_secrets.py @@ -0,0 +1,31 @@ +from aws_lambda_powertools.utilities import parameters +from aws_lambda_powertools.utilities.typing import LambdaContext + + +def lambda_handler(event, context: LambdaContext): + # Retrieve multiple secrets in a single API call + secrets = parameters.get_secrets_by_name( + [ + "database/password", + "api/key", + "jwt/secret", + ], + ) + + # Access individual secrets + db_password = secrets["database/password"] + api_key = secrets["api/key"] + jwt_secret = secrets["jwt/secret"] + + do_stuff_with_secrets(db_password, api_key, jwt_secret) + + # Use secrets in your application logic + return { + "statusCode": 200, + "body": f"Retrieved {len(secrets)} secrets successfully", + } + + +def do_stuff_with_secrets(db_password, api_key, jwt_secret): + """Do your business logic""" + pass diff --git a/tests/functional/parameters/_boto3/test_utilities_parameters.py b/tests/functional/parameters/_boto3/test_utilities_parameters.py index f7b7a642e00..da9422ea839 100644 --- a/tests/functional/parameters/_boto3/test_utilities_parameters.py +++ b/tests/functional/parameters/_boto3/test_utilities_parameters.py @@ -23,6 +23,7 @@ BaseProvider, ExpirableValue, ) +from aws_lambda_powertools.utilities.parameters.exceptions import GetSecretError from aws_lambda_powertools.warnings import PowertoolsDeprecationWarning @@ -2978,3 +2979,481 @@ def test_raise_warning_when_using_config_parameter_secrets(config): # THEN must raise a warning with pytest.warns(PowertoolsDeprecationWarning, match="The 'config' parameter is deprecated in V3*"): SecretsProvider(config=config) + + +def test_secrets_provider_get_multiple_basic(config): + """ + Test SecretsProvider.get_multiple() with basic functionality + """ + # GIVEN a SecretsProvider instance + provider = parameters.SecretsProvider(boto_config=config) + + # WHEN calling get_multiple with a list of secret names + secret_names = ["db-password", "api-key"] + + # Stub the boto3 client + stubber = stub.Stubber(provider.client) + response = { + "SecretValues": [ + { + "ARN": "arn:aws:secretsmanager:us-east-1:132456789012:secret/db-password", + "Name": "db-password", + "VersionId": "7a9155b8-2dc9-466e-b4f6-5bc46516c84d", + "SecretString": "super-secret-password", + "CreatedDate": datetime(2015, 1, 1), + }, + { + "ARN": "arn:aws:secretsmanager:us-east-1:132456789012:secret/api-key", + "Name": "api-key", + "VersionId": "8b9266c9-3ed0-577f-c5g7-6cd57627d95e", + "SecretString": "xxxxxx", + "CreatedDate": datetime(2015, 1, 1), + }, + ], + "Errors": [], + } + expected_params = {"Filters": [{"Key": "name", "Values": secret_names}]} + stubber.add_response("batch_get_secret_value", response, expected_params) + stubber.activate() + + try: + # THEN it should return a dictionary with secret names and values + result = provider.get_multiple(secret_names) + + expected = {"db-password": "super-secret-password", "api-key": "xxxxxx"} + assert result == expected + stubber.assert_no_pending_responses() + finally: + stubber.deactivate() + + +def test_secrets_provider_get_multiple_with_binary(config): + """ + Test SecretsProvider.get_multiple() with binary secrets + """ + # GIVEN a SecretsProvider instance + provider = parameters.SecretsProvider(boto_config=config) + + # WHEN calling get_multiple with secrets containing binary data + secret_names = ["binary-secret", "string-secret"] + + # Stub the boto3 client + stubber = stub.Stubber(provider.client) + response = { + "SecretValues": [ + { + "ARN": "arn:aws:secretsmanager:us-east-1:132456789012:secret/binary-secret", + "Name": "binary-secret", + "VersionId": "7a9155b8-2dc9-466e-b4f6-5bc46516c84d", + "SecretBinary": b"binary-data", + "CreatedDate": datetime(2015, 1, 1), + }, + { + "ARN": "arn:aws:secretsmanager:us-east-1:132456789012:secret/string-secret", + "Name": "string-secret", + "VersionId": "8b9266c9-3ed0-577f-c5g7-6cd57627d95e", + "SecretString": "string-data", + "CreatedDate": datetime(2015, 1, 1), + }, + ], + "Errors": [], + } + expected_params = {"Filters": [{"Key": "name", "Values": secret_names}]} + stubber.add_response("batch_get_secret_value", response, expected_params) + stubber.activate() + + try: + # THEN it should return both binary and string secrets correctly + result = provider.get_multiple(secret_names) + + expected = {"binary-secret": b"binary-data", "string-secret": "string-data"} + assert result == expected + stubber.assert_no_pending_responses() + finally: + stubber.deactivate() + + +def test_secrets_provider_get_multiple_with_pagination(config): + """ + Test SecretsProvider.get_multiple() with pagination + """ + # GIVEN a SecretsProvider instance + provider = parameters.SecretsProvider(boto_config=config) + + # WHEN calling get_multiple with results that require pagination + secret_names = ["secret-1", "secret-2", "secret-3"] + + # Stub the boto3 client for first page + stubber = stub.Stubber(provider.client) + + # First page response + first_response = { + "SecretValues": [ + { + "ARN": "arn:aws:secretsmanager:us-east-1:132456789012:secret/secret-1", + "Name": "secret-1", + "VersionId": "7a9155b8-2dc9-466e-b4f6-5bc46516c84d", + "SecretString": "value-1", + "CreatedDate": datetime(2015, 1, 1), + }, + { + "ARN": "arn:aws:secretsmanager:us-east-1:132456789012:secret/secret-2", + "Name": "secret-2", + "VersionId": "8b9266c9-3ed0-577f-c5g7-6cd57627d95e", + "SecretString": "value-2", + "CreatedDate": datetime(2015, 1, 1), + }, + ], + "NextToken": "next-page-token", + "Errors": [], + } + first_expected_params = {"Filters": [{"Key": "name", "Values": secret_names}]} + + # Second page response + second_response = { + "SecretValues": [ + { + "ARN": "arn:aws:secretsmanager:us-east-1:132456789012:secret/secret-3", + "Name": "secret-3", + "VersionId": "9c0377da-4fe1-688g-d6h8-7de68738e06f", + "SecretString": "value-3", + "CreatedDate": datetime(2015, 1, 1), + }, + ], + "Errors": [], + } + second_expected_params = {"Filters": [{"Key": "name", "Values": secret_names}], "NextToken": "next-page-token"} + + stubber.add_response("batch_get_secret_value", first_response, first_expected_params) + stubber.add_response("batch_get_secret_value", second_response, second_expected_params) + stubber.activate() + + try: + # THEN it should return all secrets from both pages + result = provider.get_multiple(secret_names) + + expected = {"secret-1": "value-1", "secret-2": "value-2", "secret-3": "value-3"} + assert result == expected + stubber.assert_no_pending_responses() + finally: + stubber.deactivate() + + +def test_secrets_provider_get_multiple_with_errors(config): + """ + Test SecretsProvider.get_multiple() with some secrets failing + """ + # GIVEN a SecretsProvider instance + provider = parameters.SecretsProvider(boto_config=config) + + # WHEN calling get_multiple with some secrets that don't exist + secret_names = ["good-secret", "bad-secret"] + + # Stub the boto3 client + stubber = stub.Stubber(provider.client) + response = { + "SecretValues": [ + { + "ARN": "arn:aws:secretsmanager:us-east-1:132456789012:secret/good-secret", + "Name": "good-secret", + "VersionId": "7a9155b8-2dc9-466e-b4f6-5bc46516c84d", + "SecretString": "good-value", + "CreatedDate": datetime(2015, 1, 1), + }, + ], + "Errors": [{"SecretId": "bad-secret", "ErrorCode": "ResourceNotFoundException", "Message": "Secret not found"}], + } + expected_params = {"Filters": [{"Key": "name", "Values": secret_names}]} + stubber.add_response("batch_get_secret_value", response, expected_params) + stubber.activate() + + try: + # THEN it should return only successful secrets and log errors + result = provider.get_multiple(secret_names) + + expected = {"good-secret": "good-value"} + assert result == expected + stubber.assert_no_pending_responses() + finally: + stubber.deactivate() + + +def test_secrets_provider_get_multiple_with_additional_filters(config): + """ + Test SecretsProvider.get_multiple() with additional filters + """ + # GIVEN a SecretsProvider instance + provider = parameters.SecretsProvider(boto_config=config) + + # WHEN calling get_multiple with additional filters + secret_names = ["filtered-secret"] + additional_filters = [{"Key": "primary-region", "Values": ["us-east-1"]}] + + # Stub the boto3 client + stubber = stub.Stubber(provider.client) + response = { + "SecretValues": [ + { + "ARN": "arn:aws:secretsmanager:us-east-1:132456789012:secret/filtered-secret", + "Name": "filtered-secret", + "VersionId": "7a9155b8-2dc9-466e-b4f6-5bc46516c84d", + "SecretString": "filtered-value", + "CreatedDate": datetime(2015, 1, 1), + }, + ], + "Errors": [], + } + expected_params = { + "Filters": [{"Key": "primary-region", "Values": ["us-east-1"]}, {"Key": "name", "Values": secret_names}], + } + stubber.add_response("batch_get_secret_value", response, expected_params) + stubber.activate() + + try: + # THEN it should merge filters correctly + result = provider.get_multiple(secret_names, Filters=additional_filters) + + expected = {"filtered-secret": "filtered-value"} + assert result == expected + stubber.assert_no_pending_responses() + finally: + stubber.deactivate() + + +def test_secrets_provider_get_multiple_with_json_transform(config): + """ + Test SecretsProvider.get_multiple() with JSON transformation + """ + # GIVEN a SecretsProvider instance + provider = parameters.SecretsProvider(boto_config=config) + + # WHEN calling get_multiple with JSON transform + secret_names = ["json-secret", "plain-secret"] + + # Stub the boto3 client + stubber = stub.Stubber(provider.client) + response = { + "SecretValues": [ + { + "ARN": "arn:aws:secretsmanager:us-east-1:132456789012:secret/json-secret", + "Name": "json-secret", + "VersionId": "7a9155b8-2dc9-466e-b4f6-5bc46516c84d", + "SecretString": '{"key": "value", "number": 42}', + "CreatedDate": datetime(2015, 1, 1), + }, + { + "ARN": "arn:aws:secretsmanager:us-east-1:132456789012:secret/plain-secret", + "Name": "plain-secret", + "VersionId": "8b9266c9-3ed0-577f-c5g7-6cd57627d95e", + "SecretString": "plain-text", + "CreatedDate": datetime(2015, 1, 1), + }, + ], + "Errors": [], + } + expected_params = {"Filters": [{"Key": "name", "Values": secret_names}]} + stubber.add_response("batch_get_secret_value", response, expected_params) + stubber.activate() + + try: + # THEN it should transform JSON secrets and handle failures gracefully + result = provider.get_multiple(secret_names, transform="json") + + expected = { + "json-secret": {"key": "value", "number": 42}, + "plain-secret": None, # Transform failure should return None + } + assert result == expected + stubber.assert_no_pending_responses() + finally: + stubber.deactivate() + + +def test_secrets_provider_get_multiple_empty_names(config): + """ + Test SecretsProvider.get_multiple() with empty names list + """ + # GIVEN a SecretsProvider instance + provider = parameters.SecretsProvider(boto_config=config) + + # WHEN calling get_multiple with empty names list + # THEN it should raise GetParameterError + with pytest.raises(GetSecretError, match="You must provide at least one secret name"): + provider.get_multiple([]) + + +def test_secrets_provider_non_existing_key(config): + """ + Test SecretsProvider.get_multiple() with additional filters + """ + # GIVEN a SecretsProvider instance + provider = parameters.SecretsProvider(boto_config=config) + + # WHEN calling get_multiple with additional filters that doesnt + secret_names = ["filtered-secret"] + additional_filters = [{"Key": "error-region", "Values": ["us-east-1"]}] + + # Stub the boto3 client + stubber = stub.Stubber(provider.client) + response = { + "SecretValues": [ + { + "ARN": "arn:aws:secretsmanager:us-east-1:132456789012:secret/filtered-secret", + "Name": "filtered-secret", + "VersionId": "7a9155b8-2dc9-466e-b4f6-5bc46516c84d", + "SecretString": "filtered-value", + "CreatedDate": datetime(2015, 1, 1), + }, + ], + "Errors": [], + } + expected_params = { + "Filters": [{"Key": "primary-region", "Values": ["us-east-1"]}, {"Key": "name", "Values": secret_names}], + } + stubber.add_response("batch_get_secret_value", response, expected_params) + stubber.activate() + + # THEN it should raise an exception + with pytest.raises(GetSecretError, match="Failed to retrieve secrets*"): + provider.get_multiple(secret_names, Filters=additional_filters) + + +def test_secrets_provider_get_multiple_caching(config): + """ + Test SecretsProvider.get_multiple() caching behavior + """ + # GIVEN a SecretsProvider instance + provider = parameters.SecretsProvider(boto_config=config) + + # WHEN calling get_multiple multiple times + secret_names = ["cached-secret"] + + # Stub the boto3 client for first call only + stubber = stub.Stubber(provider.client) + response = { + "SecretValues": [ + { + "ARN": "arn:aws:secretsmanager:us-east-1:132456789012:secret/cached-secret", + "Name": "cached-secret", + "VersionId": "7a9155b8-2dc9-466e-b4f6-5bc46516c84d", + "SecretString": "cached-value", + "CreatedDate": datetime(2015, 1, 1), + }, + ], + "Errors": [], + } + expected_params = {"Filters": [{"Key": "name", "Values": secret_names}]} + stubber.add_response("batch_get_secret_value", response, expected_params) + stubber.activate() + + try: + # First call - should hit API + result1 = provider.get_multiple(secret_names, max_age=300) + expected = {"cached-secret": "cached-value"} + assert result1 == expected + + # Second call - should use cache (no additional API call) + result2 = provider.get_multiple(secret_names, max_age=300) + assert result2 == expected + + stubber.assert_no_pending_responses() + finally: + stubber.deactivate() + + +def test_get_secrets_by_name_helper(monkeypatch, mock_name, mock_value): + """ + Test get_secrets_by_name helper function + """ + + # GIVEN a mocked SecretsProvider + class TestProvider: + def get_multiple(self, names, **kwargs): + return {name: f"{mock_value}-{name}" for name in names} + + monkeypatch.setattr(parameters.secrets, "DEFAULT_PROVIDERS", {}) + monkeypatch.setattr(parameters.secrets, "SecretsProvider", TestProvider) + + # WHEN calling get_secrets_by_name + secret_names = ["helper-secret-1", "helper-secret-2"] + result = parameters.get_secrets_by_name(secret_names) + + # THEN it should return the expected values + expected = {"helper-secret-1": f"{mock_value}-helper-secret-1", "helper-secret-2": f"{mock_value}-helper-secret-2"} + assert result == expected + + +def test_get_secrets_by_name_empty_names(): + """ + Test get_secrets_by_name with empty names list + """ + # WHEN calling get_secrets_by_name with empty list + # THEN it should raise GetSecretError + with pytest.raises(GetSecretError, match="You must provide at least one secret name"): + parameters.get_secrets_by_name([]) + + +def test_secrets_provider_get_multiple_no_secrets_found(config): + """ + Test SecretsProvider.get_multiple() when no secrets are found + """ + # GIVEN a SecretsProvider instance + provider = parameters.SecretsProvider(boto_config=config) + + # WHEN calling get_multiple with names that don't match any secrets + secret_names = ["non-existent-secret"] + + # Stub the boto3 client to return empty results + stubber = stub.Stubber(provider.client) + response = {"SecretValues": [], "Errors": []} + expected_params = {"Filters": [{"Key": "name", "Values": secret_names}]} + stubber.add_response("batch_get_secret_value", response, expected_params) + stubber.activate() + + try: + # THEN it should raise GetSecretError + with pytest.raises(GetSecretError, match="No secrets found matching the provided names"): + provider.get_multiple(secret_names) + + stubber.assert_no_pending_responses() + finally: + stubber.deactivate() + + +def test_secrets_provider_get_multiple_with_json_transform_error(config): + """ + Test SecretsProvider.get_multiple() with JSON transformation + """ + # GIVEN a SecretsProvider instance + provider = parameters.SecretsProvider(boto_config=config) + + # WHEN calling get_multiple with JSON transform + secret_names = ["json-secret", "plain-secret"] + + # Stub the boto3 client + stubber = stub.Stubber(provider.client) + response = { + "SecretValues": [ + { + "ARN": "arn:aws:secretsmanager:us-east-1:132456789012:secret/json-secret", + "Name": "json-secret", + "VersionId": "7a9155b8-2dc9-466e-b4f6-5bc46516c84d", + "SecretString": '{"key": "value", "number": 42}', + "CreatedDate": datetime(2015, 1, 1), + }, + { + "ARN": "arn:aws:secretsmanager:us-east-1:132456789012:secret/plain-secret", + "Name": "plain-secret", + "VersionId": "8b9266c9-3ed0-577f-c5g7-6cd57627d95e", + "SecretString": "plain-text", + "CreatedDate": datetime(2015, 1, 1), + }, + ], + "Errors": [], + } + expected_params = {"Filters": [{"Key": "name", "Values": secret_names}]} + stubber.add_response("batch_get_secret_value", response, expected_params) + stubber.activate() + + with pytest.raises(parameters.TransformParameterError): + provider.get_multiple(secret_names, transform="binary", raise_on_transform_error=True)