fix: пароль админа больше не остаётся в истории чата (PRJ-3)

- /admin не принимает пароль аргументом команды (оставался в истории Telegram)
- пароль вводится отдельным сообщением через FSM-состояние waiting_password
- сообщение с паролем удаляется сразу после проверки (best-effort)
- добавлена защита от брутфорса: 5 неудачных попыток за 5 минут
This commit is contained in:
2026-08-04 19:36:19 +04:00
parent 896c26bd57
commit 49d7af2d18
+92 -10
View File
@@ -4,7 +4,9 @@ import time
from aiogram import Router, F from aiogram import Router, F
from aiogram.types import Message, CallbackQuery 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.db.memory import load_bookings
from app.config import ADMIN_PASSWORD from app.config import ADMIN_PASSWORD
from app.logger import logger from app.logger import logger
@@ -16,6 +18,31 @@ ADMIN_IDS = {991309145}
_approved: dict[int, float] = {} _approved: dict[int, float] = {}
APPROVED_TTL_SECONDS = 60 * 60 # 1 hour 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: 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.""" """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 _approved.pop(user_id, None) # expired session
if password and hmac.compare_digest(password.encode("utf-8"), ADMIN_PASSWORD.encode("utf-8")): if password and hmac.compare_digest(password.encode("utf-8"), ADMIN_PASSWORD.encode("utf-8")):
_approved[user_id] = now + APPROVED_TTL_SECONDS _approved[user_id] = now + APPROVED_TTL_SECONDS
_clear_failures(user_id)
return True return True
return False return False
@router.message(Command("admin")) async def _show_admin_panel(message: Message) -> None:
async def admin_panel(message: Message, command: CommandObject): logger.info("Admin {uid} opened admin panel", uid=message.from_user.id)
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)
bookings = load_bookings() bookings = load_bookings()
active = [b for b in bookings if b.get("status") == "active"] 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_")) @router.callback_query(F.data.startswith("admin_cancel_"))
async def admin_cancel_booking(callback: CallbackQuery): async def admin_cancel_booking(callback: CallbackQuery):
if not is_admin(callback.from_user.id): if not is_admin(callback.from_user.id):