6d716a83ae
secrets.py: replace Pulumi ESC (esc CLI) with SOPS decrypt. Reads .env.prod.sops via `sops --decrypt`, parses dotenv output. Same public API: get_secret(), list_secrets(), test_connection(). cli.py: update secrets subcommand help text and test command messaging. Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
68 lines
2.2 KiB
Python
68 lines
2.2 KiB
Python
"""Secrets management via SOPS + age."""
|
|
|
|
import subprocess
|
|
from functools import lru_cache
|
|
from pathlib import Path
|
|
|
|
# Default secrets file path (relative to repo root)
|
|
_DEFAULT_SECRETS_PATH = Path(__file__).parent.parent.parent / ".env.prod.sops"
|
|
|
|
|
|
def _parse_dotenv(text: str) -> dict[str, str]:
|
|
"""Parse dotenv-format text into a dict, skipping comments and blanks."""
|
|
result = {}
|
|
for line in text.splitlines():
|
|
line = line.strip()
|
|
if not line or line.startswith("#"):
|
|
continue
|
|
if "=" not in line:
|
|
continue
|
|
key, _, value = line.partition("=")
|
|
result[key.strip()] = value.strip()
|
|
return result
|
|
|
|
|
|
@lru_cache(maxsize=1)
|
|
def _load_environment(secrets_path: str = None) -> dict[str, str]:
|
|
"""Decrypt and load secrets from a SOPS-encrypted dotenv file."""
|
|
path = Path(secrets_path) if secrets_path else _DEFAULT_SECRETS_PATH
|
|
assert path.exists(), f"Secrets file not found: {path}"
|
|
try:
|
|
result = subprocess.run(
|
|
["sops", "--input-type", "dotenv", "--output-type", "dotenv", "--decrypt", str(path)],
|
|
capture_output=True,
|
|
text=True,
|
|
check=True,
|
|
timeout=30,
|
|
)
|
|
return _parse_dotenv(result.stdout)
|
|
except subprocess.CalledProcessError as e:
|
|
raise RuntimeError(f"Failed to decrypt secrets: {e.stderr.strip()}")
|
|
except FileNotFoundError:
|
|
raise RuntimeError(
|
|
"sops not found. Install with: brew install sops "
|
|
"or see https://github.com/getsops/sops/releases"
|
|
)
|
|
|
|
|
|
def get_secret(key: str, secrets_path: str = None) -> str | None:
|
|
"""Get a secret value by key."""
|
|
env = _load_environment(secrets_path)
|
|
return env.get(key)
|
|
|
|
|
|
def list_secrets(secrets_path: str = None) -> list[str]:
|
|
"""List all available secret keys."""
|
|
env = _load_environment(secrets_path)
|
|
return list(env.keys())
|
|
|
|
|
|
def test_connection(secrets_path: str = None) -> bool:
|
|
"""Test that sops is available and can decrypt the secrets file."""
|
|
try:
|
|
_load_environment.cache_clear()
|
|
_load_environment(secrets_path)
|
|
return True
|
|
except Exception:
|
|
return False
|