"""Encryption at rest for OAuth tokens and other sensitive material. A single Fernet key (from settings.ENCRYPTION_KEY or a persisted generated key) encrypts all sensitive columns. Refresh tokens / client secrets are NEVER stored in plaintext and NEVER logged. """ import base64 import os from cryptography.fernet import Fernet from app.core.config import settings _KEY_FILE = os.path.join(os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__)))), ".encryption_key") def _load_or_create_key() -> bytes: env_key = settings.ENCRYPTION_KEY if env_key: # Accept urlsafe-base64 Fernet key (44 chars) directly. try: Fernet(env_key.encode()) return env_key.encode() except Exception: pass if os.path.exists(_KEY_FILE): with open(_KEY_FILE, "rb") as f: return f.read().strip() key = Fernet.generate_key() fd = os.open(_KEY_FILE, os.O_WRONLY | os.O_CREAT | os.O_TRUNC, 0o600) try: os.write(fd, key) finally: os.close(fd) return key _f = Fernet(_load_or_create_key()) def encrypt(value: str) -> str: if value is None: return None return _f.encrypt(value.encode("utf-8")).decode("ascii") def decrypt(value: str) -> str: if not value: return "" return _f.decrypt(value.encode("ascii")).decode("utf-8") def rotate_key_clear() -> None: # pragma: no cover - admin utility os.remove(_KEY_FILE)