bedrock.external.secrets_handler

 1import re
 2import json
 3import traceback
 4import base64
 5from botocore.exceptions import ClientError
 6from bedrock.config import get_secrets, set_secrets, get_config_params
 7from bedrock.log import log_config
 8from bedrock.external.aws import make_client
 9
10log = log_config("SecretsHandlerEndpoint")
11SECRETS_MANAGER = make_client('secretsmanager')
12
13
14def find_secret(secret_arn) -> dict or None:  # pragma: integration
15    """
16    Find a secret in the secrets manager and returns its content.
17    Wraps `get_secret` to return None if the secret is not found.
18
19    :param secret_arn: The secret ARN
20    """
21    config = get_config_params()
22    if config["secrets"][secret_arn]:  # pragma: no cover - No testing AWS secrets
23        try:
24            return _get_secret(config["secrets"][secret_arn])
25        except:
26            log.warning(f"Unable to fetch secret {secret_arn}")
27            return None
28    return None
29
30
31def find_secrets_like(config_secret_name_pattern: str) -> list:  # pragma: integration
32    """
33    Finds secrets that match the given pattern in the config.
34
35    :param config_secret_name_pattern: The pattern to match a secret that was defined in the config.
36    """
37    config = get_config_params()
38    secrets = [find_secret(key) for key in config["secrets"] if re.match(config_secret_name_pattern, key)]
39    if not secrets:
40        log.warning(f"Unable to fetch secret with pattern {config_secret_name_pattern}")
41        return []
42    return secrets
43
44
45def _get_secret(secret_arn: str) -> dict:  # pragma: unit
46    """
47    Find a secret in the secrets manager and returns its content.
48
49    :param secret_arn:
50    :return:
51    """
52    secrets = get_secrets()
53
54    if secret_arn is None:
55        raise ValueError("No secret name defined")
56
57    if secret_arn in secrets:
58        log.debug(f"Getting {secret_arn} from secrets temporary cache")
59        return secrets[secret_arn]
60
61    log.debug(f"Extracting secrets from {secret_arn}")
62    try:
63        secrets_value = SECRETS_MANAGER.get_secret_value(SecretId=secret_arn)
64    except ClientError as e:  # pragma: no cover
65        log.error(f"Unable to get secret '{secret_arn}': {e.response['Error']['Code']}")
66        raise e
67    except Exception as e:  # pragma: no cover
68        log.error(f"Unknown error while getting secret '{secret_arn}'")
69        log.debug(f"    {traceback.format_exc()}")
70        raise e
71
72    log.debug(f"Loading secret values...")
73    try:
74        secret = json.loads(secrets_value['SecretString'])
75    except KeyError:
76        secret = json.loads(base64.b64decode(secrets_value['SecretBinary']))
77
78    log.debug(f"Got secret for {secret_arn}!")
79    set_secrets({**secrets, secret_arn: secret})
80
81    return secret