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

1"""Symmetric encryption mirror for the worker. 

2 

3Identical surface to `backend/app/utils/crypto.py` so behaviour cannot drift. 

4The shared `CREDENTIAL_ENCRYPTION_KEY` lets the worker decrypt envelopes the 

5backend pushed onto the Celery queue without ever touching the database. 

6""" 

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. The worker requires the same " 

22 "Fernet key as the backend to decrypt credential envelopes." 

23 ) 

24 try: 

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

26 except (ValueError, TypeError) as e: 

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

28 

29 

30_cipher = _build_cipher() 

31 

32 

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

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

35 

36 

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

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

39 

40 

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

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

43 

44 

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

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

47 

48 

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