auth/authenticat.py - master - developer_atatanoff/simple_auth_fastapi

auth/authenticat.py
178 строк6 KB
Alexey N. Tatanoff
Добавлен csrf сертификация
05 янв 2026, 18:16
05 янв 2026, 18:16cc27909
from fastapi.security.utils import get_authorization_scheme_param
from auth.schemas import CurrentUser, AuthSession, UserOut
from time import time
from datetime import datetime, timezone
from typing import Protocol
from config.logger import logger
from passlib.context import CryptContext
from pydantic_core import ValidationError


class Token(Protocol):
    def decode_token(self, data):
        pass

    async def create_access_token(self, data: dict):
        pass

    def encode_token(self, to_encode: str):
        pass

    async def verify_access_token(self, token: str):
        pass

    def generate_token(self):
        pass
    
    def is_valid_csrftoken(self, request):...


class AuthRepository(Protocol):
    token: Token

    async def get_user(self, key, item) -> UserOut:
        pass

    async def get_session(self, key):
        pass

    async def create_refresh_token(self, id, refresh_token):
        pass

    async def delete_refresh_token(self, key):
        pass


class AuthenticatBase:
    def __init__(self, auth_repository: AuthRepository):
        self._auth_repository = auth_repository

    async def authenticate(self, request):
        raise NotImplementedError("Подклассы должны реализовать этот метод!")

    async def get_curent_user(self, user):
        c_user: CurrentUser = CurrentUser.model_validate(user)
        payload = c_user.model_dump(
            exclude={"access_token", "refresh_token", "csrftoken"}
        )
        c_user.access_token = await self._auth_repository.token.create_access_token(
            payload
        )
        c_user.refresh_token = await self._auth_repository.create_refresh_token(payload)
        c_user.csrftoken = (self._auth_repository.token.generate_token(),)

        return c_user


class AuthenticatToken(AuthenticatBase):
    async def authenticate(self, request):
        auth_header = request.headers.get("Authorization")
        scheme, param = get_authorization_scheme_param(auth_header)
        if not auth_header or scheme.lower() != "token":
            return False
        return await self.__is_token(param)

    async def __is_token(self, token):
        # мыряем в базу проверям есть ли тех токен
        user: UserOut = await self._auth_repository.get_user("token", token)
        if user and user.is_active:
            c_user = CurrentUser.model_validate(user)
            c_user.groups.append("token")
            return c_user



class AuthenticatAccessToken(AuthenticatBase):
    async def authenticate(self, request):
        # не хедер а куки
        access_token = request.cookies.get("access_token")
        refresh_token = request.cookies.get("refresh_token")
        if not self._auth_repository.token.is_valid_csrftoken(request):
            return False
        # if not access_token and not refresh_token:
        if not refresh_token:
            return False
        user = await self.__is_access_token(access_token, refresh_token)
        return user if user else await self.__is_refresh_token(refresh_token)

    async def __is_access_token(self, access_token, refresh_token):
        if not access_token:
            return False
        payload: dict | False = await self._auth_repository.token.verify_access_token(
            access_token
        )

        if payload:
            return CurrentUser(
                access_token=await self._auth_repository.token.create_access_token(
                    payload
                ),
                refresh_token=refresh_token,
                csrftoken=self._auth_repository.token.generate_token(),
                **payload,
            )
        else:
            return False

    async def __is_refresh_token(self, token):
        session_token: AuthSession = await self._auth_repository.get_session(token)
        if not session_token:
            return False

        current_time = datetime.now(timezone.utc)
        expire_date = session_token.expire_date
        if expire_date.tzinfo is None:
            # Если почему-то naive, преобразуем в UTC
            expire_date = expire_date.replace(tzinfo=timezone.utc)
        if current_time >= expire_date:
            await self._auth_repository.delete_refresh_token(token)
            return False

        payload = self._auth_repository.token.decode_token(session_token.session_data)
        try:
            return CurrentUser(
                access_token=await self._auth_repository.token.create_access_token(
                    payload
                ),
                refresh_token=token,
                csrftoken=self._auth_repository.token.generate_token(),
                **payload,
            )
        except ValidationError as e:
            logger.error(e)
            return False


class AuthenticatRemoteUser(AuthenticatBase):
    async def authenticate(self, request):
        if not self._auth_repository.token.is_valid_csrftoken(request):
            return False
        remote_user = request.headers.get("Remote-User")
        if not remote_user:
            return False
        return await self.__is_user(remote_user)

    async def __is_user(self, username, *arg):
        user = await self._auth_repository.get_user("username", username)
        if user and user.is_active:
            c_user = await self.get_curent_user(user)
        return c_user


class AuthenticatLogin(AuthenticatBase):
    
    pwd_context = CryptContext(schemes=["argon2"], deprecated="auto")

    async def authenticate(self, user_data, request):
        if not self._auth_repository.token.is_valid_csrftoken(request):
            return False
        user = await self.__is_user(user_data.username, user_data.password)
        if user:
            return await self.get_curent_user(user)
        return None

    async def __is_user(self, username: str, password: str):
        user = await self._auth_repository.get_user("username", username)
        if user and user.is_active and self.pwd_context.verify(password, user.password):
            return user
        return None