# security/crypto.py import json import base64 import hashlib from cryptography.fernet import Fernet from config.settings import settings import logging logger = logging.getLogger(__name__) def _get_fernet() -> Fernet: """ Erzeugt einen Fernet-Schlüssel aus dem AES-Secret der Settings. Da Fernet einen base64-kodierten 32-Byte-Schlüssel erfordert, hashen wir den String aus den Settings und kodieren ihn um. """ # GRAVITY_TOKEN is an acceptable fallback so a deployment cannot silently # encrypt durable checkpoints with a public, repository-wide default key. secret = (settings.aes_secret_key or settings.gravity_token).strip() if len(secret) < 16: raise RuntimeError("AES_SECRET_KEY or GRAVITY_TOKEN must contain at least 16 characters") key_hash = hashlib.sha256(secret.encode('utf-8')).digest() b64_key = base64.urlsafe_b64encode(key_hash) return Fernet(b64_key) def encrypt_state(state: dict) -> str: """ Verschlüsselt ein Profil-State (Dict) in einen AES-verschlüsselten String. """ try: f = _get_fernet() json_bytes = json.dumps(state).encode('utf-8') encrypted = f.encrypt(json_bytes) return encrypted.decode('utf-8') except Exception as e: logger.error(f"Fehler bei der Verschlüsselung des Profile-States: {e}") return "" def decrypt_state(encrypted_token: str) -> dict | None: """ Entschlüsselt einen AES-verschlüsselten State in ein Dictionary. """ if not encrypted_token: return None try: f = _get_fernet() decrypted_bytes = f.decrypt(encrypted_token.encode('utf-8')) return json.loads(decrypted_bytes.decode('utf-8')) except Exception as e: logger.error(f"Fehler bei der Entschlüsselung des Profile-States: {e}") return None def encrypt_bytes(data: bytes) -> bytes: """ Verschlüsselt Rohdaten (z.B. Bilder) mit Fernet / AES-128-CBC + HMAC-SHA256. """ f = _get_fernet() return f.encrypt(data) def decrypt_bytes(encrypted_data: bytes) -> bytes: """ Entschlüsselt Rohdaten. """ f = _get_fernet() return f.decrypt(encrypted_data)