DispelEvilBot/admin/routes/auth.py

"""Вход в админку через Telegram Login Widget."""

from __future__ import annotations

import logging
import secrets

from fastapi import APIRouter, Request, Response
from fastapi.responses import HTMLResponse, JSONResponse, RedirectResponse

from admin.security import check_csrf, safe_redirect_target, verify_telegram_auth
from common import __version__
from common.repositories import admin_login_tokens_repo, admins_repo

logger = logging.getLogger(__name__)

router = APIRouter(tags=["auth"])

#: Поля, которые Telegram включает в подпись (остальные в query игнорируем).
TELEGRAM_FIELDS = frozenset({"id", "first_name", "last_name", "username", "photo_url", "auth_date", "hash"})


def _deny(request: Request, message: str, status_code: int = 403) -> HTMLResponse:
    return request.app.state.templates.TemplateResponse(
        request=request,
        name="auth/denied.html",
        context={"version": __version__, "message": message},
        status_code=status_code,
    )


@router.get("/login", response_class=HTMLResponse)
async def login_page(request: Request, next: str = "/") -> Response:
    """Страница входа с кнопкой Telegram."""
    target = safe_redirect_target(next)
    if request.session.get("admin_id"):
        return RedirectResponse(url=target, status_code=303)

    settings = request.app.state.settings
    return request.app.state.templates.TemplateResponse(
        request=request,
        name="auth/login.html",
        context={
            "version": __version__,
            "bot_username": settings.bot_username,
            "public_url": settings.admin_public_url.rstrip("/"),
            "configured": bool(settings.bot_token and settings.bot_username),
            "deeplink": bool(settings.bot_username and settings.admin_public_url),
            "deeplink_pending": bool(request.cookies.get("debot_login_token")),
            "deeplink_error": request.query_params.get("error") == "deeplink",
            "next": target,
        },
    )


@router.get("/auth/deeplink/start")
async def deeplink_start(request: Request) -> Response:
    """Создать токен и открыть установленный Telegram с deep-link."""
    settings = request.app.state.settings
    username = settings.bot_username.lstrip("@")
    if not username:
        return RedirectResponse(url="/login?error=deeplink", status_code=303)

    token = await admin_login_tokens_repo.create_token()
    response = RedirectResponse(
        url=f"https://t.me/{username}?start=login_{token}",
        status_code=303,
    )
    response.set_cookie(
        "debot_login_token",
        token,
        max_age=300,
        httponly=True,
        samesite="lax",
        secure=settings.is_prod,
    )
    return response


@router.get("/auth/deeplink/status")
async def deeplink_status(request: Request) -> JSONResponse:
    """Проверить, подтвердил ли пользователь вход в боте, и открыть сессию."""
    token = request.cookies.get("debot_login_token")
    if not token:
        return JSONResponse({"status": "none"})

    try:
        telegram_id = await admin_login_tokens_repo.consume_token(token)
    except Exception:
        logger.exception("Не удалось проверить deep-link токен")
        return JSONResponse({"status": "pending"})

    if telegram_id is None:
        return JSONResponse({"status": "pending"})

    try:
        admin = await admins_repo.get_admin(telegram_id)
    except Exception:
        logger.exception("Ошибка проверки администратора при deep-link входе")
        return JSONResponse({"status": "pending"})

    if admin is None or not admin["enabled"]:
        response = JSONResponse({"status": "denied"})
        response.delete_cookie("debot_login_token")
        return response

    request.session.clear()
    request.session["admin_id"] = telegram_id
    request.session["admin_username"] = admin["username"] or ""
    request.session["csrf"] = secrets.token_urlsafe(32)

    try:
        await admins_repo.touch_login(telegram_id)
    except Exception:
        logger.warning("Не удалось обновить last_login_at", exc_info=True)

    logger.info("Вход в админку по deep-link: %s", telegram_id)
    response = JSONResponse({"status": "ok"})
    response.delete_cookie("debot_login_token")
    return response


@router.get("/auth/telegram")
async def telegram_callback(request: Request) -> Response:
    """Проверить подпись Telegram, роль в БД и открыть сессию."""
    settings = request.app.state.settings
    data = {key: value for key, value in request.query_params.items() if key in TELEGRAM_FIELDS}

    if not verify_telegram_auth(data, settings.bot_token):
        logger.warning("Отклонён Telegram-логин: неверная подпись или устаревший auth_date")
        return _deny(request, "Не удалось проверить вход через Telegram. Попробуйте ещё раз.")

    try:
        telegram_id = int(str(data.get("id")))
    except (TypeError, ValueError):
        return _deny(request, "Telegram не передал идентификатор пользователя.", status_code=400)

    try:
        admin = await admins_repo.get_admin(telegram_id)
    except Exception:
        logger.exception("Ошибка проверки администратора при входе")
        return _deny(request, "Не удалось проверить доступ. Попробуйте позже.", status_code=503)

    if admin is None or not admin["enabled"]:
        logger.info("Отказ во входе: %s нет среди администраторов", telegram_id)
        return _deny(request, "У вас нет доступа к админке. Обратитесь к владельцу.")

    request.session.clear()
    request.session["admin_id"] = telegram_id
    request.session["admin_username"] = admin["username"] or data.get("username") or data.get("first_name") or ""
    request.session["csrf"] = secrets.token_urlsafe(32)

    try:
        await admins_repo.touch_login(telegram_id)
    except Exception:
        logger.warning("Не удалось обновить last_login_at", exc_info=True)

    logger.info("Вход в админку: %s", telegram_id)
    return RedirectResponse(url=safe_redirect_target(request.query_params.get("next")), status_code=303)


@router.post("/logout")
async def logout(request: Request) -> Response:
    """Выйти: очистить сессию."""
    form = await request.form()
    check_csrf(request, str(form.get("csrf_token", "")))
    request.session.clear()
    return RedirectResponse(url="/login", status_code=303)