73 lines
3.0 KiB
Python
73 lines
3.0 KiB
Python
"""Зависимости FastAPI для аутентификации (RBAC): user / guest / admin."""
|
||
|
||
import uuid
|
||
from typing import Annotated
|
||
|
||
import jwt
|
||
from fastapi import Depends, HTTPException, status
|
||
from fastapi.security import OAuth2PasswordBearer
|
||
from sqlalchemy.ext.asyncio import AsyncSession
|
||
|
||
from core.db import get_session
|
||
from core.security import decode_token
|
||
from models.user import User
|
||
from repositories.users import UserRepository
|
||
|
||
# `auto_error=False`, чтобы отсутствие заголовка не приводило к автоматической
|
||
# ошибке — guest (отсутствие JWT) обрабатывается явно в get_current_user_optional.
|
||
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="/api/v1/auth/token", auto_error=False)
|
||
|
||
|
||
async def get_current_user(
|
||
token: Annotated[str | None, Depends(oauth2_scheme)],
|
||
session: Annotated[AsyncSession, Depends(get_session)],
|
||
) -> User:
|
||
"""Вернуть текущего пользователя по access-токену; 401 если не аутентифицирован."""
|
||
user = await _user_from_token(token, session)
|
||
if user is None:
|
||
raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="not_authenticated")
|
||
return user
|
||
|
||
|
||
async def get_current_user_optional(
|
||
token: Annotated[str | None, Depends(oauth2_scheme)],
|
||
session: Annotated[AsyncSession, Depends(get_session)],
|
||
) -> User | None:
|
||
"""Вернуть текущего пользователя либо `None` для guest (без ошибки).
|
||
|
||
Роль guest в системе — это отсутствие JWT, а не отдельное enum-значение в БД.
|
||
"""
|
||
return await _user_from_token(token, session)
|
||
|
||
|
||
async def require_admin(user: Annotated[User, Depends(get_current_user)]) -> User:
|
||
"""Требовать роль `admin`; иначе 403."""
|
||
if user.role != "admin":
|
||
raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="admin_required")
|
||
return user
|
||
|
||
|
||
async def _user_from_token(token: str | None, session: AsyncSession) -> User | None:
|
||
"""Общая логика резолва пользователя из access-токена (или None при любой проблеме).
|
||
|
||
Заблокированный администратором пользователь (`is_blocked`) трактуется
|
||
так же, как отсутствие пользователя — блокировка действует немедленно,
|
||
не дожидаясь истечения уже выданного access-токена.
|
||
"""
|
||
if token is None:
|
||
return None
|
||
try:
|
||
payload = decode_token(token)
|
||
except jwt.PyJWTError:
|
||
return None
|
||
if payload.get("type") != "access":
|
||
return None
|
||
try:
|
||
user_id = uuid.UUID(str(payload.get("sub")))
|
||
except (ValueError, TypeError):
|
||
return None
|
||
user = await UserRepository(session).get_by_id(user_id)
|
||
if user is not None and user.is_blocked:
|
||
return None
|
||
return user
|