48 lines
1.4 KiB
Python
48 lines
1.4 KiB
Python
"""
|
|
Transparent at-rest encryption for sensitive DB columns (OAuth tokens).
|
|
|
|
Uses a SQLAlchemy TypeDecorator so existing code needs zero changes —
|
|
values are encrypted on write and decrypted on read automatically.
|
|
|
|
Key derivation: SHA-256(SECRET_KEY) → 32 bytes → Fernet-compatible base64 key.
|
|
|
|
Migration: the decryptor falls back to returning the raw value when
|
|
decryption fails, so existing plaintext tokens keep working until
|
|
they are re-written (e.g. next token refresh or reconnect).
|
|
"""
|
|
import base64
|
|
import hashlib
|
|
import logging
|
|
|
|
from sqlalchemy import types
|
|
|
|
log = logging.getLogger('app.crypto')
|
|
|
|
|
|
def _fernet():
|
|
from flask import current_app
|
|
from cryptography.fernet import Fernet
|
|
secret = current_app.config['SECRET_KEY']
|
|
key = base64.urlsafe_b64encode(hashlib.sha256(secret.encode()).digest())
|
|
return Fernet(key)
|
|
|
|
|
|
class EncryptedText(types.TypeDecorator):
|
|
"""Store text columns encrypted at rest using Fernet symmetric encryption."""
|
|
impl = types.Text
|
|
cache_ok = True
|
|
|
|
def process_bind_param(self, value, dialect):
|
|
if value is None:
|
|
return None
|
|
return _fernet().encrypt(value.encode()).decode()
|
|
|
|
def process_result_value(self, value, dialect):
|
|
if value is None:
|
|
return None
|
|
try:
|
|
return _fernet().decrypt(value.encode()).decode()
|
|
except Exception:
|
|
# Value is plaintext (pre-migration) — return as-is
|
|
return value
|