99 lines
4.1 KiB
Python
99 lines
4.1 KiB
Python
"""Вход в админку через Telegram Login Widget."""
|
||
|
||
from __future__ import annotations
|
||
|
||
import logging
|
||
import secrets
|
||
|
||
from fastapi import APIRouter, Request, Response
|
||
from fastapi.responses import HTMLResponse, RedirectResponse
|
||
|
||
from admin.security import check_csrf, safe_redirect_target, verify_telegram_auth
|
||
from common import __version__
|
||
from common.repositories import 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),
|
||
"next": target,
|
||
},
|
||
)
|
||
|
||
|
||
@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)
|