mirror of
https://github.com/1Panel-dev/MaxKB.git
synced 2025-12-26 01:33:05 +00:00
52 lines
1.9 KiB
Python
52 lines
1.9 KiB
Python
# coding=utf-8
|
||
"""
|
||
@project: qabot
|
||
@Author：虎
|
||
@file： authenticate.py
|
||
@date：2023/9/4 11:16
|
||
@desc: 认证类
|
||
"""
|
||
|
||
from common.constants.permission_constants import Auth, get_permission_list_by_role, RoleConstants
|
||
from common.exception.app_exception import AppAuthenticationFailed
|
||
from django.core import cache
|
||
from django.core import signing
|
||
from rest_framework.authentication import TokenAuthentication
|
||
from smartdoc.settings import JWT_AUTH
|
||
from users.models.user import User
|
||
|
||
token_cache = cache.caches['token_cache']
|
||
|
||
|
||
class AnonymousAuthentication(TokenAuthentication):
|
||
def authenticate(self, request):
|
||
return None, None
|
||
|
||
|
||
class TokenAuth(TokenAuthentication):
|
||
# 重新 authenticate 方法，自定义认证规则
|
||
def authenticate(self, request):
|
||
auth = request.META.get('HTTP_AUTHORIZATION', None
|
||
)
|
||
# 未认证
|
||
if auth is None:
|
||
raise AppAuthenticationFailed(1003, '未登录,请先登录')
|
||
try:
|
||
# 解析 token
|
||
user = signing.loads(auth)
|
||
if 'id' in user:
|
||
cache_token = token_cache.get(auth)
|
||
if cache_token is None:
|
||
raise AppAuthenticationFailed(1002, "登录过期")
|
||
user = User.objects.get(id=user['id'])
|
||
# 续期
|
||
token_cache.touch(auth, timeout=JWT_AUTH['JWT_EXPIRATION_DELTA'].total_seconds())
|
||
rule = RoleConstants[user.role]
|
||
return user, Auth(role_list=[rule],
|
||
permission_list=get_permission_list_by_role(RoleConstants[user.role]))
|
||
else:
|
||
raise AppAuthenticationFailed(1002, "身份验证信息不正确！非法用户")
|
||
|
||
except Exception as e:
|
||
raise AppAuthenticationFailed(1002, "身份验证信息不正确！非法用户")
|