drf-rack/net/xeaf/rack/managers/session_manager.py

# DRF Rack
# Библиотека классов расширений для Django REST Framework
#
# Автор: Николай В. Анохин <n.anokhin@xeaf.net>
# Все права защищены. Лицензия: MIT

"""
Описание класса SessionManager
"""

from typing import NoReturn
from typing import Type

from django.apps import apps
from django.contrib.auth import authenticate
from environ import ImproperlyConfigured

from net.xeaf.rack.enums import AuthRejectReason
from net.xeaf.rack.managers import ConfigurationManager
from net.xeaf.rack.managers import CryptoManager
from net.xeaf.rack.models import AccountModel
from net.xeaf.rack.models import SessionModel
from net.xeaf.rack.utils.exceptions import UnauthorizedException
from .account_manager import AccountManager


class SessionManager[T:SessionModel]:
    """
    Реализует методы работы с Сессиями
    """

    DEFAULT_SESSION_TTL = 3600
    """ Время жизни сессии по умолчанию """

    __model_class: Type[T] | None = None
    """ Хранилище для класса сессии """

    @classmethod
    def authenticate(cls, username: str, password: str, device_id: str | None = None) -> T:
        """
        Авторизует новую сессию

        :param username: Имя пользователя
        :param password: Пароль
        :param device_id: Идентификатор устройства

        :return: Объект данных сессии
        """

        # Отсекаем ошибки активности и регистрации
        reason = AccountManager.check_username_auth(username)
        if reason is not None:
            cls._raise_auth_error(reason)

        # noinspection PyTypeChecker
        account: AccountModel | None = authenticate(username=username, password=password)

        # Обрабатываем ошибки проверки пароля
        if account is None:
            cls._raise_auth_error(AuthRejectReason.INVALID_USERNAME_OR_PASSWORD)

        # Создаем и возвращаем объект сессии
        AccountManager.restore_account(account)
        session = cls.open_session(account, account, device_id)
        return session

    @classmethod
    def find_by_token(cls, token: str) -> T | None:
        """
        Возвращает объект данных сессии по токену

        :param token: Токен

        :return: Объект данных сессии или None, если не найдена или просрочена
        """

        model = cls._get_session_model_class()
        session = model.objects.filter(token__exact=token).first()
        if session is not None and not session.is_expired():
            return session

        return None

    @classmethod
    def open_session(cls, account: AccountModel, actual_account: AccountModel, device_id: str | None = None) -> T:
        """
        Открывает сессию для аккаунта

        :param account: Модель данных Аккаунта пользователя
        :param actual_account: Актуальный Аккаунт пользователя
        :param device_id: Идентификатор устройства

        :return: Информация о сессии
        """

        session = cls._create_session_model(account, actual_account)
        session.device_id = device_id
        cls.renew_session(session)
        return session

    @classmethod
    def close_session(cls, session: T) -> None:
        """
        Закрывает сессию пользователя

        :param session: Модель данных сессии
        """

        session.expire(save=True)

    @classmethod
    def renew_session(cls, session: T) -> None:
        """
        Обновляет время существование сессии

        :param session: Модель данных сессии
        """

        ttl = ConfigurationManager.get_int("SESSION_TTL", cls.DEFAULT_SESSION_TTL)
        session.renew(ttl, save=True)

    @classmethod
    def _create_session_model(cls, account: AccountModel, actual_account: AccountModel) -> T:
        """
        Создает новую модель Сессии

        :param account: Аккаунт пользователя
        :param actual_account: Актуальный Аккаунт пользователя
        """

        model = cls._get_session_model_class()
        return model.objects.create(
            token=CryptoManager.gen_token(),
            account=account,
            actual_account=actual_account,
        )

    @classmethod
    def _get_session_model_class(cls) -> Type[T]:
        """
        Возвращает класс модели данных сессии

        :return: Класс модели данных сессии
        """

        if cls.__model_class is None:
            model_path = ConfigurationManager.get_string("AUTH_SESSION_MODEL", "net_xeaf_rack.SessionModel")
            app_label, model_name = model_path.split(".")

            try:
                model = apps.get_model(app_label, model_name)
            except LookupError:
                model = None

            # Нет класса
            if not model:
                raise ImproperlyConfigured("Correct AUTH_SESSION_MODEL must be set in settings.py")

            # Некорректный класс
            if not issubclass(model, SessionModel):
                raise ImproperlyConfigured(f"AUTH_SESSION_MODEL must be a subclass of {SessionModel.__name__}")

            cls.__model_class = model

        # noinspection PyTypeChecker
        return cls.__model_class

    @classmethod
    def _raise_auth_error(cls, reason: AuthRejectReason) -> NoReturn:
        """
        Выбрасывает исключение с ошибкой аутентификации

        :param reason: Причина ошибки
        """

        detail = AuthRejectReason.get_message(reason)
        raise UnauthorizedException(detail=detail, reason=reason)