48 lines
1.9 KiB
Python
48 lines
1.9 KiB
Python
"""Middleware админки: пускает дальше только вошедших администраторов."""
|
||
|
||
from __future__ import annotations
|
||
|
||
import logging
|
||
import secrets
|
||
from urllib.parse import quote
|
||
|
||
from starlette.middleware.base import BaseHTTPMiddleware, RequestResponseEndpoint
|
||
from starlette.requests import Request
|
||
from starlette.responses import RedirectResponse, Response
|
||
|
||
from common.repositories import admins_repo
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
#: Эндпоинты без авторизации.
|
||
PUBLIC_PATHS = frozenset({"/healthz", "/login", "/auth/telegram"})
|
||
|
||
|
||
class AdminAuthMiddleware(BaseHTTPMiddleware):
|
||
"""Сессия + роль из БД на каждом запросе (кроме служебных и логина)."""
|
||
|
||
async def dispatch(self, request: Request, call_next: RequestResponseEndpoint) -> Response:
|
||
if request.url.path in PUBLIC_PATHS:
|
||
return await call_next(request)
|
||
|
||
admin_id = request.session.get("admin_id") if "session" in request.scope else None
|
||
if not admin_id:
|
||
target = quote(request.url.path)
|
||
return RedirectResponse(url=f"/login?next={target}", status_code=303)
|
||
|
||
try:
|
||
admin = await admins_repo.get_admin(int(admin_id))
|
||
except Exception:
|
||
logger.exception("Не удалось проверить администратора")
|
||
return RedirectResponse(url="/login", status_code=303)
|
||
|
||
if admin is None or not admin["enabled"]:
|
||
request.session.clear()
|
||
return RedirectResponse(url="/login", status_code=303)
|
||
|
||
request.state.admin = admin
|
||
# CSRF-токен живёт в сессии; если её создали раньше — добираем.
|
||
if not request.session.get("csrf"):
|
||
request.session["csrf"] = secrets.token_urlsafe(32)
|
||
return await call_next(request)
|