JWT认证

JWT简介

# Json web token (JWT)
在用户注册或登录后，我们想记录用户的登录状态，或者为用户创建身份认证的凭证。我们不再使用Session认证机制，而使用Json Web Token（本质就是token）认证机制。

Json web token (JWT), 是为了在网络应用环境间传递声明而执行的一种基于JSON的开放标准（(RFC 7519).
该token被设计为紧凑且安全的，特别适用于分布式站点的单点登录（SSO）场景。JWT的声明一般被用来在身份提供者和服务提供者间传递被认证的用户身份信息，
以便于从资源服务器获取资源，也可以增加一些额外的其它业务逻辑所必须的声明信息，该token也可直接被用于认证，也可被加密。

原理

# 原理
1）jwt分三段式：头.体.签名 （head.payload.sgin）
2）头和体是可逆加密，让服务器可以反解出user对象；签名是不可逆加密，保证整个token的安全性的
3）头体签名三部分，都是采用json格式的字符串，进行加密，可逆加密一般采用base64算法，不可逆加密一般采用hash(md5)算法
4）头中的内容是基本信息：公司信息、项目组信息、token采用的加密方式信息
{
    "company": "公司信息",
    ...
}
5）体中的内容是关键信息：用户主键、用户名、签发时客户端信息(设备号、地址)、过期时间
{
    "user_id": 1,
    ...
}
6）签名中的内容时安全信息：头的加密结果 + 体的加密结果 + 服务器不对外公开的安全码 进行md5加密
{
    "head": "头的加密字符串",
    "payload": "体的加密字符串",
    "secret_key": "安全码"
}

jwt认证算法：签发与校验

签发：根据登录请求提交来的 账号 + 密码 + 设备信息 签发 token

"""
1）用基本信息存储json字典，采用base64算法加密得到 头字符串
2）用关键信息存储json字典，采用base64算法加密得到 体字符串
3）用头、体加密字符串再加安全码信息存储json字典，采用hash md5算法加密得到 签名字符串

账号密码就能根据User表得到user对象，形成的三段字符串用 . 拼接成token返回给前台
"""

校验：根据客户端带token的请求 反解出 user 对象

"""
1）将token按 . 拆分为三段字符串，第一段 头加密字符串 一般不需要做任何处理
2）第二段 体加密字符串，要反解出用户主键，通过主键从User表中就能得到登录用户，过期时间和设备信息都是安全信息，确保token没过期，且时同一设备来的
3）再用 第一段 + 第二段 + 服务器安全码 不可逆md5加密，与第三段 签名字符串 进行碰撞校验，通过后才能代表第二段校验得到的user对象就是合法的登录用户
"""

drf项目的jwt认证开发流程（重点）

"""
1）用账号密码访问登录接口，登录接口逻辑中调用 签发token 算法，得到token，返回给客户端，客户端自己存到cookies中

2）校验token的算法应该写在认证类中(在认证类中调用)，全局配置给认证组件，所有视图类请求，都会进行认证校验，所以请求带了token，就会反解出user对象，在视图类中用request.user就能访问登录的用户

注：登录接口需要做 认证 + 权限 两个局部禁用
"""

# 第三方写好的  django-rest-framework-jwt
# 安装 pip install djangorestframework-jwt

# 新建一个项目，继承AbstractUser表

# 创建超级用户

# 简单使用
# urls.py:
"""
from rest_framework_jwt.views import ObtainJSONWebToken, VerifyJSONWebToken, RefreshJSONWebToken, obtain_jwt_token

# 基类：JSONWebTokenAPIView 继承了APIView
# ObtainJSONWebToken,VerifyJSONWebToken,RefreshJSONWebToken都继承了JSONWebTokenAPIView
"""
obtain_jwt_token = ObtainJSONWebToken.as_view()
refresh_jwt_token = RefreshJSONWebToken.as_view()
verify_jwt_token = VerifyJSONWebToken.as_view()
"""

urlpatterns = [
    path('admin/', admin.site.urls),
    # path('login/', ObtainJSONWebToken.as_view()),
    path('login/', obtain_jwt_token),
]

"""  

自定义jwt认证

# auth.py
from rest_framework_jwt.authentication import jwt_decode_handler
from rest_framework import exceptions
from rest_framework_jwt.authentication import BaseJSONWebTokenAuthentication

class MyToken(BaseJSONWebTokenAuthentication):
    def authenticate(self, request):
        jwt_value = str(request.META.get('HTTP_AUTHORIZATION'))
        # 认证
        try:
            payload = jwt_decode_handler(jwt_value)
        except Exception:
            raise exceptions.AuthenticationFailed('认证失败')
        user = self.authenticate_credentials(payload)
        return user, None

    
# views.py
from django.shortcuts import render

# Create your views here.

from rest_framework.views import APIView
from rest_framework.response import Response
from rest_framework_jwt.authentication import JSONWebTokenAuthentication
from app01.auth import MyToken


class BookView(APIView):
    authentication_classes = [MyToken, ]

    def get(self, request):
        print(request.user.email)
        return Response('ok')
    
    
# urls.py
from django.contrib import admin
from django.urls import path
from app01 import views
from rest_framework_jwt.views import ObtainJSONWebToken, VerifyJSONWebToken, RefreshJSONWebToken, obtain_jwt_token

# 基类：JSONWebTokenAPIView 继承了APIView
# ObtainJSONWebToken,VerifyJSONWebToken,RefreshJSONWebToken都继承了JSONWebTokenAPIView
"""
obtain_jwt_token = ObtainJSONWebToken.as_view()
refresh_jwt_token = RefreshJSONWebToken.as_view()
verify_jwt_token = VerifyJSONWebToken.as_view()
"""

urlpatterns = [
    path('admin/', admin.site.urls),
    # path('login/', ObtainJSONWebToken.as_view()),
    path('login/', obtain_jwt_token),
    path('books/', views.BookView.as_view()),
]


# models.py
from django.db import models
from django.contrib.auth.models import AbstractUser

# Create your models here.

class User(AbstractUser):
    phone = models.CharField(max_length=64)
    icon = models.ImageField(upload_to='icon')  # ImageField依赖于pillow模块


    
# settings.py
# 一定不要忘了
AUTH_USER_MODEL = 'app01.user'

# 全局配置
# REST_FRAMEWORK = {
#     'DEFAULT_AUTHENTICATION_CLASSES': ['rest_framework_jwt.authentication.JSONWebTokenAuthentication'],
# }

JWT_AUTH = {
    'JWT_AUTH_HEADER_PREFIX': '',
}

控制用户登录后才能访问，和不登录就能访问

# 控制用户登录后才能访问，和不登录就能访问
from django.shortcuts import render

# Create your views here.
from rest_framework.views import APIView
from rest_framework.response import Response

# 使用内置jet提供的认证类，局部使用
from rest_framework_jwt.authentication import JSONWebTokenAuthentication
# 内置权限类
from rest_framework.permissions import IsAuthenticated

# 只有登录了才能访问 IsAuthenticated
# 可以通过认证类JSONWebTokenAuthentication和权限类IsAuthenticated，来控制用户登录以后才能访问某些接口
# 如果用户不登录就可以访问，只需要把权限类去掉
class OrderView(APIView):
    authentication_classes = [JSONWebTokenAuthentication, ]
    # 权限控制
    permission_classes = [IsAuthenticated, ]  # 校验用户权限有没有过

    def get(self, request, *args, **kwargs):
        return Response('这是订单信息')


# 不登录就可以访问
class UserInfoAPIView(APIView):
    authentication_classes = [JSONWebTokenAuthentication, ]
    # # 权限控制
    # permission_classes = [IsAuthenticated, ]

    def get(self, request, *args, **kwargs):
        return Response('UserInfoAPIView')

控制登录接口返回的数据格式

# 控制登录接口返回的数据格式
    -第一种方案，自己写登录接口
    -第二种写法，用内置，控制登录接口返回的数据格式
        -jwt的配置信息中
            'JWT_RESPONSE_PAYLOAD_HANDLER':
            'rest_framework_jwt.utils.jwt_response_payload_handler',
        -重写jwt_response_payload_handler，配置成自己的
       
# 第二种
# utils.py：
def my_jwt_response_payload_handler(token, user=None, request=None):  # 返回什么，前端就能看到什么样子
    return {
        'token': token,
        'msg': '成功',
        'status': 100,
        'username': user.username
    }

# settings.py:
# jwt的配置
JWT_AUTH = {
    'JWT_RESPONSE_PAYLOAD_HANDLER': 'app02.utils.my_jwt_response_payload_handler'
}

自定义基于jwt的权限类

# utils.py
def my_jwt_response_payload_handler(token, user=None, request=None):  # 返回什么，前端就能看到什么样子
    return {
        'token': token,
        'msg': '成功',
        'status': 100,
        'username': user.username
    }


from rest_framework.authentication import BaseAuthentication
from rest_framework_jwt.authentication import BaseJSONWebTokenAuthentication
from rest_framework.exceptions import AuthenticationFailed

from rest_framework_jwt.authentication import jwt_decode_handler
from rest_framework_jwt.utils import jwt_decode_handler  # 跟上面是一个
import jwt
from app01 import models


# 基于BaseAuthentication
class MyJwtAuthentication(BaseAuthentication):
    def authenticate(self, request):
        jwt_value = request.META.get('HTTP_AUTHORIZATION')
        if jwt_value:
            try:
                # jwt提供了通过三段token，取出payload的方法，并且有校验功能
                payload = jwt_decode_handler(jwt_value)
            except jwt.ExpiredSignature:
                raise AuthenticationFailed('签名过期')
            except jwt.InvalidTokenError:
                raise AuthenticationFailed('用户非法')
            except Exception as e:
                # 所有异常都会走到这里
                raise AuthenticationFailed(str(e))
            # 因为payload就是用户信息的字典
            print(payload)
            # return payload, jwt_value
            # 需要得到user对象,第一种，去数据库查
            # user = models.User.objects.filter(pk=payload.get('user_id'))
            # 第二种不查库
            user = models.User(id=payload.get('user_id'), username=payload.get('username'))
            return user, jwt_value
        # 没有值，直接抛异常
        raise AuthenticationFailed('您没有携带认证信息')


# 基于BaseJSONWebTokenAuthentication
class MyJwtJSONWebTokenAuthentication(BaseJSONWebTokenAuthentication):
    def authenticate(self, request):
        jwt_value = request.META.get('HTTP_AUTHORIZATION')
        if jwt_value:
            try:
                # jwt提供了通过三段token，取出payload的方法，并且有校验功能
                payload = jwt_decode_handler(jwt_value)
            except jwt.ExpiredSignature:
                raise AuthenticationFailed('签名过期')
            except jwt.InvalidTokenError:
                raise AuthenticationFailed('用户非法')
            except Exception as e:
                # 所有异常都会走到这里
                raise AuthenticationFailed(str(e))
            user = self.authenticate_credentials(payload)
            return user, jwt_value
        # 没有值，直接抛异常
        raise AuthenticationFailed('您没有携带认证信息')

手签发token(多方式登录)

# 使用用户名，手机号，邮箱都可以登录
# 前端需要传的数据格式
    {
        "username":"lqz/13356001233/666@qq.com",
        "password":123456
    }

# views.py:
from app02 import ser
from rest_framework.viewsets import ViewSet


# 手动签发token 完成多方式登录
# class Login2View(ViewSetMixin, APIView):
class Login2View(ViewSet):  # 跟上面完全一样
    # 这是登录接口
    # def post(self, request):  # 不写post, 直接写login
    #     pass

    def login(self, request, *args, **kwargs):
        # 1 需要有一个序列化的类
        login_ser = ser.LoginModelSerializer(data=request.data, context={'request': request})

        # 2 生成序列化类对象
        # 3 调用序列化对象的id_valid
        login_ser.is_valid(raise_exception=True)
        token = login_ser.context.get('token')
        username = login_ser.context.get('username')
        # 4 return
        return Response({'status': 100, 'msg': '登录成功', 'token': token, 'username': username})

    
# ser.py:
from rest_framework import serializers
from app01 import models
import re
from rest_framework.exceptions import ValidationError
from rest_framework_jwt.utils import jwt_encode_handler, jwt_payload_handler


class LoginModelSerializer(serializers.ModelSerializer):
    username = serializers.CharField()  # 重新覆盖username字段，数据中它是unique，post认为你保存数据，自己有校验没过
    class Meta:
        model = models.User
        fields = ['username', 'password']

    def validate(self, attrs):

        print(self.context)

        # 在这里写逻辑
        username = attrs.get('username')  # 用户名有三种方式
        password = attrs.get('password')
        # 通过判断，username数据不同，查询字段不一样
        # 正则匹配，如果是手机号
        if re.match('^1[3-9][0-9]{9}$', username):
            user = models.User.objects.filter(mobile=username).first()
        elif re.match('^[a-zA-Z0-9_-]+@[a-zA-Z0-9_-]+(.[a-zA-Z0-9_-]+)+$', username):
            user = models.User.objects.filter(email=username).first()
        else:
            user = models.User.objects.filter(username=username).first()
        if user:  # 存在用户
            # 校验密码 因为是密文 要用check_password
            if user.check_password(password):
                # 签发token
                payload = jwt_payload_handler(user)  # 把user传入，得到payload
                token = jwt_encode_handler(payload)  # 把payload传入，得到token
                self.context['token'] = token
                self.context['username'] = user.username
                return attrs
            else:
                raise ValidationError('密码错误')
        else:
            raise ValidationError('用户不存在')


"""
jwt_payload_handler = api_settings.JWT_PAYLOAD_HANDLER
jwt_encode_handler = api_settings.JWT_ENCODE_HANDLER

payload = jwt_payload_handler(user)  # 把user传入，得到payload
token = jwt_encode_handler(payload)  # 把payload传入，得到token
"""


# urls.py:
path('login2/', views.Login2View.as_view({'post': 'login'})),

配置签发过期时间

import datetime
# jwt的配置
JWT_AUTH = {
    'JWT_EXPIRATION_DELTA': datetime.timedelta(days=7),  # 过期时间，手动配置
}