feat(secrets): rewrite secrets.py for SOPS, update cli.py

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>
This commit is contained in:
Deeman
2026-02-26 10:44:25 +01:00
parent 9d0e6843f4
commit 6d716a83ae
2 changed files with 45 additions and 22 deletions

View File

@@ -1,44 +1,67 @@
"""Secrets management via Pulumi ESC."""
"""Secrets management via SOPS + age."""
import json
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() -> dict[str, str]:
"""Load secrets from Pulumi ESC environment."""
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(
["esc", "env", "open", "beanflows/prod", "--format", "json"],
["sops", "--input-type", "dotenv", "--output-type", "dotenv", "--decrypt", str(path)],
capture_output=True,
text=True,
check=True,
timeout=30,
)
data = json.loads(result.stdout)
return data.get("environmentVariables", {})
return _parse_dotenv(result.stdout)
except subprocess.CalledProcessError as e:
raise RuntimeError(f"Failed to load ESC environment: {e.stderr}")
raise RuntimeError(f"Failed to decrypt secrets: {e.stderr.strip()}")
except FileNotFoundError:
raise RuntimeError("ESC CLI not found. Install with: curl -fsSL https://get.pulumi.com/esc/install.sh | sh")
raise RuntimeError(
"sops not found. Install with: brew install sops "
"or see https://github.com/getsops/sops/releases"
)
def get_secret(key: str) -> str | None:
def get_secret(key: str, secrets_path: str = None) -> str | None:
"""Get a secret value by key."""
env = _load_environment()
env = _load_environment(secrets_path)
return env.get(key)
def list_secrets() -> list[str]:
def list_secrets(secrets_path: str = None) -> list[str]:
"""List all available secret keys."""
env = _load_environment()
env = _load_environment(secrets_path)
return list(env.keys())
def test_connection() -> bool:
"""Test ESC connection."""
def test_connection(secrets_path: str = None) -> bool:
"""Test that sops is available and can decrypt the secrets file."""
try:
_load_environment()
_load_environment.cache_clear()
_load_environment(secrets_path)
return True
except Exception:
return False