WIP pass
support
This commit is contained in:
parent
f876cb87c5
commit
463f92167e
|
@ -1,5 +1,6 @@
|
||||||
import logging
|
import logging
|
||||||
import os
|
import os
|
||||||
|
import subprocess
|
||||||
|
|
||||||
from frozendict import frozendict
|
from frozendict import frozendict
|
||||||
|
|
||||||
|
@ -15,26 +16,37 @@ ENV_KEY_VAULT_URL = 'VAULT_URL'
|
||||||
ENV_KEY_VAULT_TOKEN = 'VAULT_TOKEN'
|
ENV_KEY_VAULT_TOKEN = 'VAULT_TOKEN'
|
||||||
ENV_KEY_VAULT_MOUNT_POINT = 'VAULT_MOUNT_POINT'
|
ENV_KEY_VAULT_MOUNT_POINT = 'VAULT_MOUNT_POINT'
|
||||||
|
|
||||||
|
ENV_KEY_PASS_FOLDER = 'PASS_FOLDER'
|
||||||
|
|
||||||
class SecretLoader:
|
class SecretLoader:
|
||||||
"""System for loading secrets from a variety of sources.
|
"""System for loading secrets from a variety of sources.
|
||||||
|
|
||||||
Priority order:
|
Priority order:
|
||||||
|
|
||||||
0. Hardcoded values. This is purely for prototyping.
|
0. Hardcoded values. **This is purely for debugging and prototyping.**
|
||||||
1. Files pointed to by environment variables.
|
1. Files pointed to by environment variables. Docker friendly.
|
||||||
2. Secrets folder.
|
2. Secrets folder. Also Docker friendly.
|
||||||
3. Vault instance if configured. Most suited for production environments.
|
3. [Pass: the standard unix password
|
||||||
|
manager](https://www.passwordstore.org/). Most suited for personal
|
||||||
|
usage; very unsuited for server environments. Requires `pass` installed
|
||||||
|
locally, and configuration of the `PASS_FOLDER` through one of the above
|
||||||
|
methods.
|
||||||
|
4. Vault instance if configured. Suited for production environments.
|
||||||
"""
|
"""
|
||||||
|
|
||||||
def __init__(self, env_key_prefix: str, hardcoded: dict[str, str] | None = None):
|
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 {}
|
self.hardcoded: dict[str, str] = hardcoded if hardcoded is not None else {}
|
||||||
|
self.pass_folder = None
|
||||||
|
self.vault_client = None
|
||||||
|
|
||||||
|
# Setup pass
|
||||||
|
if pass_folder := self._load_or_none(ENV_KEY_PASS_FOLDER):
|
||||||
|
self.pass_folder = pass_folder
|
||||||
|
del pass_folder
|
||||||
|
|
||||||
# Setup vault
|
# Setup vault
|
||||||
self.vault_client = None
|
|
||||||
if hvac:
|
if hvac:
|
||||||
self.vault_client = hvac.Client(
|
self.vault_client = hvac.Client(
|
||||||
url=self._load_or_none(ENV_KEY_VAULT_URL),
|
url=self._load_or_none(ENV_KEY_VAULT_URL),
|
||||||
|
@ -61,6 +73,7 @@ class SecretLoader:
|
||||||
return (
|
return (
|
||||||
self.hardcoded.get(env_key)
|
self.hardcoded.get(env_key)
|
||||||
or self._load_or_none_path_or_file(env_key)
|
or self._load_or_none_path_or_file(env_key)
|
||||||
|
or self._load_or_none_local_password_store(env_key)
|
||||||
or self._load_or_none_vault(env_key)
|
or self._load_or_none_vault(env_key)
|
||||||
)
|
)
|
||||||
|
|
||||||
|
@ -76,7 +89,19 @@ class SecretLoader:
|
||||||
except Exception:
|
except Exception:
|
||||||
return None
|
return None
|
||||||
|
|
||||||
|
def _load_or_none_local_password_store(self, env_key: str) -> str | None:
|
||||||
|
if self.pass_folder is None:
|
||||||
|
return None
|
||||||
|
|
||||||
|
cmd = ['pass', 'show', f'{self.pass_folder}/{env_key.lower()}']
|
||||||
|
process = subprocess.run(cmd, capture_output = True, check = True)
|
||||||
|
print(process)
|
||||||
|
|
||||||
|
return process.stdout.decode('utf8')
|
||||||
|
|
||||||
def _load_or_none_vault(self, env_key: str) -> str | None:
|
def _load_or_none_vault(self, env_key: str) -> str | None:
|
||||||
|
print(self.hardcoded)
|
||||||
|
print(self.vault_client)
|
||||||
if self.vault_client is None:
|
if self.vault_client is None:
|
||||||
return None
|
return None
|
||||||
|
|
||||||
|
|
|
@ -4,3 +4,7 @@ import secret_loader
|
||||||
def test_init():
|
def test_init():
|
||||||
loader = secret_loader.SecretLoader('TEST', hardcoded={'KEY': 'VALUE'})
|
loader = secret_loader.SecretLoader('TEST', hardcoded={'KEY': 'VALUE'})
|
||||||
assert loader.load('KEY') == 'VALUE'
|
assert loader.load('KEY') == 'VALUE'
|
||||||
|
|
||||||
|
def test_lookup_unknown():
|
||||||
|
loader = secret_loader.SecretLoader('TEST', hardcoded={})
|
||||||
|
assert loader.load('UNKNOWN') is None
|
||||||
|
|
Loading…
Reference in New Issue
Block a user