Source code for fictive.aws.secrets_manager.ext.flask

"""
Use AWS Secrets Manager to store sensitive Flask configuration values

"""

import json
import typing

import flask

import fictive.flask.extension
import fictive.aws.secrets_manager


[docs]class SecretsManager(fictive.flask.extension.AbstractFlaskExtension): """ Use AWS Secrets Manager to store sensitive Flask configuration values """ CLIENT_CLASS = fictive.aws.secrets_manager.SecretsManagerClient EXTENSION_KEY = 'secrets_manager' @property def default_secret_id(self) -> str: """ a default secret ID based on the current flask environment e.g., `development.flask_config` """ return f"{flask.helpers.get_env()}.flask_config"
[docs] def secret_id(self, app: flask.Flask) -> str: """ get the correct secret ID for `app` (using default if not set) """ appenv = self.appenv(app) return appenv.kwargs.get('secret_id', self.default_secret_id)
[docs] def init_app( self, app: flask.Flask, *args: typing.Any, **kwargs: typing.Any) -> None: """ load the secrets from AWS Secrets Manager and update `app.config` """ super().init_app(app, *args, **kwargs) client = self.get_client(app).client secret_id = self.secret_id(app) secret = client.get_secret_value(SecretId=secret_id) secret_value = json.loads(secret['SecretString']) # manually check for upper case keys and use `.update()` # (rather than `.from_mapping()`) to get around a quirk of # DynaConf # # see https://github.com/rochacbruno/dynaconf/pull/581 secret_value = { key: value for key, value in secret_value.items() if key.isupper() } app.config.update(secret_value)
[docs] def get_client( self, app: flask.Flask, *args: typing.Any, **kwargs: typing.Any) -> fictive.aws.secrets_manager.SecretsManagerClient: """ separate configuration kwargs for the cleitn from those for the extension """ appenv = self.appenv(app) nested_config = flask.Config(None) nested_config.update(appenv.kwargs) client_kwargs = nested_config.get('client_kwargs', {}) client_kwargs.update(nested_config.get_namespace('client_kwargs_')) return self.CLIENT_CLASS(*args, **client_kwargs)