demo_spa_bot/app/handlers/admin.py

"""Admin handler for spa-telegram-bot — secure with password."""
import hmac
import time

from aiogram import Router, F
from aiogram.types import Message, CallbackQuery
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

router = Router()

ADMIN_IDS = {991309145}
# Store approved sessions — user_ids that passed password check
_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."""
    if user_id in ADMIN_IDS:
        return True
    now = time.time()
    expires_at = _approved.get(user_id)
    if expires_at and expires_at > now:
        return True
    if expires_at:
        _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


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"]

    if not active:
        await message.answer("📋 Активных записей нет")
        return

    from aiogram.types import InlineKeyboardMarkup, InlineKeyboardButton

    for booking in active:
        buttons = [[
            InlineKeyboardButton(
                text="❌ Отменить",
                callback_data=f"admin_cancel_{booking['id']}"
            )
        ]]
        user_name = booking.get("user_name") or f"id{booking['user_id']}"
        await message.answer(
            f"📋 Запись #{booking['id']}\n"
            f"👤 Клиент: {user_name}\n"
            f"💆 Услуги: {booking.get('service', '?')}\n"
            f"👤 Мастер: {booking.get('master', '?')}\n"
            f"📅 Дата: {booking.get('date', '?')}\n"
            f"🕒 Время: {booking.get('time', '?')}",
            reply_markup=InlineKeyboardMarkup(inline_keyboard=buttons)
        )


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):
        await callback.answer("⛔ Нет доступа")
        return

    try:
        booking_id = int(callback.data.split("_", 2)[2])
    except (ValueError, IndexError):
        await callback.answer("Некорректные данные")
        return

    logger.info("Admin {uid} cancelled booking #{bid}", uid=callback.from_user.id, bid=booking_id)

    from app.db.memory import cancel_booking
    cancel_booking(booking_id)

    await callback.message.edit_text(
        f"✅ Запись #{booking_id} отменена админом"
    )
    await callback.answer("Запись отменена")