""" Модуль безопасности — защита от случайной отправки сообщений в общий чат """ import logging import os from functools import wraps from typing import Set logger = logging.getLogger(__name__) # ============================================================================ # НАСТРОЙКИ БЕЗОПАСНОСТИ # ============================================================================ # Режим безопасности: 'production' или 'safe' # В 'safe' режиме ВСЕ отправки в общие чаты блокируются SAFETY_MODE = os.getenv('SAFETY_MODE', 'production').lower() # Критические чаты — НИКОГДА не отправлять туда тестовые сообщения # Это ID общего чата дома CRITICAL_CHAT_IDS: Set[int] = set() def register_critical_chat(chat_id: int): """Зарегистрировать критический чат (общий чат дома)""" CRITICAL_CHAT_IDS.add(chat_id) logger.info(f"🔒 Зарегистрирован критический чат: {chat_id}") def is_critical_chat(chat_id: int) -> bool: """Проверить является ли чат критическим""" return chat_id in CRITICAL_CHAT_IDS def can_send_to_chat(chat_id: int, context: str = "") -> tuple[bool, str]: """ Проверить можно ли отправлять сообщение в этот чат Returns: (можно_ли_отправить, причина) """ if SAFETY_MODE == 'safe': if is_critical_chat(chat_id): return False, f"🔒 SAFETY_MODE: Отправка в чат {chat_id} заблокирована. Контекст: {context}" return True, "" async def safe_send_message(bot, chat_id: int, text: str, context: str = "", **kwargs): """ Безопасная отправка сообщения — проверяет можно ли отправлять Args: bot: Bot instance chat_id: ID чата text: Текст сообщения context: Контекст отправки (для логов) **kwargs: Дополнительные аргументы для send_message """ can_send, reason = can_send_to_chat(chat_id, context) if not can_send: logger.warning(f"🔒 БЛОКИРОВКА ОТПРАВКИ: {reason}") return None try: return await bot.send_message(chat_id, text, **kwargs) except Exception as e: logger.error(f"Ошибка отправки: {e}") raise async def safe_send_photo(bot, chat_id: int, photo, caption: str = "", context: str = "", **kwargs): """ Безопасная отправка фото Args: bot: Bot instance chat_id: ID чата photo: Фото caption: Подпись context: Контекст отправки **kwargs: Дополнительные аргументы """ can_send, reason = can_send_to_chat(chat_id, context) if not can_send: logger.warning(f"🔒 БЛОКИРОВКА ОТПРАВКИ ФОТО: {reason}") return None try: return await bot.send_photo(chat_id, photo, caption=caption, **kwargs) except Exception as e: logger.error(f"Ошибка отправки фото: {e}") raise # ============================================================================ # ДЕКОРАТОР ДЛЯ БЛОКИРОВКИ РАССЫЛОК # ============================================================================ def require_production_mode(func): """ Декоратор — блокирует функцию если SAFETY_MODE != 'production' Использование: @require_production_mode async def send_broadcast_to_all(): ... """ @wraps(func) async def wrapper(*args, **kwargs): if SAFETY_MODE != 'production': msg = f"🔒 Функция {func.__name__} заблокирована: SAFETY_MODE='{SAFETY_MODE}'" logger.warning(msg) raise PermissionError(msg) return await func(*args, **kwargs) return wrapper # ============================================================================ # ИНИЦИАЛИЗАЦИЯ # ============================================================================ def init_safety_module(admin_chat_id: int = None): """ Инициализировать модуль безопасности Args: admin_chat_id: ID общего чата администратора (блокируется в safe режиме) """ if admin_chat_id: register_critical_chat(admin_chat_id) logger.info(f"🔒 Режим безопасности: {SAFETY_MODE}") logger.info(f"🔒 Критические чаты: {CRITICAL_CHAT_IDS}") return SAFETY_MODE