Coverage for app/utils/crypto.py: 100.00%

1"""Symmetric encryption for at-rest credentials and Celery-envelope payloads. 

2 

3The Fernet key (`CREDENTIAL_ENCRYPTION_KEY`) is shared between the backend 

4and the worker. The backend encrypts when storing in Postgres and forwards 

5the ciphertext (base64) through Celery; the worker decrypts in-process. 

6Plaintext never leaves either container's memory. 

7""" 

8from __future__ import annotations 

9 

10import base64 

11 

12from cryptography.fernet import Fernet, InvalidToken 

13 

14from app.config import settings 

15 

16 

17def _build_cipher() -> Fernet: 

18 key = settings.CREDENTIAL_ENCRYPTION_KEY 

19 if not key: 

20 raise RuntimeError( 

21 "CREDENTIAL_ENCRYPTION_KEY is not set. Generate one with: " 

22 "python -c 'from cryptography.fernet import Fernet; " 

23 "print(Fernet.generate_key().decode())'" 

24 ) 

25 try: 

26 # Fernet validates the 32-byte url-safe-base64 key shape itself. 

27 return Fernet(key.encode() if isinstance(key, str) else key) 

28 except (ValueError, TypeError) as e: 

29 raise RuntimeError(f"CREDENTIAL_ENCRYPTION_KEY is malformed: {e}") from e 

30 

31 

32_cipher = _build_cipher() 

33 

34 

35def encrypt(plaintext: str) -> bytes: 

36 """Encrypt a string. Returns Fernet ciphertext as bytes (store as BYTEA).""" 

37 return _cipher.encrypt(plaintext.encode("utf-8")) 

38 

39 

40def decrypt(token: bytes) -> str: 

41 """Decrypt Fernet ciphertext bytes. Raises InvalidToken on tampering / wrong key.""" 

42 return _cipher.decrypt(token).decode("utf-8") 

43 

44 

45def encrypt_b64(plaintext: str) -> str: 

46 """Encrypt and base64-encode for JSON-safe transport (Celery args).""" 

47 return base64.b64encode(encrypt(plaintext)).decode("ascii") 

48 

49 

50def decrypt_b64(token_b64: str) -> str: 

51 """Inverse of `encrypt_b64`. Raises InvalidToken on bad input.""" 

52 return decrypt(base64.b64decode(token_b64.encode("ascii"))) 

53 

54 

55__all__ = ["encrypt", "decrypt", "encrypt_b64", "decrypt_b64", "InvalidToken"]