78 lines
2.6 KiB
Python
78 lines
2.6 KiB
Python
"""Проверка доступа пользователя к функциональности бота."""
|
||
|
||
from __future__ import annotations
|
||
|
||
import logging
|
||
import time
|
||
|
||
from aiogram import Bot
|
||
from aiogram.enums import ChatMemberStatus
|
||
from aiogram.types import Message
|
||
|
||
from bot.services import alerts
|
||
from bot.services import users as users_service
|
||
from common.analytics import EventType, record, remember_user
|
||
from common.config import get_settings
|
||
from common.texts import alerts as alerts_text
|
||
from common.texts.render import render
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
_ALLOWED_STATUSES = {
|
||
ChatMemberStatus.MEMBER,
|
||
ChatMemberStatus.CREATOR,
|
||
ChatMemberStatus.ADMINISTRATOR,
|
||
ChatMemberStatus.RESTRICTED,
|
||
}
|
||
|
||
#: TTL-кеш членства, чтобы не дёргать get_chat_member на каждое сообщение.
|
||
_MEMBERSHIP_TTL = 300.0
|
||
_membership_cache: dict[int, tuple[float, bool]] = {}
|
||
|
||
|
||
async def _is_member(bot: Bot, user_id: int) -> bool:
|
||
now = time.monotonic()
|
||
cached = _membership_cache.get(user_id)
|
||
if cached is not None and now - cached[0] < _MEMBERSHIP_TTL:
|
||
return cached[1]
|
||
|
||
member = await bot.get_chat_member(get_settings().chat_id, user_id)
|
||
allowed = member.status in _ALLOWED_STATUSES
|
||
_membership_cache[user_id] = (now, allowed)
|
||
return allowed
|
||
|
||
|
||
async def check_user(message: Message) -> bool:
|
||
"""Пропустить пользователя: доступ только участникам основного чата (или админам)."""
|
||
settings = get_settings()
|
||
|
||
user = message.from_user
|
||
if user is None:
|
||
return False
|
||
|
||
if user.id in settings.admin_ids or await _is_member(message.bot, user.id):
|
||
await users_service.register_from_message(message)
|
||
remember_user(user.id, user.username)
|
||
return True
|
||
|
||
record(
|
||
EventType.ACCESS_DENIED,
|
||
user_id=user.id,
|
||
username=user.username,
|
||
chat_id=message.chat.id,
|
||
payload={"reason": "not_member"},
|
||
)
|
||
await users_service.delete_user(user.id)
|
||
logger.warning("Пользователь %s стучится в чат, доступ запрещён", user.id)
|
||
await _notify_denied(message)
|
||
return False
|
||
|
||
|
||
async def _notify_denied(message: Message) -> None:
|
||
"""Сообщить в чат логов (если задан) о попытке доступа извне."""
|
||
user = message.from_user
|
||
if user is None:
|
||
return
|
||
text = render(alerts_text.MSG_ACCESS_DENIED, {"user": alerts.user_tag(user.username, user.id)})
|
||
await alerts.notify_alert(message.bot, text, forward=message)
|