JWT

Json web token (JWT)

jwt的原理

jwt是一种前后端的登陆认证方式,它分token的签发和认证. 签发的意思是用户登陆成功,生成三段式的token串.
认证指的是用户访问某个接口, 需要携带token串过来, 我们完成认证.
可以再具体说说三段式是什么; 头.荷载.签名 每一段都通过base64编码!!
认证怎么认证?通过头和荷载两部分再用同样的加密方式(eg:md5加盐)得到签名,比较两个签名是否一样.

jwt的构成

JWT的构成: 三段式!
JWT就是一段字符串,由三段信息构成的,将这三段信息文本用.链接一起就构成了Jwt字符串.
1> 头部 header - 一般放公司信息,加密方式(注意,像什么加密的盐肯定不会放进去的)
2> 载荷 payload - 用户信息,eg: 'user_id'、'expire过期时间'、user_name、email
3> 签证 signature - 第一部分和第二部分加密得到的

eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.
eyJzdWIiOiIxMjM0NTY3ODkwIiwibmFtZSI6IkpvaG4gRG9lIiwiYWRtaW4iOnRydWV9.
TJVA95OrM7E2cBab30RMHrHDcEfxjoYZgeFONFh7HgQ

▼ header
完整的头部就像下面这样的JSON：
{
  'typ': 'JWT',  # -- 声明类型,这里是jwt
  'alg': 'HS256' # -- 声明加密的算法 通常直接使用 HMAC SHA256
}


▼ payload
载荷就是存放有效信息的地方.这个名字像是特指飞机上承载的货品.
{
  "sub": "1234567890",
  "name": "John Doe",
  "admin": true
}


▼ signature
base64转码后的header和payload通过加密+secret加盐处理得到
// Javascript
var encodedString = base64UrlEncode(header) + '.' + base64UrlEncode(payload);
var signature = HMACSHA256(encodedString, 'secret'); // TJVA95OrM7E2cBab30RMHrHDcEfxjoYZgeFONFh7HgQ

★ 注意:secret是保存在服务器端的,jwt的签发生成也是在服务器端的,secret就是用来进行jwt的签发和jwt的验证
所以,它就是你服务端的私钥,在任何场景都不应该流露出去.
     一旦客户端得知这个secret,那就意味着客户端是可以自我签发jwt了!!

base64编解码

base64的编码长度都是4的倍数,如果不是4的倍数,需要用 = 来填充, 最多填充三个.

import base64
import json

dic_info = {
    "sub": "1234567890",
    "name": "lqz",
    "admin": True
}
byte_info = json.dumps(dic_info).encode('utf-8')
# -- base64编码
base64_str = base64.b64encode(byte_info)
print(base64_str)
# -- base64解码
base64_str = 'eyJzdWIiOiAiMTIzNDU2Nzg5MCIsICJuYW1lIjogImxxeiIsICJhZG1pbiI6IHRydWV9'
str_url = base64.b64decode(base64_str).decode("utf-8")
print(str_url)

---

drf-jwt

使用第三方模块: django-rest-framework-jwt (作者好多年都没改了,但不影响使用.)
有的公司会用新的, django-rest-framework-simplejwt (换汤不换药..其实差不多 但这个作者一直在维护)

快速签发和认证

签发

注意哦,发送的json字符串的用户名必须是 username, 密码必须是 password.

urls.py

from django.contrib import admin
from django.urls import path
from rest_framework_jwt.views import obtain_jwt_token
from app01 import views

urlpatterns = [
    path('admin/', admin.site.urls),
    # -- 登陆功能不用写,drf-jwt帮忙写好了..
    #    有了登陆功能后,就可以《签发》token啦,但该功能依赖于auth的user表!!
    #    So,为了实验,我们得迁移数据库并创建一个超级用户!!
    path('login/', obtain_jwt_token),
    path('index/', views.IndexView.as_view())
]

指定签发返回格式

step1: 在app01下创建文件 utils.py

utils.py

# -- 函数名怎么命名随便,但参数必须得是这三个!!源码规定了.
def jwt_response_payload_handler(token, user=None, request=None):
    # -- return什么,前端就显示什么.
    return {
        'status': 0,
        'msg': 'ok',
        'data': {
            'token': token,
            'username': user.username
        }
    }

step2: 在项目的settings.py文件中进行配置

# -- 对drf_jwt进行配置
JWT_AUTH = {
    # -- 如果不自定义,返回的格式是固定的(只有token字段)
    'JWT_RESPONSE_PAYLOAD_HANDLER': 'app01.utils.jwt_response_payload_handler',
    # 可以指定过期时间,比如1天
    #'JWT_EXPIRATION_DELTA': datetime.timedelta(days=1),
}

为啥这样配置就ok呢?

- 先看下drf_jwt的默认settings.py配置文件
  DEFAULTS = {
      # ... ...
      'JWT_PAYLOAD_GET_USER_ID_HANDLER':
      'rest_framework_jwt.utils.jwt_get_user_id_from_payload_handler',
      # ... ...
  }


- 查看rest_framework_jwt.utils文件下的jwt_get_user_id_from_payload_handler方法,源码如下:
  def jwt_response_payload_handler(token, user=None, request=None):
    """
    Example:

    def jwt_response_payload_handler(token, user=None, request=None):
        return {
            'token': token,
            # -- 提供了一种方式,可以返回序列化的结果
            'user': UserSerializer(user, context={'request': request}).data
        }
    """
    return {
        'token': token
    }

认证

必须在请求头里带上键值对 'Authorization':'jwt token值'

views.py

from rest_framework.views import APIView
from rest_framework.response import Response
# -- drf_jwt内置的认证类
from rest_framework_jwt.authentication import JSONWebTokenAuthentication
# -- drf内置的权限类
from rest_framework.permissions import IsAuthenticated


class IndexView(APIView):
    # -- 要想使用drf_jwt内置的《认证》,必须得嵌套一个权限一起使用.. 作者就是这样规定的,no why.
    authentication_classes = [JSONWebTokenAuthentication]
    permission_classes = [IsAuthenticated]

    def get(self, request):
        # print(request.user)  # -- 当前登陆用户
        return Response('ok')

签发的源码分析

path('login/', obtain_jwt_token), -- ObtainJSONWebToken.as_view()
ObtainJSONWebToken视图类的源码如下:
class ObtainJSONWebToken(ObtainJSONWebToken):
    # -- 该视图类就只配置了一个序列化类!
    serializer_class = JSONWebTokenSerializer
    

继续分析,登陆请求的本质是携带用户名和密码向后端发了一个post请求.So,我们需要找到视图类中的post方法!!
ObtainJSONWebToken视图类里没有,就去它继承的父类ObtainJSONWebToken里找!
源码如下:
class JSONWebTokenAPIView(APIView):
    # -- 赋值为空元组,表明登陆接口,不走认证也不走权限.
    permission_classes = ()
    authentication_classes = ()

    def get_serializer_context(self):
        return {
            'request': self.request,
            'view': self,
        }

    # -- 像GenericAPIView一样重写了get_serializer_class和get_serializer方法!代码几乎一摸一样.
    def get_serializer_class(self):
        assert self.serializer_class is not None, (
            "'%s' should either include a `serializer_class` attribute, "
            "or override the `get_serializer_class()` method."
            % self.__class__.__name__)
        return self.serializer_class

    def get_serializer(self, *args, **kwargs):
        serializer_class = self.get_serializer_class()
        kwargs['context'] = self.get_serializer_context()
        return serializer_class(*args, **kwargs)

    def post(self, request, *args, **kwargs):
        # -- request.data是前端传过来的包含用户名密码的json数据;实例化得到序列化类的一个实例
        # -- 序列化类就是配置在ObtainJSONWebToken视图类中的 `serializer_class = JSONWebTokenSerializer`
        serializer = self.get_serializer(data=request.data)

        # -- 执行序列化类实例的is_valid方法,会执行字段自己的校验规则、局部钩子、全局钩子
        #    查看序列化类的源码,得知校验成功,serializer.is_valid()返回了一个{'token':xxx,'user':user}
        if serializer.is_valid():
            # -- 取出user、取出token
            user = serializer.object.get('user') or request.user
            token = serializer.object.get('token')
            # -- 执行我们在项目配置文件中配置的函数!(指定签发返回格式的配置!!!) - 决定了返回前端什么格式的数据.
            #    command+b 点进去看. 项目配置文件没有,就会去jwt的默认配置文件中找.
            response_data = jwt_response_payload_handler(token, user, request)
            # -- 我们编写的指定签发返回格式的那个函数,return的是一个字典,这里返回给前端!
            response = Response(response_data)
            # ... ...
            return response

        return Response(serializer.errors, status=status.HTTP_400_BAD_REQUEST)
      
      

读一读序列化类JSONWebTokenSerializer的源码:
class JSONWebTokenSerializer(Serializer):
    def __init__(self, *args, **kwargs):
        super(JSONWebTokenSerializer, self).__init__(*args, **kwargs)
        
        # -- 下方两行代码就相当于如下的伪代码,在序列化类里定义了两个字段
        #    username = serializers.CharField(max_length=32)
        #    password = serializers.CharField(max_length=32)
        self.fields[self.username_field] = serializers.CharField()
        self.fields['password'] = PasswordField(write_only=True)

    @property
    def username_field(self):
        return get_username_field()

    # -- 全局钩子
    #    简单来说,取出前端传过来的数据,然后认证用户,然后得到用户、荷载、token,最后返回
    def validate(self, attrs):
        # -- 定义了一个字典,就相当于{'username':'lqz','password':123456}
        credentials = {
            self.username_field: attrs.get(self.username_field),
            'password': attrs.get('password')
        }

        if all(credentials.values()):
            # -- authenticate方法是Django里的auth组件提供的用户认证功能,即验证用户名以及密码是否正确
            user = authenticate(**credentials)  # -- 通过用户名密码认证auth的user表中的用户

            if user:
                if not user.is_active:  # -- 用户不活跃抛异常,会被全局捕获到
                    msg = _('User account is disabled.')  # _()是国际化
                    raise serializers.ValidationError(msg)

                # -- ☆通过user得到荷载 jwt_payload_handler具体怎么写的不用看
                payload = jwt_payload_handler(user)

                # -- 正常来将,全局钩子应该返回attrs,但这里不这样做.Hhh.
                #    这样的话直接从 `序列化类的实例.data` ser.data中就可以取到token和user.
                return {
                    # -- ☆通过荷载得到三段式的Token串 jwt_encode_handler具体怎么写的不用看
                    'token': jwt_encode_handler(payload),
                    'user': user
                }
            else:
                # ... ...
        else:
            # ... ...

认证的源码分析

authentication_classes = [JSONWebTokenAuthentication]
关键在于读认证类JSONWebTokenAuthentication的authenticate方法!!
JSONWebTokenAuthentication没有该方法,就去其父类BaseJSONWebTokenAuthentication中找!!
BaseJSONWebTokenAuthentication类的authenticate方法源码如下:
    def authenticate(self, request):
        # -- 取出前端请求头中的键Authorization对应的value值!
        #    查看JSONWebTokenAuthentication类的get_jwt_value的源码!得知jwt_value就是前端在请求头中传入的token串!!
        jwt_value = self.get_jwt_value(request)
        # -- 若jwt_value为None,证明前端没有传认证的token串.
        #    但这里写的很奇怪,没传就返回一个None,并没有抛异常,也认证通过..
        #    所以作者多加了个权限类来限制.╮(￣▽￣"")╭
        if jwt_value is None:
            return None

        try:
            # -- 通过token传获的payload!!会进行验签(检查是否被篡改)、检查过期时间.
            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()

        # -- 通过payload获得当前用户: 通过payload中的user_id去auth的user表中获取当前用户!
        user = self.authenticate_credentials(payload)

        return (user, jwt_value)  # -- 返回user和token串
      
      

查看JSONWebTokenAuthentication类的get_jwt_value的源码如下:
   def get_jwt_value(self, request):
        """
        def get_authorization_header(request):
            auth = request.META.get('HTTP_AUTHORIZATION', b'')
            if isinstance(auth, str):
                auth = auth.encode(HTTP_HEADER_ENCODING)
            return auth
        """
        # -- 取出前端传入的token串,通过空格切分!!
        auth = get_authorization_header(request).split()
        auth_header_prefix = api_settings.JWT_AUTH_HEADER_PREFIX.lower()

        if not auth:  # -- 若没有传认证的token,返回None
            if api_settings.JWT_AUTH_COOKIE:
                return request.COOKIES.get(api_settings.JWT_AUTH_COOKIE)
            return None

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

        if len(auth) == 1:  # -- 若auth元组的长度等于1,抛异常,因为我们是`jwt token串`这样传的,切分后元组长度必定为2!
            msg = _('Invalid Authorization header. No credentials provided.')
            raise exceptions.AuthenticationFailed(msg)
        elif len(auth) > 2:  # -- 若auth元组的长度大于2,抛异常.
            msg = _('Invalid Authorization header. Credentials string '
                    'should not contain spaces.')
            raise exceptions.AuthenticationFailed(msg)

        return auth[1]  # -- 返回token串!
      
      
      
# -- 认证类:token串传不传都给认证通过
authentication_classes = [JSONWebTokenAuthentication]
# -- 加了权限类进行限制,限制必须携带token串!!
permission_classes = [IsAuthenticated]
权限类的源码如下:
class IsAuthenticated(BasePermission):
    def has_permission(self, request, view):
        return bool(request.user and request.user.is_authenticated)

自定义用户表签发token

models.py

from django.db import models


class UserInfo(models.Model):
    username = models.CharField(max_length=32)
    password = models.CharField(max_length=32)
    
    
# -- 迁移数据库,并添加一条数据 egon 123456

views.py

from rest_framework.views import APIView
from rest_framework.response import Response
from rest_framework_jwt.settings import api_settings
from .models import UserInfo


# -- (⁎⁍̴̛ᴗ⁍̴̛⁎) 缺啥就去源码里拷贝啥就行..记,不可能记的.Hhh
jwt_payload_handler = api_settings.JWT_PAYLOAD_HANDLER
jwt_encode_handler = api_settings.JWT_ENCODE_HANDLER


# -- 使用自己的用户表UserInfo,签发Token
class LoginView(APIView):
    def post(self, request):
        username = request.data.get('username')
        password = request.data.get('password')
        # -- 注意哦!!自己弄的用户表,认证就不能用auth组件的认证啦,这是不同的两套东西!!别整混了!
        user = UserInfo.objects.filter(username=username, password=password).first()
        if user:  # -- 用户名和密码正确,登陆成功,签发token
            payload = jwt_payload_handler(user)  # -- 通过用户得到荷载,源码里的
            token = jwt_encode_handler(payload)  # -- 通过荷载得到token,源码里的
            return Response({
                'code': '100',
                'msg': '登陆成功',
                'token': token,
                'username': user.username
            })
        else:
            return Response({'code': '101', 'msg': '用户名或密码错误.'})
          


# -- Ps:记得配置路由
path('my_login/', views.LoginView.as_view())

自定义用户表的认证类

auth.py

from rest_framework.authentication import BaseAuthentication
from rest_framework import exceptions
from rest_framework_jwt.settings import api_settings
from .models import UserInfo
import jwt

jwt_decode_handler = api_settings.JWT_DECODE_HANDLER


# -- 我们自定义的认证类必须携带token,不然抛出异常!
class JwtAuthentication(BaseAuthentication):
    def authenticate(self, request):
        # -- 取出前端传入的token串
        #    可按照我们自己的规定来,不一定就必须是jwt token串,键也不一定叫做Authorization
        #    这里我们规定键名叫做token,取的时候记得加上HTTP_
        token = request.META.get('HTTP_TOKEN')
        # -- 通过token获得payload
        try:
            # -- 通过token传获的payload!!会进行验签(检查是否被篡改)、检查过期时间.
            payload = jwt_decode_handler(token)
        except jwt.ExpiredSignature:  # -- 签名过期抛异常
            msg = '签名过期'
            raise exceptions.AuthenticationFailed(msg)
        except jwt.DecodeError:  # -- 签名解析出错抛异常
            msg = '解码错误'
            raise exceptions.AuthenticationFailed(msg)
        except jwt.InvalidTokenError:  # -- 不合法的签名抛异常.
            raise exceptions.AuthenticationFailed()
        # -- 不想做这么细粒化的报错,可直接一个Exception搞定!
        # except Exception:
        #     raise exceptions.AuthenticationFailed("token认证失败!")
        # -- 通过payload获得当前用户(自己的表,自己拿,所以源码里的那个用不到,源码里是从auth_user表里取的)
        #    思考一个问题,若每次认证都要查询数据库拿user对象,是不是还可以优化??
        # user = UserInfo.objects.filter(pk=payload['user_id']).first()  # -- 这是可以的!
        # -- 稍微优化以下,不是每次都去查询当前登陆用户,
        #    注意:优化方案1、2在视图类中就只能拿到当前登陆用户的id和用户名,其它的拿不到!!
        #    优化方案1:
        # user = {'id': payload['user_id'], 'username': payload['username']}
        #    优化方案2:
        user = UserInfo(id=payload['user_id'], username=payload['username'])

        return user, token
      
      
      
"""
举一反三,跟上面的代码是一样的效果.
但这个认证类在视图类中需要配合权限类一起使用!!permission_classes = [IsAuthenticated]
"""
from rest_framework_jwt.authentication import BaseJSONWebTokenAuthentication
class MyAuth(BaseJSONWebTokenAuthentication):
    def get_jwt_value(self, request):
        return request.META.get('HTTP_TOKEN')

    def authenticate_credentials(self, payload):
        from .models import UserInfo
        return UserInfo.objects.filter(pk=payload['user_id']).first()

views.py

from rest_framework.views import APIView
from rest_framework.response import Response
from .auth import JwtAuthentication


# -- 使用自己的认证类
class MyIndexView(APIView):
    authentication_classes = [JwtAuthentication]

    def get(self, request):
        # print(request.user)  # -- 查询数据库得到user对象
        # print(request.user['id'])  # -- 返回一个user字典
        print(request.user.id)
        return Response('ok')

---

rbac

RBAC 基于角色的访问控制. 一般用在公司内部的管理系统中.

(python的django写公司内部项目较多,rbac很重要!)
Django的Auth组件采用的认证规则就是RBAC!!

★ rbac表分析
- 用户表         auth_user
- 角色表/部门表   auth_group
- 权限表         auth_permission

- 角色:权限 = n:n ---> 中间表      auth_group_permissions
  eg:市场部的角色既可以报销工资又可以申请出差等,申请出差和报销工资的权限其它部门也可以有.
- 用户:角色 = n:n ---> 中间表      auth_user_group
  eg:某个用户和老板都是既属于市场部又属于财务部.
- django的auth组件多了一个表 用户:权限 = n:n ---> 中间表      auth_user_user_permissions
  eg:老板想亲自发工资,不是将老板加入财务部,而是单单只给老板发工资的权限,财务部的其它权限老板是没有的!!

Django有个admin后台管理,配合auth,可以快速搭建出一个基于rbac的后台管理系统!!

from django.contrib import admin
from .models import UserInfo

admin.site.register(UserInfo)

"""
几年前,很多公司的内部管理系统,就是使用django的admin快速搭建的.
但admin不好看,可以使用第三方进行美化
- django1.x上很火的Xadmin: 前端基于jq+bootstrap; 2.x上支持不好, 3.x直接不能用了.. 而且作者也弃坑了!!
- simple-ui 完美的支持了django3.x
  pip3 install django-simpleui
  安装好后,进行应用注册,注册到第一的位置!
  
可以集成echars进行图表的展示.
"""

---

drf大回顾

1. drf入门规范
   - 前后端开发模式
   - API接口
   - postman的使用
         - 做接口测试
   - restful规范:10条
   - djangorestframework
   - APIView的执行流程
         - 包装新的request;执行三大认证;处理全局异常
   - Request对象的源码
         - 重写了__getattr__;data;query_params
2. 序列化
   - 序列化和反序列化
         - 序列化: instance, many=True, ser.data
         - 反序列化: data, ser.is_valid --> ser.save (update,create)
   - Serializer
         - 写一个个字段
         - 字段类型 CharField 、IntegerField 、DateField
         - 字段属性 read_only 、write_only
         - (跟表模型没有直接联系),需重写update,create方法,一定要返回当前对象 
         - 局部钩子、全局钩子
   - ModelSerializer
         - 重写字段
         - SerializerMethodField+配套一个`get_字段名`函数名
         - class Meta:
               表模型
               序列化和反序列化的字段
         - 局部钩子、全局钩子
         - 可以重写update和create
3. 请求和响应
   - Request类
   - 解析的编码: 局部和全局配置
   - Response类: data、status、header
   - 响应的格式: 局部和全局配置
4. 视图
   - 两个视图基类
         - APIView
               - 认证类
               - 权限类
         - GenericAPIView
               - 两个类属性
               - 三个方法: 获取单个、获取所有、获取序列化的类
   - 5个视图扩展类
         - 5个接口: list、retrieve、destory、update、create
   - 9个视图子类
         - 写两个类属性
   - 视图集
         - ViewSetMixin: 重写了as_view,路由配置变了,自动生成路由
         - ViewSet: ViewSetMixin + APIView
         - GenericViewSet: ViewSetMixin + GenericAPIView
         - ModelViewSet
         - ReadOnlyModelViewSet
5. 路由
   - 自动生成路由: 必须得继承自ViewSetMixin的视图类
   - action装饰器
         - 在视图类对象中存在: self.action
6. 认证、权限、频率
   - 源码: 为什么认证类配置到视图类中就会执行.
   - 写一个类,继承基类,重写某个方法,全局配置和局部配置
7. 过滤,排序(查询所有)
   - 继承了GenericAPIView+ListModelMixin
   - 在视图类属性中配置 filter_backends = [内置/第三方/自己写的]
   - 自定义过滤类,继承BaseFilterBackend,重写filter_queryset
8. 分页
   - 三种分页方式
   - 配置在继承了GenericAPIView+ListModelMixin的视图类的pagination_class类属性
   - 继承APIView,需要自己写
9. 全局异常处理
   - 写一个函数
   - 配置文件配置
10. 自动生成接口文档
   - 接口文档有规范
   - yapi:如何使用
11. jwt
12. simple-ui的使用
13. rbac
14. Book系列多表群操作 (这种需求很少) 单增、群增
    参考链接:https://www.liuqingzheng.top/python/Django-rest-framework框架/11-Book系列多表群操作/

---