ryan
863ebe8e20
Tortoise calls to_python_value() when setting field values at create() time, not only when loading from DB. Add _is_fernet_token() to distinguish encrypted DB values (≥73 decoded bytes, version 0x80) from plaintext being set — the latter passes through untouched while wrong-key errors on real tokens still raise. Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
91 lines
3.0 KiB
Python
91 lines
3.0 KiB
Python
"""
|
|
Encryption service for email credentials.
|
|
|
|
Provides transparent Fernet encryption for sensitive fields in the database.
|
|
"""
|
|
|
|
import base64
|
|
import os
|
|
from cryptography.fernet import Fernet
|
|
from tortoise import fields
|
|
|
|
|
|
def _is_fernet_token(value: str) -> bool:
|
|
"""Return True if value looks like a Fernet token.
|
|
|
|
Fernet tokens are URL-safe base64 and decode to at least 73 bytes
|
|
(1 version + 8 timestamp + 16 IV + 16 min-ciphertext + 32 HMAC).
|
|
The first decoded byte is always 0x80 (the Fernet version byte).
|
|
"""
|
|
try:
|
|
padded = value + "=" * (-len(value) % 4)
|
|
raw = base64.urlsafe_b64decode(padded.encode())
|
|
return len(raw) >= 73 and raw[0] == 0x80
|
|
except Exception:
|
|
return False
|
|
|
|
|
|
class EncryptedTextField(fields.TextField):
|
|
"""
|
|
Custom Tortoise ORM field that transparently encrypts/decrypts text values.
|
|
|
|
Uses Fernet symmetric encryption with a key from FERNET_KEY environment variable.
|
|
"""
|
|
|
|
def __init__(self, **kwargs):
|
|
super().__init__(**kwargs)
|
|
# Load encryption key from environment
|
|
key = os.getenv("FERNET_KEY")
|
|
if not key:
|
|
raise ValueError(
|
|
"FERNET_KEY environment variable required for encrypted fields. "
|
|
'Generate with: python -c "from cryptography.fernet import Fernet; print(Fernet.generate_key().decode())"'
|
|
)
|
|
try:
|
|
self.fernet = Fernet(key.encode())
|
|
except Exception as e:
|
|
raise ValueError(f"Invalid FERNET_KEY format: {e}")
|
|
|
|
def to_db_value(self, value: str, instance) -> str:
|
|
"""Encrypt value before storing in database."""
|
|
if value is None:
|
|
return None
|
|
# Encrypt and return as URL-safe base64 string
|
|
return self.fernet.encrypt(value.encode()).decode()
|
|
|
|
def to_python_value(self, value: str) -> str:
|
|
"""Decrypt value when loading from database.
|
|
|
|
Plaintext values (e.g. freshly-set before the first DB round-trip) are
|
|
returned as-is. Values that look like Fernet tokens are decrypted — a
|
|
wrong key still raises InvalidToken, preserving the security guarantee.
|
|
"""
|
|
if value is None:
|
|
return None
|
|
if not _is_fernet_token(value):
|
|
return value
|
|
return self.fernet.decrypt(value.encode()).decode()
|
|
|
|
|
|
def validate_fernet_key():
|
|
"""
|
|
Validate that FERNET_KEY is set and functional.
|
|
|
|
Raises:
|
|
ValueError: If key is missing or invalid
|
|
"""
|
|
key = os.getenv("FERNET_KEY")
|
|
if not key:
|
|
raise ValueError("FERNET_KEY environment variable not set")
|
|
|
|
try:
|
|
f = Fernet(key.encode())
|
|
# Test encryption/decryption cycle
|
|
test_value = b"test_encryption"
|
|
encrypted = f.encrypt(test_value)
|
|
decrypted = f.decrypt(encrypted)
|
|
if decrypted != test_value:
|
|
raise ValueError("Encryption/decryption test failed")
|
|
except Exception as e:
|
|
raise ValueError(f"FERNET_KEY validation failed: {e}")
|