"""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