42 lines
1.7 KiB
Python
42 lines
1.7 KiB
Python
from __future__ import annotations
|
|
|
|
from django.utils import timezone
|
|
from rest_framework.authentication import BaseAuthentication
|
|
from rest_framework.exceptions import AuthenticationFailed, PermissionDenied
|
|
|
|
from apps.api.errors import api_error
|
|
from apps.api.throttles import throttle_api_auth_failure
|
|
from apps.licensing.models import ClientDevice, DeviceSession
|
|
|
|
|
|
class DeviceSessionAuthentication(BaseAuthentication):
|
|
header_name = "HTTP_X_DEVICE_SESSION"
|
|
|
|
def authenticate(self, request):
|
|
raw_token = request.META.get(self.header_name, "").strip()
|
|
if not raw_token:
|
|
return None
|
|
|
|
try:
|
|
session = DeviceSession.objects.select_related("device", "device__user").get(
|
|
token_hash=DeviceSession.hash_token(raw_token)
|
|
)
|
|
except DeviceSession.DoesNotExist as exc:
|
|
raise self.authentication_failed(request) from exc
|
|
|
|
if not session.matches_token(raw_token) or not session.is_active_at(timezone.now()):
|
|
raise self.authentication_failed(request)
|
|
if session.device.status != ClientDevice.Status.ACTIVE:
|
|
raise PermissionDenied(api_error("device_revoked", "设备已被吊销"))
|
|
if not session.device.user.is_business_active:
|
|
raise PermissionDenied(api_error("account_disabled", "账号或设备已禁用"))
|
|
return session.device.user, session
|
|
|
|
@staticmethod
|
|
def authentication_failed(request):
|
|
throttle_api_auth_failure(request)
|
|
return AuthenticationFailed(api_error("device_session_invalid", "缺失或无效设备会话"))
|
|
|
|
def authenticate_header(self, request) -> str:
|
|
return "DeviceSession"
|