74 lines
2.2 KiB
Python
74 lines
2.2 KiB
Python
import logging
|
|
import os
|
|
|
|
logger = logging.getLogger(__name__)
|
|
_TRUTHY = {"1", "true", "yes", "on"}
|
|
|
|
class MissingCredentialError(RuntimeError):
|
|
"""Neither the environment variable nor the secret file provided a value."""
|
|
|
|
def env_flag(name:str, default: bool = False) -> bool:
|
|
value = os.getenv(name)
|
|
if value is None:
|
|
return default
|
|
return value.strip().lower() in _TRUTHY
|
|
|
|
def get_secret(env_var: str, path: str | None = None) -> str:
|
|
"""Resolve a secret from an environment variable, falling back to a file.
|
|
|
|
Resolution order:
|
|
1. The environment variable ``env_var`` (if set and non-empty).
|
|
2. The file at ``path`` (if given and it exists).
|
|
|
|
Raises:
|
|
MissingCredentialError: If neither source provides a value.
|
|
"""
|
|
value = os.getenv(env_var, "").strip()
|
|
if value:
|
|
logger.debug(f"Loaded secret from environment variable {env_var}")
|
|
return value
|
|
|
|
if path:
|
|
try:
|
|
return return_credentials(path)
|
|
except FileNotFoundError:
|
|
pass
|
|
|
|
sources = f"environment variable {env_var}" + (f" or file {path}" if path else "")
|
|
raise MissingCredentialError(f"No credential found: set {sources}")
|
|
|
|
def return_credentials(path: str) -> str:
|
|
"""Read and return credentials from a file."""
|
|
logger.debug(f"Opening credentials for path:{path}")
|
|
try:
|
|
with open(path) as file:
|
|
return file.read().strip()
|
|
except FileNotFoundError:
|
|
logger.fatal(f"[FATAL] Secret file not found: {path}")
|
|
raise
|
|
except Exception as e:
|
|
logger.fatal(f"[FATAL] Failed to read secret file {path}: {e}")
|
|
raise
|
|
|
|
|
|
def cleanup_secret_files(secret_paths: set[str], *, force: bool = False) -> None:
|
|
"""Delete secret files after they've been loaded into memory.
|
|
Skipped when KEEP_SECRET_FILES is set (e.g. in development), unless force=True.
|
|
|
|
Args:
|
|
secret_paths: Set of file paths to delete
|
|
"""
|
|
if not force and env_flag("KEEP_SECRET_FILES"):
|
|
logger.info("KEEP_SECRET_FILES is set, not deleting secret files")
|
|
return
|
|
|
|
for path in secret_paths:
|
|
try:
|
|
if os.path.exists(path):
|
|
os.remove(path)
|
|
logger.debug(f"Deleted secret file: {path}")
|
|
else:
|
|
logger.debug(f"Secret file already removed: {path}")
|
|
except Exception:
|
|
logger.exception(f"Could not delete secret file {path}")
|
|
raise |