Рефакторинг - перенос пакета utils.drf в middleware
This commit is contained in:
75
net/xeaf/rack/middleware/expiring_token_authentication.py
Normal file
75
net/xeaf/rack/middleware/expiring_token_authentication.py
Normal file
@@ -0,0 +1,75 @@
|
||||
# DRF Rack
|
||||
# Библиотека классов расширений для Django REST Framework
|
||||
#
|
||||
# Автор: Николай В. Анохин <n.anokhin@xeaf.net>
|
||||
# Все права защищены. Лицензия: MIT
|
||||
|
||||
"""
|
||||
Описание класса ExpiringTokenAuthentication
|
||||
"""
|
||||
|
||||
from typing import Any
|
||||
|
||||
from rest_framework.authentication import TokenAuthentication
|
||||
from rest_framework.request import Request
|
||||
|
||||
from net.xeaf.rack.managers import AccountManager
|
||||
from net.xeaf.rack.managers import LocaleManager
|
||||
from net.xeaf.rack.managers import SessionManager
|
||||
from net.xeaf.rack.models import AccountModel
|
||||
from net.xeaf.rack.models import SessionModel
|
||||
|
||||
|
||||
class ExpiringTokenAuthentication(TokenAuthentication):
|
||||
"""
|
||||
Реализует механизм авторизации для Токенов со сроком действия
|
||||
"""
|
||||
|
||||
AUTH_TOKEN_PARAM = "x-auth-token"
|
||||
""" Идентификатор параметра токена """
|
||||
|
||||
def authenticate(self, request: Request) -> tuple[AccountModel | None, SessionModel | None] | tuple[Any, Any] | None:
|
||||
"""
|
||||
Метод авторизации
|
||||
|
||||
:param request: Информация о запросе
|
||||
"""
|
||||
|
||||
LocaleManager.adjust_request_language(request)
|
||||
|
||||
key = request.query_params.get(self.AUTH_TOKEN_PARAM, None)
|
||||
if key is not None:
|
||||
return self.authenticate_credentials(key)
|
||||
|
||||
return super().authenticate(request)
|
||||
|
||||
def authenticate_credentials(self, key: str) -> tuple[AccountModel | None, SessionModel | None]:
|
||||
"""
|
||||
Проверка Токена авторизации
|
||||
|
||||
:param key: Токен авторизации
|
||||
"""
|
||||
|
||||
session = SessionManager[SessionModel].find_by_token(token=key)
|
||||
if not session or session.is_expired():
|
||||
return None, None
|
||||
|
||||
account = session.account
|
||||
if not self._check_account(account):
|
||||
return None, None
|
||||
|
||||
SessionManager.renew_session(session)
|
||||
return session.account, session
|
||||
|
||||
@classmethod
|
||||
def _check_account(cls, account: AccountModel) -> bool:
|
||||
"""
|
||||
Проверка учетной записи
|
||||
|
||||
:param account: Учетная запись
|
||||
"""
|
||||
|
||||
# Можно прошедшим проверку и не удаленным
|
||||
reject = AccountManager.check_account_auth(account)
|
||||
result = reject is None and not account.is_deleted()
|
||||
return result
|
||||
Reference in New Issue
Block a user