"""Secrets Manager.""" import json import os import boto3 from flask.globals import current_app from secrets_manager.utils import create_or_update_secret class FlaskSecretsManager(object): def __init__(self, app=None, **kwargs): """Init. Args: app: This is a flask application instance. Keyword Args: application_context (bool): Set to False if you wish to use this without an application instance present. If you call init_app then the presence of an application context will override the Keyword Args. local_to_remote_name (dict): Local secret to remote secret map. service_name (str): The namespace of your secrets. aws_region (str): us-east-1 is the default. environment (str): dev/prod/qa. """ self.app = app self.boto3_session = boto3.session.Session() self.application_context = kwargs.get('application_context') self._options = kwargs if app is not None: self.init_app(app, **kwargs) def init_app(self, app, **kwargs): """Initialize the app.""" app.config.setdefault('ENVIRONMENT', 'dev') app.config.setdefault('AWS_REGION', 'us-east-1') app.config.setdefault('SERVICE_NAME', 'secretsmanager') self._options.update(**kwargs) if 'aws_region' in kwargs: app.config['AWS_REGION'] = kwargs['aws_region'] if 'environment' in kwargs: app.config['ENVIRONMENT'] = kwargs['environment'] if 'service_name' in kwargs: app.config['SERVICE_NAME'] = kwargs['service_name'] def get_client(self): """Get boto3 client. This will use default system profile or expect ENV vars to be present. Returns: boto3.client """ region_name = None if current_app: region_name = current_app.config['AWS_REGION'] elif self.application_context is False: region_name = self._options.get('region_name', 'us-east-1') if region_name: client = self.boto3_session.client( 'secretsmanager', region_name=region_name) return client def get_secret(self, secret_name): """Get a secret from AWS Secrets Manager. Returns: Credential (string) """ client = self.get_client() get_secret_value_response = client.get_secret_value( SecretId=secret_name) secret_string = get_secret_value_response['SecretString'] try: return json.loads(secret_string) except ValueError: pass return secret_string def store_cred(self, key, secret, **kwargs): """Store a new credential. Args: key (str): The name of the key in secret manager. secret (str/int): The string value of the secret. Keyword Args: *: Anything passed here is passed to boto3 create_secret method """ environment = None service_name = None if current_app: environment = current_app.config['ENVIRONMENT'] service_name = current_app.config['SERVICE_NAME'] elif self.application_context is False: environment = self._options.get('environment', 'dev') service_name = self._options.get('service_name') if environment and service_name: if environment.lower() in ['dev', 'test']: os.environ[key] = secret return client = self.get_client() secret_name = '{environment}/{service_name}/{secret_name}'.format( environment=environment, service_name=service_name, secret_name=key) if type(secret) in [str, int]: secret = {secret_name: secret} return create_or_update_secret(client, secret_name, secret, **kwargs) def get_cred(self, env_var): """Get a secret from secret manager. Args: env_var (str): Pulls from env in dev and secret manager in qa/prod. Returns: str or dict: The secret text if only one key pair is found. Otherwise it returns a dictionary of key/value pairs. """ environment = None service_name = None if current_app: environment = current_app.config['ENVIRONMENT'] service_name = current_app.config['SERVICE_NAME'] elif self.application_context is False: environment = self._options.get('environment', 'dev') service_name = self._options.get('service_name') if environment and service_name: force_remote = self._options.get('force_remote', False) if environment.lower() in ['dev', 'test'] and not force_remote: return os.environ.get(env_var) try: env_var = self._options['local_to_remote_name'][env_var] except KeyError: pass secret_name = '{environment}/{service_name}/{secret_name}'.format( environment=environment, service_name=service_name, secret_name=env_var) secret_text = self.get_secret(secret_name) if type(secret_text) is dict: if len(secret_text) == 1: return secret_text[secret_name] return secret_text