Developer
94 lines
3.2 KiB
Python
94 lines
3.2 KiB
Python
"""Encryption-at-rest for service secrets.
|
|
|
|
Service API keys / tokens are stored encrypted in the ``services.secrets_json``
|
|
column. Encryption uses Fernet (symmetric authenticated encryption) with a single
|
|
master key provided via the ``MANAGE_ENCRYPTION_KEY`` environment variable.
|
|
|
|
* The key **must** be a urlsafe base64-encoded 32-byte value (Fernet format).
|
|
* The key is **always required** — there is no development fallback, so secrets
|
|
are never accidentally stored in plaintext.
|
|
* Secrets are encrypted field-by-field; the ``"which secrets are set"`` metadata
|
|
can be derived from the ciphertext blob without decrypting.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import os
|
|
from functools import lru_cache
|
|
|
|
from cryptography.fernet import Fernet, InvalidToken
|
|
|
|
ENCRYPTION_KEY_ENV = "MANAGE_ENCRYPTION_KEY"
|
|
|
|
|
|
class EncryptionKeyError(RuntimeError):
|
|
"""Raised when the encryption key is missing or invalid."""
|
|
|
|
|
|
@lru_cache(maxsize=1)
|
|
def get_encryption_key() -> bytes:
|
|
"""Return the raw Fernet key, or raise if missing/invalid.
|
|
|
|
The result is cached for the process lifetime. Tests should call
|
|
:func:`reset_encryption_key_cache` after changing the environment.
|
|
"""
|
|
raw = os.environ.get(ENCRYPTION_KEY_ENV)
|
|
if not raw:
|
|
raise EncryptionKeyError(f"{ENCRYPTION_KEY_ENV} is required to store service secrets")
|
|
key = raw.strip().encode()
|
|
try:
|
|
Fernet(key)
|
|
except (ValueError, TypeError) as exc: # pragma: no cover - validated by tests
|
|
raise EncryptionKeyError(f"{ENCRYPTION_KEY_ENV} must be a valid Fernet key") from exc
|
|
return key
|
|
|
|
|
|
def reset_encryption_key_cache() -> None:
|
|
"""Drop the cached encryption key (used by tests that swap keys)."""
|
|
get_encryption_key.cache_clear()
|
|
|
|
|
|
def _fernet() -> Fernet:
|
|
return Fernet(get_encryption_key())
|
|
|
|
|
|
def encrypt_value(plaintext: str) -> str:
|
|
"""Encrypt a single secret value and return the ciphertext string."""
|
|
return _fernet().encrypt(plaintext.encode()).decode()
|
|
|
|
|
|
def decrypt_value(ciphertext: str) -> str:
|
|
"""Decrypt a single ciphertext value."""
|
|
try:
|
|
return _fernet().decrypt(ciphertext.encode()).decode()
|
|
except InvalidToken as exc:
|
|
raise EncryptionKeyError("Service secret could not be decrypted") from exc
|
|
|
|
|
|
def encrypt_secrets(values: dict[str, str]) -> dict[str, str]:
|
|
"""Encrypt every provided secret value."""
|
|
fernet = _fernet()
|
|
return {key: fernet.encrypt(value.encode()).decode() for key, value in values.items()}
|
|
|
|
|
|
def decrypt_secrets(blob: dict[str, str]) -> dict[str, str]:
|
|
"""Decrypt every secret value in a blob."""
|
|
fernet = _fernet()
|
|
result: dict[str, str] = {}
|
|
for key, ciphertext in blob.items():
|
|
try:
|
|
result[key] = fernet.decrypt(ciphertext.encode()).decode()
|
|
except InvalidToken as exc:
|
|
raise EncryptionKeyError(f"Service secret '{key}' could not be decrypted") from exc
|
|
return result
|
|
|
|
|
|
def generate_development_key() -> str:
|
|
"""Return a freshly generated Fernet key (helper for operators/docs)."""
|
|
return Fernet.generate_key().decode()
|
|
|
|
|
|
def validate_encryption_key() -> None:
|
|
"""Eagerly validate that the encryption key is present and well-formed."""
|
|
get_encryption_key() # raises EncryptionKeyError on failure
|