import base64 from typing import Optional import boto3 from botocore.exceptions import ClientError from cachetools import cached from structlog import get_logger from delphi_api.const import AWS_DEFAULT_REGION from delphi_api.core.caches import ClientsListCache LOG = get_logger() @cached(ClientsListCache) def get_secret(secret_name: str, region_name: str = AWS_DEFAULT_REGION) -> Optional[str]: """Makes a request to AWS Secrets Manager via the 'GetSecretValue' API. See Also: https://docs.aws.amazon.com/secretsmanager/latest/apireference/API_GetSecretValue.html Args: secret_name: Name of the secret (foo/bar/etc) region_name: AWS region Returns: The decrypted and decoded secret value, or ``None`` if there was an error. """ # Create a Secrets Manager client session = boto3.session.Session() client = session.client(service_name='secretsmanager', region_name=region_name) error_codes = { 'DecryptionFailureException': 'Secrets Manager cannot decrypt the protected secret text using the provided KMS key.', 'InternalServiceErrorException': 'An error occurred on the server side.', 'InvalidParameterException': 'You provided an invalid value for a parameter.', 'InvalidRequestException': 'You provided a param value that is not valid for the current state of the resource.', 'ResourceNotFoundException': 'We cannot find the resource that you asked for.', } try: get_secret_value_response = client.get_secret_value(SecretId=secret_name) except ClientError as e: try: code = e.response['Error']['Code'] except KeyError: # pragma: no cover code = 'UnknownException' LOG.exception(e, description='GetSecretValue failed.', secret_name=secret_name, aws_code=code, aws_description=error_codes.get(code, '')) # we are not going to rethrow the error here as this implementation is currently just for # tracing; we don't want the user to receive an error response. return # Decrypts secret using the associated KMS CMK. # Depending on whether the secret is a string or binary, one of these fields will be populated. if 'SecretString' in get_secret_value_response: return get_secret_value_response['SecretString'] return base64.b64decode(get_secret_value_response['SecretBinary'])