"""AWS Secret Manager connector.""" from functools import lru_cache import json import boto3 from feature_fm import config @lru_cache(1000) def get_secret(): """Get shared token, either from local cache or from AWS. Returns: string: with the shared token or None """ return get_aws_secret() def get_aws_secret(): """Get shared secret value from AWS. Returns: string: with the shared token or None """ secret_name = config.SECRET_TOKEN endpoint_url = config.AWS_SECRET_MANAGER_URL region_name = config.AWS_REGION session = boto3.session.Session() client = session.client( service_name='secretsmanager', region_name=region_name, endpoint_url=endpoint_url ) get_secret_value_response = client.get_secret_value(SecretId=secret_name) # Decrypted secret using the associated KMS CMK # Depending on whether the secret was a string or binary, # one of these fields will be populated if 'SecretString' in get_secret_value_response: return json.loads(get_secret_value_response['SecretString']) else: return json.loads(get_secret_value_response['SecretBinary'])