"""Шифрование секретов, хранимых в БД (Telegram api_hash, ключи AI). Решение по итогам ревью: ключи шифруются симметричным ключом; ключ шифрования пока живёт в env (LEADRADAR_ENCRYPTION_KEY). Для локальной разработки без env ключ генерируется и кладётся в data/encryption.key (с предупреждением). """ from __future__ import annotations import base64 import logging import os from cryptography.fernet import Fernet, InvalidToken from . import config log = logging.getLogger("leadradar.crypto") _fernet: Fernet | None = None def _get_fernet() -> Fernet: global _fernet if _fernet is not None: return _fernet key = config.ENCRYPTION_KEY if key: try: _fernet = Fernet(key.encode() if not key.endswith("=") else key.encode()) return _fernet except Exception: # noqa: BLE001 log.error("LEADRADAR_ENCRYPTION_KEY не похож на Fernet-ключ (32 байта urlsafe b64)") raise if config.ENCRYPTION_KEY_FILE.exists(): key = config.ENCRYPTION_KEY_FILE.read_text(encoding="utf-8").strip() else: config.ensure_dirs() key = Fernet.generate_key().decode() config.ENCRYPTION_KEY_FILE.write_text(key, encoding="utf-8") log.warning("Ключ шифрования создан в %s (для продакшена задайте LEADRADAR_ENCRYPTION_KEY)", config.ENCRYPTION_KEY_FILE) _fernet = Fernet(key.encode()) return _fernet def encrypt_text(value: str) -> str: if not value: return "" token = _get_fernet().encrypt(value.encode()) return "enc:" + token.decode() def decrypt_text(value: str) -> str: if not value: return "" if not value.startswith("enc:"): return value # совместимость с незашифрованными значениями ранних версий try: return _get_fernet().decrypt(value[4:].encode()).decode() except InvalidToken: log.error("Не удалось расшифровать секрет (неверный ключ шифрования)") return "" def maybe_encrypt(value: str) -> str: """Шифруем только непустые значения.""" return encrypt_text(value) if value else "" def random_key() -> str: return Fernet.generate_key().decode()