"""Fernet-based symmetric encryption for OAuth tokens stored in DB."""
from cryptography.fernet import Fernet
from navi.config import settings
class TokenEncryptor:
"""Encrypt/decrypt opaque tokens before writing to DB."""
def __init__(self, key: str) -> None:
if not key:
raise ValueError("NAVI_AUTH_ENCRYPTION_KEY is required for token encryption")
self._fernet = Fernet(key.encode())
def encrypt(self, plain: str) -> str:
return self._fernet.encrypt(plain.encode()).decode()
def decrypt(self, cipher: str) -> str:
return self._fernet.decrypt(cipher.encode()).decode()
# Singleton instance, lazily created
_encryptor: TokenEncryptor | None = None
def get_encryptor() -> TokenEncryptor:
global _encryptor
if _encryptor is None:
_encryptor = TokenEncryptor(settings.navi_auth_encryption_key)
return _encryptor