bigdataAPI/utils/JWTAuthentication_diy.py

from rest_framework_jwt.serializers import jwt_encode_handler, jwt_decode_handler
from rest_framework.authentication import BaseAuthentication
from rest_framework.exceptions import AuthenticationFailed
import jwt
from .AESencipher import aescrypt
from .db_utils import MySQLTool
from bigdataAPI.settings import SECRET_KEY
from django.core.cache import cache as default_cache

PrAes = aescrypt(SECRET_KEY, 'ECB', '', 'gbk')


def get_token(user):
    data = str(user.id) + "," + user.username + "," + str(user.user_modules)
    data = PrAes.aesencrypt(data)
    payload = {"token": data}

    token = jwt_encode_handler(payload)
    return token


class MyJWTAuthentication(BaseAuthentication):
    def authenticate(self, request):
        token = request.META.get('HTTP_AUTHORIZATION')
        if token:
            try:
                payload = jwt_decode_handler(token)
                data = payload["token"]
                if '%2B' in data or " " in data:
                    data = str(data).replace("%2B", "+").replace(" ", "+")
                data = PrAes.aesdecrypt(data)
                data_list = data.split(",")
                user = {"uid": data_list[0], "username": data_list[1], "user_modules": data_list[2]}
                return (user, token)
            except jwt.ExpiredSignature:
                msg = 'token过期'
                raise AuthenticationFailed(msg)
            except jwt.DecodeError:
                msg = 'toke非法'
                raise AuthenticationFailed(msg)
            except jwt.InvalidTokenError:
                msg = '用户非法'
                raise AuthenticationFailed(msg)
            except Exception as e:
                msg = str(e)
                raise AuthenticationFailed(msg)
        raise AuthenticationFailed('token为空')


def API_get_uid(token):
    sql = f"select uid from sa_device_user where api_token='{token}';"
    m = MySQLTool()
    result = m.execute_by_one(sql)
    if result:
        uid = result['uid']
    else:
        uid = ""
    return uid    


class APIAuthentication(BaseAuthentication):
    cache = default_cache
    def authenticate(self, request):
        token = request.META.get('HTTP_AUTHORIZATION')
        if token:
            uid_cache = self.cache.get(token)
            if uid_cache:
                return (uid_cache,token)
            uid_sql = API_get_uid(token)
            if uid_sql:
                self.cache.set(token,uid_sql,60*1)
                return (uid_sql,token)
            
            raise AuthenticationFailed("校验失败")
        raise AuthenticationFailed('Authorization为空')