23ee551271
Техническая защита: - services/safety_guard.py — модуль безопасности - config.py — SAFETY_MODE переменная - main.py — инициализация при запуске - .env.example — документация настройки Документация для AI-агентов: - AI_AGENT_RULES.md — полные правила поведения - КРИТИЧЕСКИЕ-ПРАВИЛА.md — обновлено с Правилом №4 Как работает: - SAFETY_MODE=production — обычная работа - SAFETY_MODE=safe — блокировка отправок в общий чат - Логирование всех заблокированных попыток - Декоратор @require_production_mode Co-authored-by: Qwen-Coder <qwen-coder@alibabacloud.com>
130 lines
4.9 KiB
Python
130 lines
4.9 KiB
Python
"""
|
||
Модуль безопасности — защита от случайной отправки сообщений в общий чат
|
||
"""
|
||
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
|