diff --git a/filters/admin.py b/filters/admin.py index 2d6b8cbd..eba89e04 100644 --- a/filters/admin.py +++ b/filters/admin.py @@ -1,16 +1,16 @@ from aiogram.filters import BaseFilter -from aiogram.types import Message +from aiogram.types import Message, CallbackQuery from sqlalchemy import select from database.models import Admin from database.db import async_session_maker class IsAdminFilter(BaseFilter): - async def __call__(self, message: Message) -> bool: + async def __call__(self, event: Message | CallbackQuery) -> bool: try: async with async_session_maker() as session: result = await session.execute( - select(Admin).where(Admin.tg_id == message.from_user.id) + select(Admin).where(Admin.tg_id == event.from_user.id) ) admin = result.scalar_one_or_none() return admin is not None diff --git a/handlers/admin/management/management_handler.py b/handlers/admin/management/management_handler.py index 65dd9424..e456c444 100644 --- a/handlers/admin/management/management_handler.py +++ b/handlers/admin/management/management_handler.py @@ -254,6 +254,13 @@ async def set_admin_role(callback: CallbackQuery, callback_data: AdminPanelCallb await callback.message.edit_text("❌ Неверный формат.") return + if tg_id == callback.from_user.id: + await callback.message.edit_text( + "🚫 Нельзя изменить свою собственную роль!", + reply_markup=build_single_admin_menu(tg_id) + ) + return + result = await session.execute(select(Admin).where(Admin.tg_id == tg_id)) admin = result.scalar_one_or_none() if not admin: diff --git a/middlewares/__init__.py b/middlewares/__init__.py index bae5d0da..e8b05b8b 100644 --- a/middlewares/__init__.py +++ b/middlewares/__init__.py @@ -25,10 +25,10 @@ def register_middleware( """Регистрирует middleware в диспетчере.""" if middlewares is None: available_middlewares = { - "admin": AdminMiddleware(), "session": ( SessionMiddleware(sessionmaker) if sessionmaker else SessionMiddleware() ), + "admin": AdminMiddleware(), "maintenance": MaintenanceModeMiddleware(), "logging": LoggingMiddleware(), "throttling": ThrottlingMiddleware(), diff --git a/middlewares/admin.py b/middlewares/admin.py index 2aad6b7c..ba476ccb 100644 --- a/middlewares/admin.py +++ b/middlewares/admin.py @@ -3,8 +3,10 @@ from typing import Any from aiogram import BaseMiddleware from aiogram.types import CallbackQuery, Message, TelegramObject +from sqlalchemy import select from config import ADMIN_ID +from database.models import Admin class AdminMiddleware(BaseMiddleware): @@ -25,18 +27,30 @@ class AdminMiddleware(BaseMiddleware): data: dict[str, Any], ) -> Any: """Обрабатывает событие и добавляет флаг администратора в data.""" - data["admin"] = self._check_admin_access(event) + data["admin"] = await self._check_admin_access(event, data.get("session")) return await handler(event, data) - def _check_admin_access(self, event: TelegramObject) -> bool: + async def _check_admin_access(self, event: TelegramObject, session) -> bool: """Проверяет, имеет ли пользователь права администратора.""" try: + user_id = None if isinstance(event, Message): - return event.from_user and event.from_user.id in self._admin_ids + user_id = event.from_user.id if event.from_user else None elif isinstance(event, CallbackQuery): - return event.from_user and event.from_user.id in self._admin_ids + user_id = event.from_user.id if event.from_user else None + else: + user_id = getattr(getattr(event, "from_user", None), "id", None) - user_id = getattr(getattr(event, "from_user", None), "id", None) - return user_id in self._admin_ids if user_id else False + if not user_id: + return False + + if user_id in self._admin_ids: + return True + + if session: + result = await session.execute(select(Admin).where(Admin.tg_id == user_id)) + return result.scalar_one_or_none() is not None + + return False except Exception: return False