使用jwt时,使用自己的用户类校验

Posted 2022-06-02 robert-zhou

坏境:
web端使用django的auth_user表进行用户登陆认证
由于某种原因app端需使用新用户表进行用户登陆
都采用jwt 验证

基于jwt认证的登陆

class JSONWebTokenAuthentication(BaseJSONWebTokenAuthentication):
    www_authenticate_realm = 'api'

    def get_jwt_value(self, request):
        auth = get_authorization_header(request).split()
        auth_header_prefix = api_settings.JWT_AUTH_HEADER_PREFIX.lower()

        if not auth or smart_text(auth[0].lower()) != auth_header_prefix:
            return None

        if len(auth) == 1:
            msg = _('Invalid Authorization header. No credentials provided.')
            raise exceptions.AuthenticationFailed(msg)
        elif len(auth) > 2:
            msg = _('Invalid Authorization header. Credentials string '
                    'should not contain spaces.')
            raise exceptions.AuthenticationFailed(msg)

        return auth[1]

    def authenticate_header(self, request):
        return 'JWT realm="0"'.format(self.www_authenticate_realm)

执行顺序:

1.先执行BaseJSONWebTokenAuthentication的authenticate方法
    def authenticate(self, request):
        jwt_value = self.get_jwt_value(request)
        ....
        return (user, jwt_value)
    
2.执行JSONWebTokenAuthentication的get_jwt_value方法
 def get_jwt_value(self, request):
        #split空格,我们传的jwt格式,以空格分隔
        """
        jwt eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJ1c2VybmFtZSI6ImRlbW9jaGluYV9jbiIsInVzZXJfaWQiOjE1MzIsImVtYWlsIjoiMTYzdkBxcS5jbyIsImV4cCI6MTU3MjIzMDI2NX0.f7sQ8FSAmkiatiIbJ2qVgkdCC9bOj_nU0kfr3LuHwdE
        """
        auth = get_authorization_header(request).split()
        auth_header_prefix = api_settings.JWT_AUTH_HEADER_PREFIX.lower()

        #auth[0] jwt是否等于auth_header_prefix jwt
        if not auth or smart_text(auth[0].lower()) != auth_header_prefix:
            return None

        if len(auth) == 1:
            msg = _('Invalid Authorization header. No credentials provided.')
            raise exceptions.AuthenticationFailed(msg)
        elif len(auth) > 2:
            msg = _('Invalid Authorization header. Credentials string '
                    'should not contain spaces.')
            raise exceptions.AuthenticationFailed(msg)

        return auth[1]#返回的是token
    
  3.get_authorization_header(request)方法
def get_authorization_header(request):
    auth = request.META.get('HTTP_AUTHORIZATION', b'')
    if isinstance(auth, type('')):
        auth = auth.encode(HTTP_HEADER_ENCODING)
    return auth

4.回到BaseJSONWebTokenAuthentication的authenticate方法
     def authenticate(self, request):
        #返回jwt token
        jwt_value = self.get_jwt_value(request)
        if jwt_value is None:
            return None

        try:
            #反解jwt token
            payload = jwt_decode_handler(jwt_value)
            #超时错误
        except jwt.ExpiredSignature:
            msg = _('Signature has expired.')
            raise exceptions.AuthenticationFailed(msg)
            #解码错误
        except jwt.DecodeError:
            msg = _('Error decoding signature.')
            raise exceptions.AuthenticationFailed(msg)
        except jwt.InvalidTokenError:
            raise exceptions.AuthenticationFailed()

        user = self.authenticate_credentials(payload)
        
        return (user, jwt_value)

  5.
    def authenticate_credentials(self, payload):
        #获取默认的用户表
        User = utils.get_user_model()
        #通过jwt token获取用户id  生成jwt token的时候封装进去的
        user_id = jwt_get_user_id_from_payload(payload)

        if user_id is not None:
            try:
                user = User.objects.get(pk=user_id, is_active=True)
            except User.DoesNotExist:
                msg = _('Invalid signature.')
                raise exceptions.AuthenticationFailed(msg)
        else:
            msg = _('Invalid payload.')
            raise exceptions.AuthenticationFailed(msg)

        return user
   

需求:

web端登陆使用User表, app端使用CourseUser表

因为判断用户是否存在,是在authenticate_credentials()方法中实现的

故重写authenticate_credentials()方法,在CourseUser表中查找

class TokenAuthentication(JSONWebTokenAuthentication):

    def authenticate(self, request):
        credentials = super(TokenAuthentication, self).authenticate(request)
        if not credentials:
            msg = "Authentication credenttials were not provided"
            raise exceptions.AuthenticationFailed(msg)
        request.user, jwt = credentials
        return credentials

    def authenticate_credentials(self, payload):
        user_id = jwt_get_user_id_from_payload(payload)
        if user_id is not None:
            try:
                # user = User.objects.get(pk=user_id, is_active=True) or Userinfo.objects.get(pk=user_id)
                user = CourseUser.objects.get(pk=user_id)
            except :
                msg = 'Invalid signature.'
                raise exceptions.AuthenticationFailed(msg)
        else:
            msg = 'Invalid payload.'
            raise exceptions.AuthenticationFailed(msg)
        return user

登陆生成jwt token

• Header 头部

头部包含了两部分，token 类型和采用的加密算法

  "alg": "HS256",
  "typ": "JWT"

它会使用 Base64 编码组成 JWT 结构的第一部分,如果你使用Node.js，可以用Node.js的包base64url来得到这个字符串。

Base64是一种编码，也就是说，它是可以被翻译回原来的样子来的。它并不是一种加密过程。

• Payload 负载

这部分就是我们存放信息的地方了，你可以把用户 ID 等信息放在这里，JWT 规范里面对这部分有进行了比较详细的介绍，常用的由 iss（签发者），exp（过期时间），sub（面向的用户），aud（接收方），iat（签发时间）。

    "iss": "lion1ou JWT",
    "iat": 1441593502,
    "exp": 1441594722,
    "aud": "www.example.com",
    "sub": "[e-mail]"

同样的，它会使用 Base64 编码组成 JWT 结构的第二部分

• Signature 签名

前面两部分都是使用 Base64 进行编码的，即前端可以解开知道里面的信息。Signature 需要使用编码后的 header 和 payload 以及我们提供的一个密钥，然后使用 header 中指定的签名算法（HS256）进行签名。签名的作用是保证 JWT 没有被篡改过。

详情请查看这篇文章? https://www.jianshu.com/p/180a870a308a

def get_jwt(user):
    jwt_encode_handler = api_settings.JWT_ENCODE_HANDLER
    #头部 user_id
    #exp 过期时间
    payload = 
        'user_id': user.pk,
        'exp': datetime.datetime.utcnow() + datetime.timedelta(days=365)
    
    #生成token
    token = jwt_encode_handler(payload)
    return "token": token, "user_id": user.pk