[GSoC2020] Secrets Manager

README.md

Secrets Manager - Django GSoC

The modules documented in this gist present the crux of my idea for Secrets Manager for the Django framework. The complete prototype can be view and played around with here.

Design

env.py (included in this gist: env.py)

env.py resides in the env/ folder in the Django root, the directory where settings.py resides in a new project.
This is the entry point for a developer to register secrets sources with the SecretsManager. An object of class SecretsManager is instantiated here. Any environment secrets sources can be registered here to be used later anywhere in the project.

settings.py (included in this gist: settings.py)

The developer will retrieve the created SecretsManager object (singleton, so developer doesn't have to do much). get_secrets()/get_secrets(env_name) is triggered as per requirement to get secrets from the appropriate source as a dict.

secrets_manager.py (included in this gist: secrets_manager.py)

In this prototype, secrets_manager.py resides in secrets_manager package. However, it will reside in the Django codebase later.
The SecretsManager class is an instance of Singleton. Singleton is a design pattern that ensures that only one object is created throughout the project life cycle. This pattern facilitates the use of a SecretsManager instantiated once, to be used throughout the project in any other module.
The SecretsManager's instantiating parameters define how the SecretsManager is constructed:

1. default_env_name defines the default environment secrets source to be returned by default.
   • --env=$env_name passed to manage.py can override this
   • env_name passed to get_secrets(env_name) can overide this
2. lazy_loading defines if secrets sources are read when registered or when requested. (True by default)
3. auto_register defines if SecretsManager should automatically register .env and python module secret sources present in the env/ directory (False by default)
   • The naming convention for auto registered environment sources in the prototype can be viewed in SecretsManager's instance method auto_register(). This convention is only for demonstration purposes and will be perfected later.
   • during automatic registration, if there is a source file that contains 'base' in it's name, that source will be set as the base.

The SecretsManager's instance methods allow the developer work the required magic:

1. register(env_name, src, auto_reload=False, payload=None, headers=None, Auth=None) allows registration of a secrets source with the secrets manager.
2. set_base(env_name) allows the developer to set base environment secrets, pulling from a registered source.
3. auto_register() can be triggered by the developer in case (s)he wants to trigger automatic registration inspite of setting auto_register=False while instantiating SecretsManager.
4. get_secrets(env_name=None) returns secrets of the specified environment, if env_name is passed. It env_name is not passed, Returns secrets of environment name passed to manage.py using --env argument. If --env is not passed, SecretsManager's default environment's secrets are returned. A dict is returned to the calling module.
5. reload_secrets(env_name) allows developer to force reload/re-read of secrets associated with env_name even if the source was registered with auto_reload=False
6. unregister(env_name) allows the developer to unregister a previously registered secrets source.

secrets.py (included in this gist: secrets.py)

In this prototype, secrets.py resides in secrets_manager module. However, it will reside in the Django codebase later.
The SecretsAbstract class is an instance of SingletonByArgument. SingletonByArgument is a design pattern that ensures that only one object is created for a particular combination of registration parameters. DotEnvSecrets, ModuleSecrets, HTTPSecrets implement SecretsAbstract and implement the abstract method read_secrets() and read corresponding types of secret sources.
HTTPSecrets in particular needs more parameters, payload, headers and Auth.

1. Auth is the type of authentication, if required, for an API based secrets source. HTTPSecrets will inspect this authentication type's constructor and request CLI input for the required parameters.

env.py

from requests.auth import HTTPDigestAuth

from secrets_manager.secrets_manager import SecretsManager

secrets_manager = SecretsManager(
    default_env_name="http_get_with_auth_dev", auto_register=True
)

# Not required as auto_register=True
# To manually register .env and modules as secret sources and set_base()
#
# secrets_manager.register("module_base", "base.py")
# secrets_manager.set_base("module_base")
# secrets_manager.register("dot_env_dev", ".DEV")
# secrets_manager.register("dot_env_prod", ".PROD")
# secrets_manager.register("dot_env_test", ".TEST")
# secrets_manager.register("module_dev", "DEV.py")
# secrets_manager.register("module_prod", "PROD.py")
# secrets_manager.register("module_test", "TEST.py")

secrets_manager.register(
    "http_get_dev", "http://127.0.0.1:8080/secrets/DEV/", auto_reload=True
)
secrets_manager.register(
    "http_get_prod", "http://127.0.0.1:8080/secrets/PROD/", auto_reload=True
)
secrets_manager.register(
    "http_get_test", "http://127.0.0.1:8080/secrets/TEST/", auto_reload=False
)
secrets_manager.register(
    "http_post_dev",
    "http://127.0.0.1:8080/secrets/",
    payload='{"env_name":"DEV"}',
    auto_reload=True,
)
secrets_manager.register(
    "http_post_prod",
    "http://127.0.0.1:8080/secrets/",
    payload='{"env_name":"PROD"}',
    auto_reload=True,
)
secrets_manager.register(
    "http_post_test",
    "http://127.0.0.1:8080/secrets/",
    payload='{"env_name":"TEST"}',
    auto_reload=False,
)

secrets_manager.register(
    "http_get_with_auth_dev",
    "http://127.0.0.1:8080/secrets/auth/DEV",
    auto_reload=True,
    Auth=HTTPDigestAuth,
)
secrets_manager.register(
    "http_get_with_auth_prod",
    "http://127.0.0.1:8080/secrets/auth/PROD",
    auto_reload=True,
    Auth=HTTPDigestAuth,
)
secrets_manager.register(
    "http_get_with_auth_test",
    "http://127.0.0.1:8080/secrets/auth/TEST",
    auto_reload=False,
    Auth=HTTPDigestAuth,
)

secrets.py

import abc
import importlib
import inspect

import requests
from dotenv import dotenv_values


class SingletonByArgument(type, metaclass=abc.ABCMeta):
    _instances = {}

    def __call__(cls, *args, **kwargs):
        if (args, tuple(kwargs.keys()), tuple(kwargs.values())) not in cls._instances:
            cls._instances[
                (args, tuple(kwargs.keys()), tuple(kwargs.values()))
            ] = super(SingletonByArgument, cls).__call__(*args, **kwargs)
        return cls._instances[(args, tuple(kwargs.keys()), tuple(kwargs.values()))]


class SecretsAbstract(metaclass=SingletonByArgument):
    def __init__(
        self,
        env_name=None,
        src=None,
        auto_reload=False,
        payload=None,
        headers=None,
        Auth=None,
    ):
        self.env_name = env_name
        self.src = src
        self.auto_reload = auto_reload
        self.payload = payload
        self.headers = headers
        self.Auth = Auth
        self.secrets = None
        self.auth_parameters = None
        self.reload()

    def reload(self):
        self.secrets = self.read_secrets()

    @abc.abstractmethod
    def read_secrets(self):
        """implement in subclass"""


class DotEnvSecrets(SecretsAbstract):
    def read_secrets(self):
        secrets = dotenv_values(self.src)
        return secrets


class ModuleSecrets(SecretsAbstract):
    def read_secrets(self):
        spec = importlib.util.spec_from_file_location(
            self.src.split("/")[-1][:-3], self.src
        )
        mod = importlib.util.module_from_spec(spec)
        spec.loader.exec_module(mod)
        secrets = {
            key: vars(mod)[key] for key in vars(mod).keys() if not key.startswith("__")
        }
        return secrets


class HTTPSecrets(SecretsAbstract):
    def auth_parameters_CLI(self):
        parameters = list(inspect.signature(self.Auth.__init__).parameters)
        parameters.remove("self")
        print(
            self.Auth.__name__
            + " Authentication defined for environment '"
            + self.env_name
            + "'.\n"
            + "Secrets Source Endpoint: "
            + self.src
        )
        kwargs = {}
        for parameter in parameters:
            value = input("Enter " + parameter + ":")
            kwargs[parameter] = value
        return kwargs

    def read_secrets(self):
        if self.auth_parameters is None and self.Auth is not None:
            self.auth_parameters = self.auth_parameters_CLI()
            print()
        if self.payload is None:
            response = requests.get(
                self.src,
                auth=self.Auth(**self.auth_parameters)
                if self.auth_parameters is not None
                else None,
            )
        else:
            response = requests.post(
                self.src,
                data=self.payload,
                headers=self.headers,
                auth=self.Auth(**self.auth_parameters)
                if self.auth_parameters is not None
                else None,
            )
        secrets = response.json()
        return secrets

secrets_manager.py

import os
from pathlib import Path

from django.core.validators import URLValidator
from secrets_manager.secrets import DotEnvSecrets, ModuleSecrets, HTTPSecrets


class Singleton(type):
    _instances = {}

    def __call__(cls, *args, **kwargs):
        if cls not in cls._instances:
            cls._instances[cls] = super(Singleton, cls).__call__(*args, **kwargs)
        return cls._instances[cls]


class SecretsManager(metaclass=Singleton):
    def __init__(self, default_env_name=None, lazy_loading=True, auto_register=False):
        self.env = {}
        self.env_configs = {}
        try:
            self.deploy_env = os.environ["env"]
        except KeyError:
            self.deploy_env = default_env_name

        root_path = str(Path(__file__).resolve(strict=True).parents[1])
        project_name = os.path.basename(root_path)
        self.env_dir = root_path + "/" + project_name + "/env/"
        self.is_base_set = False
        self.lazy_loading = lazy_loading
        self.env_secrets = {}
        self.base_env_name = None
        if auto_register:
            self.auto_register()

    def auto_register(self):
        file_names = list(os.walk(self.env_dir))[0][2]
        file_names = [
            file_name
            for file_name in file_names
            if file_name not in ["env.py", "__init__.py"]
        ]
        for file_name in file_names:
            env_name = "dot_env_" + file_name[1:]
            if file_name.endswith(".py"):
                env_name = "module_" + file_name[:-3]
            self.register(env_name, file_name)
            if "base" in env_name:
                self.set_base(env_name)

    def set_base(self, env_name):
        if env_name in self.env_configs:
            self.is_base_set = True
            self.base_env_name = env_name

    def register(
        self, env_name, src, auto_reload=False, payload=None, headers=None, Auth=None
    ):
        if self.deploy_env is None:
            self.deploy_env = env_name
        self.env_configs[env_name] = (
            env_name,
            src,
            auto_reload,
            payload,
            headers,
            Auth,
        )
        if not self.lazy_loading:
            self.load_secrets(env_name)

    def load_secrets(self, env_name):
        if env_name not in self.env_secrets:
            _, src, auto_reload, _, _, _ = self.env_configs[env_name]
            try:
                validate_url = URLValidator()
                validate_url(src)
            except:
                if src.endswith(".py"):
                    self.env_secrets[env_name] = ModuleSecrets(
                        env_name, self.env_dir + src, auto_reload
                    )
                else:
                    self.env_secrets[env_name] = DotEnvSecrets(
                        env_name, self.env_dir + src, auto_reload
                    )
            else:
                self.env_secrets[env_name] = HTTPSecrets(*self.env_configs[env_name])

        return self.env_secrets[env_name]

    def get_secrets(self, env_name=None):
        env_name = self.deploy_env if env_name is None else env_name
        env = self.load_secrets(env_name)
        if env.auto_reload:
            self.reload_secrets(env=env)
        secrets = env.secrets
        if self.is_base_set and env_name is not self.base_env_name:
            all_secrets = self.get_secrets(self.base_env_name)
            all_secrets.update(secrets)
            return all_secrets
        return secrets

    def reload_secrets(self, env_name=None, env=None):
        if env_name is not None and env_name in self.env_configs:
            env = self.load_secrets(env_name)
        if env is not None:
            env.reload()

    def unregister(self, env_name):
        if env_name in self.env_configs:
            env = self.env_secrets.pop(env_name, None)
            if env is not None:
                del env
            del self.env_configs[env_name]
        if env_name == self.base_env_name:
            self.base_env_name = None
            self.is_base_set = False

settings.py

from secrets_manager.secrets_manager import SecretsManager
from secrets_manager_prototype.env import env

# get the environment secrets as per --env passed to manage.py
# or default environment secrets if --env not passed
secrets = SecretsManager().get_secrets()

# or the developer can do this to always get environment secrets for dev environment
# secrets = SecretsManager().get_secrets('dev')

# all the secrets are returned in a dict with keys specified in the source
SECRET_KEY = secrets["SECRET_KEY"]
DEBUG = secrets["DEBUG"]

# and so on...