"""Fernet-based encryption for credential blobs. The key is a single 32-byte value (urlsafe-base64-encoded; what Fernet.generate_key produces) stored at a fixed path inside the images/data root. Created on first boot if absent; mode 0600. No KDF needed — the file contents are already maximum-entropy random bytes. Operator backup procedure must include this file alongside the rest of /images/ — losing it makes existing encrypted_blob rows undecryptable (recovery = delete the rows and re-upload). """ import os from pathlib import Path from cryptography.fernet import Fernet, InvalidToken class InvalidCredentialBlob(Exception): """Raised when decryption fails (wrong key, tampered blob, …).""" class CredentialCrypto: """Fernet encrypt/decrypt with an on-disk key file. Instantiate with a path; the file is created on first access and reused thereafter. Tests pass a tmp_path; production calls with `IMAGES_ROOT / "secrets" / "credential_key.b64"`. """ def __init__(self, key_path: Path): self._key_path = Path(key_path) self._fernet = Fernet(self._load_or_create_key()) def _load_or_create_key(self) -> bytes: if self._key_path.exists(): return self._key_path.read_bytes() 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