Rustam Khalimov
9e07568ddd
Первый коммит: модульный монолит ядра (.NET 10) и gRPC-сервисы ai/ml/telegram, фронтенд Vue 3/Vite/Tailwind, документация (ТЗ, инструкция пользователя, техдокументация, код-стайл), бэклог, скрипты развёртывания и архив прототипа LeadRadar.
71 lines
2.5 KiB
Python
71 lines
2.5 KiB
Python
"""Шифрование секретов, хранимых в БД (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()
|