88 lines
2.7 KiB
Python
88 lines
2.7 KiB
Python
import logging
|
|
import os
|
|
|
|
from frozendict import frozendict
|
|
|
|
logger = logging.getLogger(__name__)
|
|
logger.setLevel(logging.INFO)
|
|
|
|
try:
|
|
import hvac
|
|
except ImportError:
|
|
hvac = None
|
|
|
|
ENV_KEY_VAULT_URL = 'VAULT_URL'
|
|
ENV_KEY_VAULT_TOKEN = 'VAULT_TOKEN'
|
|
ENV_KEY_VAULT_MOUNT_POINT = 'VAULT_MOUNT_POINT'
|
|
|
|
|
|
class SecretLoader:
|
|
"""System for loading secrets from a variety of sources.
|
|
|
|
Priority order:
|
|
|
|
0. Hardcoded values. This is purely for prototyping.
|
|
1. Files pointed to by environment variables.
|
|
2. Secrets folder.
|
|
3. Vault instance if configured. Most suited for production environments.
|
|
"""
|
|
|
|
def __init__(self, env_key_prefix: str, hardcoded: dict[str, str] | None = None):
|
|
assert not env_key_prefix.endswith('_')
|
|
self.env_key_prefix = env_key_prefix
|
|
|
|
self.hardcoded: dict[str, str] = hardcoded if hardcoded is not None else {}
|
|
|
|
# Setup vault
|
|
self.vault_client = None
|
|
if hvac:
|
|
self.vault_client = hvac.Client(
|
|
url=self._load_or_none(ENV_KEY_VAULT_URL),
|
|
token=self._load_or_none(ENV_KEY_VAULT_TOKEN),
|
|
vault_mount_point=self._load_or_none(ENV_KEY_VAULT_MOUNT_POINT),
|
|
)
|
|
|
|
def load_or_fail(self, env_key: str) -> str:
|
|
value = self._load_or_none(env_key)
|
|
if value is None:
|
|
msg = f'Could not load secret {env_key}'
|
|
raise Exception(msg)
|
|
logger.info('Loaded secret: %s', env_key)
|
|
return value
|
|
|
|
def load(self, env_key: str) -> str | None:
|
|
value = self._load_or_none(env_key)
|
|
if value is None:
|
|
logger.exception('Could not load secret %s', env_key)
|
|
logger.info('Loaded secret: %s', env_key)
|
|
return value
|
|
|
|
def _load_or_none(self, env_key: str) -> str | None:
|
|
return (
|
|
self.hardcoded.get(env_key)
|
|
or self._load_or_none_path_or_file(env_key)
|
|
or self._load_or_none_vault(env_key)
|
|
)
|
|
|
|
def _load_or_none_path_or_file(self, env_key: str) -> str | None:
|
|
# 1. & 2.
|
|
filepath = os.environ.get(f'{self.env_key_prefix}_{env_key}')
|
|
|
|
if filepath is None:
|
|
filepath = f'./secrets/{env_key.lower()}'
|
|
try:
|
|
with open(filepath) as f:
|
|
return f.read().strip()
|
|
except Exception:
|
|
return None
|
|
|
|
def _load_or_none_vault(self, env_key: str) -> str | None:
|
|
if self.vault_client is None:
|
|
return None
|
|
|
|
read_secret_result = self.vault_client.secrets.kv.v1.read_secret(
|
|
path=env_key.lower(),
|
|
mount_point=self.vault_mount_point,
|
|
)
|
|
return read_secret_result['data']['value']
|