1.jwt认证流程

在项目开发中，一般某些url过程进行认证，即：用户登录成功之后，服务端给用户浏览器返回一个token，以后用户浏览器要携带token再去向服务端发送请求，服务端校验token的合法性，合法则给用户看数据，否则，返回一些错误信息。

传统token方式和jwt在认证方面的差异 - 传统token认证用户登录成功后，服务端生成一个随机token给用户，并且在服务端(数据库或缓存)中保存一份token，以后用户再来访问时需携带token，服务端接收到token之后，去数据库或缓存中进行校验token的是否超时、是否合法。 - jtw认证用户登录成功后，服务端通过jwt生成一个随机token给用户（服务端无需保留token），以后用户再来访问时需携带token，服务端接收到token之后，通过jwt对token进行校验是否超时、是否合法。

2.token值解析

详情查看jwt官网：jwt官网 jwt的token有三端字符串组成，以.进行切割例如： eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJzdWIiOiIxMjM0NTY3ODkwIiwibmFtZSI6IkpvaG4gRG9lIiwiaWF0IjoxNTE2MjM5MDIyfQ.SflKxwRJSMeKKF2QT4fwpMeJf36POk6yJV_adQssw5c

第一段header部分json信息，包含算法和token类型，对此进行base64url加密 - eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9

{
  "alg": "HS256",
  "typ": "JWT"
}

第二段PAYLOAD,包含用户的一些数据，也进行base64url加密，token过期时间也包含在此 - eyJzdWIiOiIxMjM0NTY3ODkwIiwibmFtZSI6IkpvaG4gRG9lIiwiaWF0IjoxNTE2MjM5MDIyfQ

{
  "sub": "1234567890",
  "name": "John Doe",
  "iat": 1516239022
}

第三段SIGNATURE部分，把前两段的base密文通过.拼接起来，然后对其进行HS256加密，再然后对hs256密文进行base64url加密，最终得到token的第三段。

SflKxwRJSMeKKF2QT4fwpMeJf36POk6yJV_adQssw5c

    base64url(
        HMACSHA256(
          base64UrlEncode(header) + "." + base64UrlEncode(payload),
          your-256-bit-secret (秘钥加盐)
        )
    )

最后将三段字符串通过 .拼接起来就生成了jwt的token。

注意：base64url加密是先做base64加密，然后再将 - 替代 + 及 _ 替代 / 。

3.代码演示

pip3 install pyjwt
安装模块

up主会用登录接口，获取订单接口（需要token认证）进行实例，下面是总的路由

urls.py

from django.urls import path
from api import views

urlpatterns = [
    #服务端返回token，并将token保存在服务端，用户再来访问时，需要携带token，服务端获取token后，再去数据库获取token进行校验
    path('login/', views.LoginView.as_view(),),
    path('order/', views.OrderView.as_view(),),
    #jwt 服务端给用户返回一个token 服务端不保存，用户再来访问时，需要携带token，服务端获取token后，再去数据库获取token进行校验
    path('jwt/login/', views.JwtLoginView.as_view(),),
    path('jwt/order/', views.JwtOrderView.as_view(),),
    # 正式
    path('pro/login/', views.ProLoginView.as_view(), ),
    path('pro/order/', views.ProOrderView.as_view(), ),
]

3.1 最原始的token认证

用户名，密码，token三个字段

from django.db import models

class UserInfo(models.Model):
    username = models.CharField(max_length=255)
    password = models.CharField(max_length=255)
    token = models.CharField(max_length=255,null=True,blank=True)

from rest_framework.views import APIView
from rest_framework.response import Response
from api import models
import uuid,datetime,jwt
from jwt import exceptions

class LoginView(APIView):
    '''
    用户登录
    '''
    def post(self, request, *args, **kwargs):
        username = request.data.get("username")
        password = request.data.get("password")
        user_obj = models.UserInfo.objects.filter(username=username, password=password).first()
        if not user_obj:
            return Response({'code':1000,'error':"用户名或者密码错误"})
        random_str = str(uuid.uuid4())
        user_obj.token = random_str
        user_obj.save()
        return Response({'code':1001,'data':random_str})

class OrderView(APIView):
    def get(self,request,*args,**kwargs):
        token = request.query_params.get('token')
        if not token:
            return Response({'code': 1000, "error": "登录成功后才可以访问"})
        user_obj = models.UserInfo.objects.filter(token=token).first()
        if not user_obj:
            return Response({'code': 1000, "error": "token无效"})
        return Response({'code':1002,"orderlist":"订单列表"})


eg:
http://127.0.0.1:8000/api/v1/login/
请求类型：post
请求参数：{"username":"kkxl","password":"123"}
返回值：
{
    "code": 1001,
    "data": "87308b83-cd20-41f1-9af5-b72baa5b7a34"
}
------------------------------------------------
http://127.0.0.1:8000/api/v1/order/?token=87308b83-cd20-41f1-9af5-b72baa5b7a34
请求类型：get
返回值：{'code':1002,"orderlist":"订单列表"}

3.2 使用jwt认证

SALT="asda1a51f0awe415ac1awsdazqwe21323zc"
class JwtLoginView(APIView):
    '''
    用户登录
    '''

    def post(self, request, *args, **kwargs):
        username = request.data.get("username")
        password = request.data.get("password")
        user_obj = models.UserInfo.objects.filter(username=username, password=password).first()
        if not user_obj:
            return Response({'code': 1000, 'error': "用户名或者密码错误"})

        headers = {
            'typ': 'jwt',
            'alg': 'HS256'
        }
        # 构造payload
        payload = {
            'user_id': user_obj.id,  # 自定义用户ID
            'username': user_obj.username,  # 自定义用户名
            'exp': datetime.datetime.utcnow() + datetime.timedelta(minutes=1)  # 超时时间
        }
        result = jwt.encode(payload=payload, key=SALT, algorithm="HS256", headers=headers).decode('utf-8')
        # random_str = str(uuid.uuid4())
        # user_obj.token = random_str
        # user_obj.save()
        return Response({'code': 1001, 'data': result})


class JwtOrderView(APIView):
    def get(self, request, *args, **kwargs):
        token = request.query_params.get('token')
        #获取token后的三步骤
        #1.切割
        #2.解密第二段。判断过期值
        #3.验证第三段合法性 是否发生篡改
        verified_payload = None
        msg = None
        try:
            # 从token中获取payload【不校验合法性】
            # unverified_payload = jwt.decode(token, None, False)
            # print(unverified_payload)
            # 从token中获取payload【校验合法性】
            verified_payload = jwt.decode(token, SALT, True) #集成盐 时间的校验
        except exceptions.ExpiredSignatureError:
            msg = 'token已失效'
        except jwt.DecodeError:
            msg = 'token认证失败'
        except jwt.InvalidTokenError:
            msg = '非法的token'
        if not verified_payload:
            return Response({'code': 1003, "error": msg})
        # token校验通过
        print(verified_payload['user_id'],verified_payload['username'])
        return Response({'code': 200,"orderlist": "订单列表。。。"})

eg：
http://127.0.0.1:8000/api/v1/jwt/login/
请求类型：post
请求参数：{"username":"kkxl","password":"123"}
返回值：
{
    "code": 1001,
    "data": "eyJ0eXAiOiJqd3QiLCJhbGciOiJIUzI1NiJ9.eyJ1c2VyX2lkIjoxLCJ1c2VybmFtZSI6ImtreGwiLCJleHAiOjE2NDA4NTkxOTF9.amTjG67AuIsE43WGOHc7z1KTU3uyclIcoSVkA5ZIf_4"
}
---------------------------------------------------------
http://127.0.0.1:8000/api/v1/jwt/order/?token=eyJ0eXAiOiJqd3QiLCJhbGciOiJIUzI1NiJ9.eyJ1c2VyX2lkIjoxLCJ1c2VybmFtZSI6ImtreGwiLCJleHAiOjE2NDA4NTkxOTF9.amTjG67AuIsE43WGOHc7z1KTU3uyclIcoSVkA5ZIf_4
请求类型：get
返回值：{'code': 200,"orderlist": "订单列表。。。"}

3.3 jwt封装使用

在app下新建一个文件夹extensions 新建python文件auth.py 重写认证

from rest_framework.authentication import BaseAuthentication
import jwt
from django.conf import settings
from rest_framework.exceptions import AuthenticationFailed #抛出的异常
from jwt import exceptions

#自定义认证组件
class JwtQueryParamsAuthentication(BaseAuthentication):
    def authenticate(self, request):
        verified_payload = None
        token = request.query_params.get('token')
        SALT = settings.SECRET_KEY
        try:
            # 从token中获取payload【不校验合法性】
            # unverified_payload = jwt.decode(token, None, False)
            # print(unverified_payload)
            # 从token中获取payload【校验合法性】
            verified_payload = jwt.decode(token, SALT, True)  # 集成盐 时间的校验
        except exceptions.ExpiredSignatureError:
            raise AuthenticationFailed({"code":1003,"error":"token已失效"})
        except jwt.DecodeError:
            raise AuthenticationFailed({"code": 1003, "error": "token认证失败"})
        except jwt.InvalidTokenError:
            raise AuthenticationFailed({"code": 1003, "error": "非法token"})
        # 可以有三种返回值
        #1.抛出异常，后续不执行
        #2.return一个元组 （1，2）认证通过；在视图中如果调用request.user就是元组中的第一个值：request.auth
        return (verified_payload, token)

在app下新建一个文件夹utils 新建python文件jwt_auth.py

import time,jwt,datetime
from django.conf import settings

def create_token(payload,timeout=720):
    SALT = settings.SECRET_KEY
    headers = {
        'typ': 'jwt',
        'alg': 'HS256'
    }
    # 构造payload
    payload['exp'] = datetime.datetime.utcnow() + datetime.timedelta(minutes=timeout)
    token = jwt.encode(payload=payload, key=SALT, algorithm="HS256", headers=headers).decode('utf-8')
    return token

正式视图

#导入重构的认证组件
from api.extensions.auth import JwtQueryParamsAuthentication
from api.utils.jwt_auth import create_token
class ProLoginView(APIView):
    authentication_classes = [] #为空表示此方法不用token认证
    def post(self, request, *args, **kwargs):
        username = request.data.get("username")
        password = request.data.get("password")
        user_obj = models.UserInfo.objects.filter(username=username, password=password).first()
        if not user_obj:
            return Response({'code':1000,'error':"用户名或者密码错误"})
        token = create_token({"id":user_obj.id, "name":user_obj.username})
        return Response({'code': 1001, 'data': token})

class ProOrderView(APIView):
    # authentication_classes = [JwtQueryParamsAuthentication,] #需要token认证 全局写在配置文件 默认用认证
    def get(self, request, *args, **kwargs):
        print(request.user)
        return Response({'code': 200,"orderlist": "订单列表。。。"})

eg：
http://127.0.0.1:8000/api/v1/pro/login/
请求类型：post
请求参数：{"username":"kkxl","password":"123"}
返回值：
{
    "code": 1001,
    "data": "eyJ0eXAiOiJqd3QiLCJhbGciOiJIUzI1NiJ9.eyJpZCI6MSwibmFtZSI6ImtreGwiLCJleHAiOjE2NDA5MDI5Mjh9.IbhIBM8IRD1UFU3inV0pVqQLAe4PFJ1yuANx1aAeTk8"
}
-------------------------
http://127.0.0.1:8000/api/v1/pro/order/?token=eyJ0eXAiOiJqd3QiLCJhbGciOiJIUzI1NiJ9.eyJpZCI6MSwibmFtZSI6ImtreGwiLCJleHAiOjE2NDA5MDI5Mjh9.IbhIBM8IRD1UFU3inV0pVqQLAe4PFJ1yuANx1aAeTk8

请求类型：get
返回值：
{
    "code": 200,
    "orderlist": "订单列表。。。"
}

3.3.1 封装

token信息丢在请求头 - views.py

#导入重构的认证组件
from api.extensions.auth import JwtAuthorizationAuthentication,JwtQueryParamAuthentication
from api.utils.jwt_auth import create_token
class ProLoginView(APIView):
    authentication_classes = [] #为空表示此方法不用token认证
    def post(self, request, *args, **kwargs):
        username = request.data.get("username")
        password = request.data.get("password")
        user_obj = models.UserInfo.objects.filter(username=username, password=password).first()
        if not user_obj:
            return Response({'code':1000,'error':"用户名或者密码错误"})
        token = create_token({"id":user_obj.id, "name":user_obj.username})
        return Response({'status': "登录成功", 'token': token})

class ProOrderView(APIView):
    # 通过url传递token
    #authentication_classes = [JwtQueryParamAuthentication, ]

    # 通过Authorization请求头传递token 前后端分离常用
    authentication_classes = [JwtAuthorizationAuthentication, ]

    def get(self, request, *args, **kwargs):
        print(request.user, request.auth)
        return Response({'data': '订单列表'})

    def post(self, request, *args, **kwargs):
        print(request.user, request.auth)
        return Response({'data': '添加订单'})

    def put(self, request, *args, **kwargs):
        print(request.user, request.auth)
        return Response({'data': '修改订单'})

    def delete(self, request, *args, **kwargs):
        print(request.user, request.auth)
        return Response({'data': '删除订单'})

auth.py

from rest_framework.authentication import BaseAuthentication
from rest_framework import exceptions
from api.utils.jwt_auth import parse_payload


class JwtQueryParamAuthentication(BaseAuthentication):
    """
    用户需要在url中通过参数进行传输token，例如：
    http://www.pythonav.com?token=eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJleHAiOjE1NzM1NTU1NzksInVzZXJuYW1lIjoid3VwZWlxaSIsInVzZXJfaWQiOjF9.xj-7qSts6Yg5Ui55-aUOHJS4KSaeLq5weXMui2IIEJU
    """

    def authenticate(self, request):
        token = request.query_params.get('token')
        payload = parse_payload(token)
        if not payload['status']:
            raise exceptions.AuthenticationFailed(payload)

        # 如果想要request.user等于用户对象，此处可以根据payload去数据库中获取用户对象。
        return (payload, token)


class JwtAuthorizationAuthentication(BaseAuthentication):
    """
    用户需要通过请求头的方式来进行传输token，例如：
    Authorization:jwt eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJleHAiOjE1NzM1NTU1NzksInVzZXJuYW1lIjoid3VwZWlxaSIsInVzZXJfaWQiOjF9.xj-7qSts6Yg5Ui55-aUOHJS4KSaeLq5weXMui2IIEJU
    """

    def authenticate(self, request):

        # 非登录页面需要校验token
        authorization = request.META.get('HTTP_AUTHORIZATION', '')
        auth = authorization.split()
        if not auth:
            raise exceptions.AuthenticationFailed({'error': '未获取到Authorization请求头', 'status': False})
        if auth[0].lower() != 'jwt':
            raise exceptions.AuthenticationFailed({'error': 'Authorization请求头中认证方式错误', 'status': False})

        if len(auth) == 1:
            raise exceptions.AuthenticationFailed({'error': "非法Authorization请求头", 'status': False})
        elif len(auth) > 2:
            raise exceptions.AuthenticationFailed({'error': "非法Authorization请求头", 'status': False})

        token = auth[1]
        result = parse_payload(token)
        if not result['status']:
            raise exceptions.AuthenticationFailed(result)

        # 如果想要request.user等于用户对象，此处可以根据payload去数据库中获取用户对象。
        return (result, token)

jwt_auth.py

#!/usr/bin/env python
# -*- coding:utf-8 -*-
import jwt
import datetime
from jwt import exceptions
from django.conf import settings

JWT_SALT = settings.SECRET_KEY


def create_token(payload, timeout=20):
    """
    :param payload:  例如：{'user_id':1,'username':'张三'}用户信息
    :param timeout: token的过期时间，默认20分钟
    :return:
    """
    headers = {
        'typ': 'jwt',
        'alg': 'HS256'
    }
    payload['exp'] = datetime.datetime.utcnow() + datetime.timedelta(minutes=timeout)
    result = jwt.encode(payload=payload, key=JWT_SALT, algorithm="HS256", headers=headers).decode('utf-8')
    return result


def parse_payload(token):
    """
    对token进行和发行校验并获取payload
    :param token:
    :return:
    """
    result = {'status': False, 'data': None, 'error': None}
    try:
        verified_payload = jwt.decode(token, JWT_SALT, True)
        result['status'] = True
        result['data'] = verified_payload
    except exceptions.ExpiredSignatureError:
        result['error'] = 'token已失效'
    except jwt.DecodeError:
        result['error'] = 'token认证失败'
    except jwt.InvalidTokenError:
        result['error'] = '非法的token'
    return result

http://127.0.0.1:8000/api/v1/pro/order/

Headers包含jwt
Authorization：JWT eyJ0eXAiOiJqd3QiLCJhbGciOiJIUzI1NiJ9.eyJpZCI6MSwibmFtZSI6ImtreGwiLCJleHAiOjE2NDA5MTY5ODh9.MDFc8wyDoLbpSLv6YxJNqjdsgk9COLEWQo9eAeb30pg

请求类型：get，post,puth
返回值：
{'data': '订单列表'}

版权声明：如无特殊说明，文章均为本站原创，转载请注明出处

本文链接：http://kkxl95.cn/article/1640864118/