Implemented hardcoded
This commit is contained in:
parent
a2b1551d4b
commit
66f82967b0
|
@ -1,5 +1,6 @@
|
||||||
import logging
|
import logging
|
||||||
import os
|
import os
|
||||||
|
from frozendict import frozendict
|
||||||
|
|
||||||
logger = logging.getLogger(__name__)
|
logger = logging.getLogger(__name__)
|
||||||
logger.setLevel(logging.INFO)
|
logger.setLevel(logging.INFO)
|
||||||
|
@ -9,33 +10,39 @@ try:
|
||||||
except ImportError:
|
except ImportError:
|
||||||
hvac = None
|
hvac = None
|
||||||
|
|
||||||
|
ENV_KEY_VAULT_URL = 'VAULT_URL'
|
||||||
|
ENV_KEY_VAULT_TOKEN = 'VAULT_TOKEN'
|
||||||
|
ENV_KEY_VAULT_MOUNT_POINT = 'VAULT_MOUNT_POINT'
|
||||||
|
|
||||||
class SecretLoader:
|
class SecretLoader:
|
||||||
'''
|
"""
|
||||||
Priority order:
|
Priority order:
|
||||||
|
|
||||||
1. Files pointed to by environment variables
|
0. Hardcoded values. This is purely for prototyping.
|
||||||
2. Secrets folder
|
1. Files pointed to by environment variables.
|
||||||
3. Vault instance if configured
|
2. Secrets folder.
|
||||||
'''
|
3. Vault instance if configured. Most suited for production environments.
|
||||||
|
"""
|
||||||
|
|
||||||
def __init__(self, env_key_prefix: str):
|
def __init__(self, env_key_prefix: str, hardcoded: dict[str,str] | None = None):
|
||||||
assert not env_key_prefix.endswith('_')
|
assert not env_key_prefix.endswith('_')
|
||||||
self.env_key_prefix = env_key_prefix
|
self.env_key_prefix = env_key_prefix
|
||||||
|
|
||||||
|
self.hardcoded: dict[str,str] = hardcoded if hardcoded is not None else {}
|
||||||
|
|
||||||
# Setup vault
|
# Setup vault
|
||||||
self.vault_client = None
|
self.vault_client = None
|
||||||
if hvac:
|
if hvac:
|
||||||
self.vault_client = hvac.Client(
|
self.vault_client = hvac.Client(
|
||||||
url=self._load_or_none('VAULT_URL'),
|
url=self._load_or_none(ENV_KEY_VAULT_URL),
|
||||||
token=self._load_or_none('VAULT_TOKEN'),
|
token=self._load_or_none(ENV_KEY_VAULT_TOKEN),
|
||||||
|
vault_mount_point=self._load_or_none(ENV_KEY_VAULT_MOUNT_POINT),
|
||||||
)
|
)
|
||||||
self.vault_mount_point=self._load_or_none('VAULT_MOUNT_POINT')
|
|
||||||
|
|
||||||
def load_or_fail(self, env_key: str) -> str:
|
def load_or_fail(self, env_key: str) -> str:
|
||||||
value = self._load_or_none(env_key)
|
value = self._load_or_none(env_key)
|
||||||
if value is None:
|
if value is None:
|
||||||
msg = 'Could not load secret {}'.format(env_key)
|
msg = f'Could not load secret {env_key}'
|
||||||
raise Exception(msg)
|
raise Exception(msg)
|
||||||
logger.info('Loaded secret: %s', env_key)
|
logger.info('Loaded secret: %s', env_key)
|
||||||
return value
|
return value
|
||||||
|
@ -48,7 +55,7 @@ class SecretLoader:
|
||||||
return value
|
return value
|
||||||
|
|
||||||
def _load_or_none(self, env_key: str) -> str | None:
|
def _load_or_none(self, env_key: str) -> str | None:
|
||||||
return self._load_or_none(env_key)
|
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:
|
def _load_or_none_path_or_file(self, env_key: str) -> str | None:
|
||||||
# 1. & 2.
|
# 1. & 2.
|
|
@ -2,5 +2,5 @@
|
||||||
import secret_loader
|
import secret_loader
|
||||||
|
|
||||||
def test_init():
|
def test_init():
|
||||||
secret_loader.SecretLoader('TEST')
|
loader = secret_loader.SecretLoader('TEST', hardcoded = {'KEY': 'VALUE'})
|
||||||
|
assert loader.load('KEY') == 'VALUE'
|
||||||
|
|
Loading…
Reference in New Issue
Block a user