bvandeusen
4df98171ab
Audit 2026-06-02: `_load_or_create_key` silently minted a new Fernet key whenever the key file was missing — no log, no warning. The failure mode the audit flagged: a partial disaster restore where the DB was restored but `/images/secrets/` was lost would produce a working-looking system in which every authenticated download fails AUTH_ERROR until the operator re-uploads every credential by hand. Two opt-ins now needed for auto-creation: 1. Explicit `bootstrap_ok=True` kwarg (tests, scripts), OR 2. `CURATOR_BOOTSTRAP_NEW_KEY=1` env var (operator first-time setup) Otherwise the constructor raises `MissingCredentialKey` so the app fails fast at startup and the operator can restore the key file from backup before encrypted_blob rows go undecryptable. Also: docstring path was wrong (said "images/data root" but actual location is `/images/secrets/credential_key.b64`) — corrected. Tests updated to pass `bootstrap_ok=True` explicitly, and two new tests cover the safety behavior (missing-key-raises, env-var-bootstraps).
98 lines
3.9 KiB
Python
98 lines
3.9 KiB
Python
"""Fernet-based encryption for credential blobs.
|
|
|
|
The key is a single 32-byte value (urlsafe-base64-encoded; what
|
|
Fernet.generate_key produces) stored at /images/secrets/credential_key.b64
|
|
(mode 0600, parent dir 0700). The 2026-06-02 audit caught a silent
|
|
key-regeneration path: on a partial disaster restore where the DB was
|
|
restored but the secrets dir was lost, the old `_load_or_create_key`
|
|
would mint a fresh key with no log, producing a working-looking system
|
|
where every authenticated download failed AUTH_ERROR until the operator
|
|
re-uploaded every credential by hand. Now the constructor refuses to
|
|
auto-generate unless either:
|
|
|
|
* the caller explicitly passes `bootstrap_ok=True` (tests, scripts), or
|
|
* the env var `CURATOR_BOOTSTRAP_NEW_KEY=1` is set (operator opt-in
|
|
during first-time setup).
|
|
|
|
Otherwise it raises `MissingCredentialKey` so the app fails fast at
|
|
startup and the operator can restore the key file from backup.
|
|
|
|
Operator backup procedure must include /images/secrets/ alongside the
|
|
rest of /images/ — losing the key file makes existing encrypted_blob
|
|
rows undecryptable (recovery = delete the rows and re-upload).
|
|
"""
|
|
|
|
import logging
|
|
import os
|
|
from pathlib import Path
|
|
|
|
from cryptography.fernet import Fernet, InvalidToken
|
|
|
|
log = logging.getLogger(__name__)
|
|
|
|
_BOOTSTRAP_ENV_VAR = "CURATOR_BOOTSTRAP_NEW_KEY"
|
|
|
|
|
|
class InvalidCredentialBlob(Exception):
|
|
"""Raised when decryption fails (wrong key, tampered blob, …)."""
|
|
|
|
|
|
class MissingCredentialKey(Exception):
|
|
"""The Fernet key file is missing AND the caller hasn't opted in to
|
|
generating a new one. Audit 2026-06-02: prevents silent key
|
|
regeneration on partial DB-restored / secrets-lost deployments.
|
|
Set CURATOR_BOOTSTRAP_NEW_KEY=1 for first-time setup, or restore the
|
|
key file from backup."""
|
|
|
|
|
|
class CredentialCrypto:
|
|
"""Fernet encrypt/decrypt with an on-disk key file.
|
|
|
|
Instantiate with a path; the file is loaded if present, or created
|
|
if absent AND the caller has opted in (bootstrap_ok=True or
|
|
CURATOR_BOOTSTRAP_NEW_KEY=1 env var). Production sites:
|
|
`IMAGES_ROOT / "secrets" / "credential_key.b64"`.
|
|
"""
|
|
|
|
def __init__(self, key_path: Path, *, bootstrap_ok: bool | None = None):
|
|
self._key_path = Path(key_path)
|
|
if bootstrap_ok is None:
|
|
bootstrap_ok = os.environ.get(_BOOTSTRAP_ENV_VAR) == "1"
|
|
self._fernet = Fernet(self._load_or_create_key(bootstrap_ok))
|
|
|
|
def _load_or_create_key(self, bootstrap_ok: bool) -> bytes:
|
|
if self._key_path.exists():
|
|
return self._key_path.read_bytes()
|
|
if not bootstrap_ok:
|
|
raise MissingCredentialKey(
|
|
f"Fernet key file not found at {self._key_path}. "
|
|
f"For first-time setup, set {_BOOTSTRAP_ENV_VAR}=1. "
|
|
f"If this is a restored instance, restore the key file "
|
|
f"from backup — generating a new one would make every "
|
|
f"existing Credential row undecryptable."
|
|
)
|
|
log.warning(
|
|
"Generating NEW Fernet credential key at %s. Any existing "
|
|
"encrypted_blob rows in the DB will be undecryptable — "
|
|
"re-upload each credential after this completes.",
|
|
self._key_path,
|
|
)
|
|
parent = self._key_path.parent
|
|
parent.mkdir(parents=True, exist_ok=True)
|
|
os.chmod(parent, 0o700)
|
|
key = Fernet.generate_key()
|
|
self._key_path.write_bytes(key)
|
|
os.chmod(self._key_path, 0o600)
|
|
return key
|
|
|
|
def encrypt(self, plaintext: str) -> bytes:
|
|
return self._fernet.encrypt(plaintext.encode("utf-8"))
|
|
|
|
def decrypt(self, ciphertext: bytes) -> str:
|
|
try:
|
|
return self._fernet.decrypt(ciphertext).decode("utf-8")
|
|
except InvalidToken as exc:
|
|
raise InvalidCredentialBlob(
|
|
"credential blob is corrupt or encrypted with a different key"
|
|
) from exc
|