| """ |
| Encryption utilities - Fernet envelope encryption for user API keys. |
| Uses ENCRYPTION_KEY (separate from JWT SECRET_KEY) when available. |
| """ |
| import base64 |
|
|
| from cryptography.fernet import Fernet, InvalidToken |
| from cryptography.hazmat.primitives import hashes |
| from cryptography.hazmat.primitives.kdf.pbkdf2 import PBKDF2HMAC |
|
|
| from app.core.config import settings |
|
|
|
|
| def _get_encryption_source() -> str: |
| """Get the key source — ENCRYPTION_KEY if set, else SECRET_KEY.""" |
| return settings.ENCRYPTION_KEY or settings.SECRET_KEY |
|
|
|
|
| def _derive_fernet_key() -> bytes: |
| """Derive Fernet key via PBKDF2.""" |
| kdf = PBKDF2HMAC( |
| algorithm=hashes.SHA256(), |
| length=32, |
| salt=b"jobportal-api-key-encryption-v2", |
| iterations=100_000, |
| ) |
| return base64.urlsafe_b64encode(kdf.derive(_get_encryption_source().encode())) |
|
|
|
|
| def encrypt_api_key(plaintext: str) -> str: |
| """Encrypt an API key for database storage.""" |
| return Fernet(_derive_fernet_key()).encrypt(plaintext.encode()).decode() |
|
|
|
|
| def decrypt_api_key(ciphertext: str) -> str | None: |
| """Decrypt an API key. Returns None on failure.""" |
| try: |
| return Fernet(_derive_fernet_key()).decrypt(ciphertext.encode()).decode() |
| except (InvalidToken, Exception): |
| return None |
|
|
|
|
| def get_key_last4(key: str) -> str: |
| """Last 4 chars for display without decryption.""" |
| return key[-4:] if len(key) >= 4 else "****" |
|
|
|
|
| def get_key_prefix(key: str) -> str: |
| """First 4 chars for display.""" |
| return key[:4] if len(key) >= 4 else "" |
|
|
|
|
| def mask_api_key(key: str) -> str: |
| """Masked display: sk-...ab4f""" |
| if len(key) <= 8: |
| return "****" |
| return f"{key[:4]}...{key[-4:]}" |
|
|
