fastapi-application/api/api_v1/auth/validation.py - main - misha_dev/game_guild_backend

fastapi-application/api/api_v1/auth/validation.py
99 строк3 KB
misha-dev
feat: categories crud
02 апр 2025, 18:50
02 апр 2025, 18:50196e207
from typing import Optional, Annotated
from fastapi import Depends, HTTPException
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
from jwt import InvalidTokenError
from sqlalchemy.ext.asyncio import AsyncSession
from starlette import status
from sqlalchemy import select

from api.api_v1.auth.helpers import TOKEN_TYPE, ACCESS_TOKEN_TYPE, REFRESH_TOKEN_TYPE
from core.models import db_helper
from core.schemas.auth import AuthRefreshTokenRequest
from utils import auth as auth_utils
from core.models.user import UserRole, User

http_bearer = HTTPBearer()


def validate_token_type(
    payload: dict,
    token_type: str,
) -> bool:
    current_token_type = payload.get(TOKEN_TYPE)
    if current_token_type == token_type:
        return True
    raise HTTPException(
        status_code=status.HTTP_401_UNAUTHORIZED,
        detail=f"Invalid token type",
    )


def get_token_payload(token: str) -> dict:
    try:
        payload = auth_utils.decode_jwt(token=token)
    except InvalidTokenError as e:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail=f"Invalid token error: {e}",
        )
    return payload


def get_current_token_payload(
    credentials: HTTPAuthorizationCredentials = Depends(http_bearer),
) -> dict:
    token = credentials.credentials
    return get_token_payload(token=token)


async def get_user_by_token_sub(
    payload: dict,
    session: AsyncSession,
) -> User:
    email: str | None = payload.get("sub")

    stmt = select(User).where(User.email == email)
    result = await session.execute(stmt)
    user = result.scalar_one_or_none()

    if not user:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Пользователь не найден",
        )

    return user


def check_user_role(user_role: UserRole, allowed_roles: Optional[list[UserRole]]):
    if allowed_roles and UserRole(user_role) not in allowed_roles:
        raise HTTPException(
            status_code=status.HTTP_403_FORBIDDEN,
            detail="You do not have permission to access this resource",
        )


def get_auth_user_with_role_check(allowed_roles: Optional[list[UserRole]] = None):
    async def get_auth_user_from_token(
        payload: dict = Depends(get_current_token_payload),
        session: AsyncSession = Depends(db_helper.session_getter),
    ) -> User:
        validate_token_type(payload, ACCESS_TOKEN_TYPE)

        user = await get_user_by_token_sub(payload, session)
        check_user_role(user.role, allowed_roles)
        return user

    return get_auth_user_from_token


async def get_current_auth_user_for_refresh(
    refresh_token: AuthRefreshTokenRequest,
    session: AsyncSession,
) -> User:
    payload = get_token_payload(refresh_token.refresh_token)

    validate_token_type(payload, REFRESH_TOKEN_TYPE)

    user = await get_user_by_token_sub(payload, session)
    return user