diff --git a/app/handlers/admin.py b/app/handlers/admin.py index 29337ae..9c4b64c 100644 --- a/app/handlers/admin.py +++ b/app/handlers/admin.py @@ -4,7 +4,9 @@ import time from aiogram import Router, F from aiogram.types import Message, CallbackQuery -from aiogram.filters import Command, CommandObject +from aiogram.filters import Command, CommandObject, StateFilter +from aiogram.fsm.context import FSMContext +from aiogram.fsm.state import StatesGroup, State from app.db.memory import load_bookings from app.config import ADMIN_PASSWORD from app.logger import logger @@ -16,6 +18,31 @@ ADMIN_IDS = {991309145} _approved: dict[int, float] = {} APPROVED_TTL_SECONDS = 60 * 60 # 1 hour +# Failed attempt tracking (per user) to slow down brute force +_failed_attempts: dict[int, list[float]] = {} +MAX_ATTEMPTS = 5 +ATTEMPT_WINDOW_SECONDS = 300 # 5 minutes + + +class AdminState(StatesGroup): + waiting_password = State() + + +def _is_rate_limited(user_id: int) -> bool: + """Return True if the user exceeded the allowed number of failed attempts.""" + now = time.time() + attempts = [t for t in _failed_attempts.get(user_id, []) if now - t < ATTEMPT_WINDOW_SECONDS] + _failed_attempts[user_id] = attempts + return len(attempts) >= MAX_ATTEMPTS + + +def _register_failure(user_id: int) -> None: + _failed_attempts.setdefault(user_id, []).append(time.time()) + + +def _clear_failures(user_id: int) -> None: + _failed_attempts.pop(user_id, None) + def is_admin(user_id: int, password: str | None = None) -> bool: """Check if user is admin: by user_id, by password, or by approved session.""" @@ -29,19 +56,13 @@ def is_admin(user_id: int, password: str | None = None) -> bool: _approved.pop(user_id, None) # expired session if password and hmac.compare_digest(password.encode("utf-8"), ADMIN_PASSWORD.encode("utf-8")): _approved[user_id] = now + APPROVED_TTL_SECONDS + _clear_failures(user_id) return True return False -@router.message(Command("admin")) -async def admin_panel(message: Message, command: CommandObject): - uid = message.from_user.id - password = command.args - - if not is_admin(uid, password=password): - return # Silent - - logger.info("Admin {uid} opened admin panel", uid=uid) +async def _show_admin_panel(message: Message) -> None: + logger.info("Admin {uid} opened admin panel", uid=message.from_user.id) bookings = load_bookings() active = [b for b in bookings if b.get("status") == "active"] @@ -71,6 +92,67 @@ async def admin_panel(message: Message, command: CommandObject): ) +async def _scrub_message(message: Message) -> None: + """Delete a message that may contain a password. Best effort.""" + try: + await message.delete() + except Exception: + logger.debug("Could not delete message with password (no rights?)") + + +@router.message(Command("admin")) +async def admin_panel(message: Message, command: CommandObject, state: FSMContext): + uid = message.from_user.id + + # Never accept the password via command arguments — it stays in chat history. + # Scrub the command message too, in case someone typed a password after /admin. + if command.args: + await _scrub_message(message) + await message.answer("🔒 Не передавайте пароль в команде — он остаётся в истории чата.") + if is_admin(uid): + await _show_admin_panel(message) + return + + if is_admin(uid): + await _show_admin_panel(message) + return + + if _is_rate_limited(uid): + logger.warning("Admin password brute-force blocked for {uid}", uid=uid) + await message.answer("⛔ Слишком много попыток. Подождите 5 минут.") + return + + await state.set_state(AdminState.waiting_password) + await message.answer( + "🔐 Введите пароль администратора отдельным сообщением.\n" + "Сообщение с паролем будет удалено и не останется в истории чата." + ) + + +@router.message(StateFilter(AdminState.waiting_password), F.text) +async def admin_password_input(message: Message, state: FSMContext): + uid = message.from_user.id + password = message.text or "" + + # Always delete the message containing the password — do not keep it in history. + await _scrub_message(message) + + if _is_rate_limited(uid): + await state.clear() + await message.answer("⛔ Слишком много попыток. Подождите 5 минут.") + return + + if is_admin(uid, password=password): + await state.clear() + await _show_admin_panel(message) + return + + _register_failure(uid) + await state.clear() + logger.warning("Failed admin password attempt from {uid}", uid=uid) + await message.answer("⛔ Неверный пароль. Попробуйте ещё раз: /admin") + + @router.callback_query(F.data.startswith("admin_cancel_")) async def admin_cancel_booking(callback: CallbackQuery): if not is_admin(callback.from_user.id):