三大认证
1.认证组件：authentication
2.权限组件：permissions
3.频率组件：throttles
drf认证组件 authentication
# models.py 继承auth的AbstractUser表,增加User表额外字段
from django.db import models
from django.contrib.auth.models import AbstractUser
class User(AbstractUser):
mobile = models.CharField(max_length=11, unique=True)
class Meta:
db_table = 'od_user'
verbose_name = '用户表'
verbose_name_plural = verbose_name
def __str__(self):
return self.username
# 重设置auth的User表后需要去settings配置文件中修改auth模块的用户表指向
'''
重点：
1.auth认证6表必须在第一次数据库迁移前确定，第一次数据库迁移完成
2.完成数据库迁移，出现了auth的用户表迁移异常，需要删除的数据库迁移文件如下：
User表所在的自定义应用下的 “0001_initial.py”、admin组件下的、auth组件下的迁移文件都得删除
'''
settings.py
# 修改auth模块的用户表指向 AUTH_USER_MODEL = '应用名.表名'
AUTH_USER_MODEL = 'api.User'
源码分析
1.从APIView的dispatch(self,request, *args, **kwargs)入手
2.dispatch方法内 self.initial(request, *args, **kwargs)进入三大认证
认证组件：校验用户 --游客、合法用户、非法用户
# 游客：代表校验通过，直接进入下一步校验(权限校验)
# 合法用户：代表校验通过，并将用户存储在request.user中，再进入下一步校验(权限校验)
# 非法用户：代表校验失败，抛出异常，返回403权限异常结果
self.perform_authentication(request) # 认证组件
权限组件：校验用户权限 --必须登录、所有用户、登录之后读写，游客只读、自定义用户角色
# 认证通过：可以进入下一步校验（频率认证）
# 认证失败：抛出异常，返回403权限异常结果
self.check_permissions(request) #权限组件
频率组件：限制视图接口被访问的频率次数 --限制的条件（IP、id、唯一键）、频率周期（s、m、h）、频率的次数（3/s)
# 没有达到限次：正常访问接口
# 达到限次：限制时间内不能访问，限制时间达到后，可以重新访问
self.check_throttles(request) #频率组件
分两方面:a,b
（a）:在initial方法上面——这个是源码settings文件配置认证组件的地方
def initialize_request(self, request, *args, **kwargs):
"""
Returns the initial request object.
"""
parser_context = self.get_parser_context(request)
return Request(
request,
parsers=self.get_parsers(),
authenticators=self.get_authenticators(),
negotiator=self.get_content_negotiator(),
parser_context=parser_context
)
def get_authenticators(self):
return [auth() for auth in self.authentication_classes]
源码settings.py文件中APISettings类
DEFAULT_AUTHENTICATION_CLASSES': [
'rest_framework.authentication.SessionAuthentication', #session认证
'rest_framework.authentication.BasicAuthentication' #基础认证
]
(b): self.inital(self,request,*args,**kwargs) 包含认证，权限，频率三大组件
def initial(self, request, *args, **kwargs):
"""
Runs anything that needs to occur prior to calling the method handler.
"""
self.format_kwarg = self.get_format_suffix(**kwargs)
# Perform content negotiation and store the accepted info on the request
neg = self.perform_content_negotiation(request)
request.accepted_renderer, request.accepted_media_type = neg
# Determine the API version, if versioning is in use.
version, scheme = self.determine_version(request, *args, **kwargs)
request.version, request.versioning_scheme = version, scheme
# Ensure that the incoming request is permitted
self.perform_authentication(request) #认证
self.check_permissions(request) #权限
self.check_throttles(request) #频率
（b---1）: self.perform_authentication(request) 认证
def perform_authentication(self, request):
"""
Perform authentication on the incoming request.
Note that if you override this and simply 'pass', then authentication
will instead be performed lazily, the first time either
`request.user` or `request.auth` is accessed.
"""
request.user
（b---2）:Request类的 方法属性 user 的get方法
@property
def user(self):
"""
Returns the user associated with the current request, as authenticated
by the authentication classes provided to the request.
"""
if not hasattr(self, '_user'):
with wrap_attributeerrors():
self._authenticate()
return self._user
(b---3):self._authenticate()
认证的细则：
# 做认证
def _authenticate(self): #这里的self就是request
# 遍历拿到一个个认证器，进行认证
# self.authenticators配置的一堆认证类产生的认证类对象组成的 list
# 即：[auth() for auth in self.authentication_classes]
for authenticator in self.authenticators:
try:
# 认证器(对象)调用认证方法authenticate(认证类对象self, request请求对象)
# 返回值：登陆的用户与认证的信息组成的 tuple
# 该方法被try包裹，代表该方法会抛异常，抛异常就代表认证失败
user_auth_tuple = authenticator.authenticate(self)
except exceptions.APIException:
self._not_authenticated()
raise
# 返回值的处理
if user_auth_tuple is not None:
self._authenticator = authenticator
# 如何有返回值，就将 登陆用户 与 登陆认证 分别保存到 request.user、request.auth
self.user, self.auth = user_auth_tuple
return
# 如果返回值user_auth_tuple为空，代表认证通过，但是没有 登陆用户 与 登陆认证信息，代表游客
self._not_authenticated()
查看用户
api.urls.py
from django.conf.urls import url
from . import views
urlpatterns = [
url(r'^test/$', views.TestAPIView.as_view()),
url(r'^test/(?P<pk>d+)/$', views.TestAPIView.as_view()),
# 返回所有的用户（超级管理员才能查看）
#url(r'^users/$', views.UserListAPIView.as_view()),
#url(r'^login/$', views.LoginAPIView.as_view()),
]
views.py
from rest_framework.views import APIView
from utils.response import APIResponse
class TestAPIView(APIView):
def get(self, request, *args, **kwargs):
print(request.user) # AnonymousUser 游客
return APIResponse()
admin注册自定义表：密文操作密码，可控制在admin后台填写哪些字段，以及密码输入是密文
admin.py
from django.contrib import admin
from . import models
# admin注册自定义User表：密文操作密码
from django.contrib.auth.admin import UserAdmin as AuthUserAdmin
class UserAdmin(AuthUserAdmin):
add_fieldsets = (
(None, {
'classes': ('wide',),
# 添加用户界面可操作的字段
'fields': ('username', 'password1', 'password2', 'mobile', 'email', 'is_staff', 'is_active'),
}),
)
list_display = ('username', 'mobile', 'email', 'is_staff', 'is_active')
# 明文操作密码，admin可视化添加的用户密码都是明文，登录时用的是密文，所以用户无法登录
# admin.site.register(models.User)
admin.site.register(models.User, UserAdmin)
自定义认证类
从源码的settings文件可以看出,认证类需要继承BasicAuthentication(在authentication.py文件)
DEFAULT_AUTHENTICATION_CLASSES': [
'rest_framework.authentication.SessionAuthentication', #会重新开启CSRF认证
'rest_framework.authentication.BasicAuthentication'
]
"""
认证模块工作原理
1.继承BaseAuthentication类，重写authenticate方法
2.认证规则（authenticate方法实现体)：
没有携带认证信息，直接返回None =>游客
有认证信息，校验失败，抛异常 =>非法用户
有认证信息，校验出User对象 =>合法用户
"""
"""
权限模块工作原理
1.继承BasePermission类，重写has_permission方法
2.权限规则（has_permission方法实现体）：
返回True,代表有权限
返回False,代表无权限
"""
# 了解 --自定义认证与权限类 --实际开发，系统和第三方提供的认证与权限已经够用了，特别特殊的需求才需要自定义
"""
新建authentications.py
# 自定义认证类
"""
认证模块工作原理
1）继承BaseAuthentication类，重写authenticate方法
2）认证规则(authenticate方法实现体)：
没有携带认证信息，直接返回None => 游客
有认证信息，校验失败，抛异常 => 非法用户
有认证信息，校验出User对象 => 合法用户
"""
from rest_framework.authentication import BaseAuthentication
from rest_framework.exceptions import AuthenticationFailed
class TokenAuthentication(BaseAuthentication):
prefix = 'Token'
def authenticate(self, request):
# 拿到前台的token
auth = request.META.get('HTTP_AUTHORIZATION')
# 没有返回None，有进行校验
if not auth:
return None
auth_list = auth.split()
if not (len(auth_list) == 2 and auth_list[0].lower() == self.prefix.lower()):
raise AuthenticationFailed('非法用户')
token = auth_list[1]
# 校验算法
user = _get_obj(token)
# 校验失败抛异常，成功返回（user, token）
return (user, token)
# 校验算法（认证类）与签发算法配套
"""
拆封token：一段 二段 三段
用户名：b64decode(一段)
用户主键：b64decode(二段)
碰撞解密：md5(用户名+用户主键+服务器秘钥) == 三段
"""
import base64, json, hashlib
from django.conf import settings
from api.models import User
def _get_obj(token):
token_list = token.split('.')
if len(token_list) != 3:
raise AuthenticationFailed('token异常')
username = json.loads(base64.b64decode(token_list[0])).get('username')
pk = json.loads(base64.b64decode(token_list[1])).get('pk')
md5_dic = {
'username': username,
'pk': pk,
'key': settings.SECRET_KEY
}
if token_list[2] != hashlib.md5(json.dumps(md5_dic).encode()).hexdigest():
raise AuthenticationFailed('token内容异常')
user_obj = User.objects.get(pk=pk, username=username)
return user_obj
""" 认证类的认证核心规则
def authenticate(self, request):
token = get_token(request)
try:
user = get_user(token) # 校验算法
except:
raise AuthenticationFailed()
return (user, token)
"""
自定义权限类
新建permissions.py
# 自定义权限类
"""
权限模块工作原理
1）继承BasePermission类，重写has_permission方法
2）权限规则(has_permission方法实现体)：
返回True，代表有权限
返回False，代表无权限
"""
from rest_framework.permissions import BasePermission
class SuperUserPermission(BasePermission):
def has_permission(self, request, view):
# print(request.user)
# print(request.auth)
return request.user and request.user.is_superuser
serializers.py
from rest_framework import serializers
from rest_framework.serializers import ValidationError
from . import models
class UserModelSerializer(serializers.ModelSerializer):
class Meta:
model = models.User
fields = ('username', 'email', 'mobile')
from django.contrib.auth import authenticate
class LoginModelSerializer(serializers.ModelSerializer):
# username和password字段默认会走系统校验，而系统的post请求校验，一定当做增方式校验，所以用户名会出现 重复 的异常
# 所以自定义两个字段接收前台的账号密码
usr = serializers.CharField(write_only=True)
pwd = serializers.CharField(write_only=True)
class Meta:
model = models.User
fields = ('usr', 'pwd')
def validate(self, attrs):
usr = attrs.get('usr')
pwd = attrs.get('pwd')
try:
user_obj = authenticate(username=usr, password=pwd)
except:
raise ValidationError({'user': '提供的用户信息有误'})
# 拓展名称空间
self.user = user_obj
# 签发token
self.token = _get_token(user_obj)
# 自定义签发token
# 分析：拿user得到token，后期还需要通过token得到user
# token：用户名(base64加密).用户主键(base64加密).用户名+用户主键+服务器秘钥(md5加密)
# eg: YWJj.Ao12bd.2c953ca5144a6c0a187a264ef08e1af1
# 签发算法：b64encode(用户名).b64encode(用户主键).md5(用户名+用户主键+服务器秘钥)
# 校验算法（认证类）与签发算法配套
"""
拆封token：一段 二段 三段
用户名：b64decode(一段)
用户主键：b64decode(二段)
碰撞解密：md5(用户名+用户主键+服务器秘钥) == 三段
"""
def _get_token(obj):
import base64, json, hashlib
from django.conf import settings
t1 = base64.b64encode(json.dumps({'username': obj.username}).encode()).decode()
t2 = base64.b64encode(json.dumps({'pk': obj.id}).encode()).decode()
t3_json = json.dumps({
'username': obj.username,
'pk': obj.id,
'key': settings.SECRET_KEY
})
t3 = hashlib.md5(t3_json.encode()).hexdigest()
return '%s.%s.%s' % (t1, t2, t3)
views.py
"""
认证模块工作原理
1）继承BaseAuthentication类，重写authenticate方法
2）认证规则(authenticate方法实现体)：
没有携带认证信息，直接返回None => 游客
有认证信息，校验失败，抛异常 => 非法用户
有认证信息，校验出User对象 => 合法用户
"""
"""
权限模块工作原理
1）继承BasePermission类，重写has_permission方法
2）权限规则(has_permission方法实现体)：
返回True，代表有权限
返回False，代表无权限
"""
# 了解 - 自定义认证与权限类 - 实际开发，系统和第三方提供的认证与权限类已经够用了，特别特殊的需求才需要自定义
"""
自定义认证类、权限类
准备数据：
"""
from rest_framework.generics import ListAPIView
from . import models, serializers
# 查看所有用户信息，前提：必须是登录的超级管理员
from utils.authentications import TokenAuthentication
from utils.permissions import SuperUserPermission
class UserListAPIView(ListAPIView):
# 同电商网站，多接口是不需要登录的，少接口需要登录，使用在需要登录的接口中完成局部配置，进行局部接口校验
authentication_classes = [TokenAuthentication]
permission_classes = [SuperUserPermission]
queryset = models.User.objects.filter(is_active=True, is_superuser=False).all()
serializer_class = serializers.UserModelSerializer
def get(self, request, *args, **kwargs):
response = self.list(request, *args, **kwargs)
return APIResponse(data=response.data)
# 前后台分离登录接口
# 登录接口：如果是超级管理员登录，返回一个可以交易出超级管理员的token字符串
# 只要有用户登录，就可以返回一个与登录用户相关的token字符串 => 返回给前台 => 签发token => user_obj -> token_str
from rest_framework.generics import GenericAPIView
class LoginAPIView(APIView):
# 登录接口一定要做：局部禁用 认证 与 权限 校验
authentication_classes = []
permission_classes = []
def post(self, request, *args, **kwargs):
serializer = serializers.LoginModelSerializer(data=request.data)
# 重点：校验成功后，就可以返回信息，一定不能调用save方法，因为该post方法只完成数据库查操作
# 所以校验会得到user对象，并且在校验过程中，会完成token签发（user_obj -> token_str）
serializer.is_valid(raise_exception=True)
return APIResponse(data={
'username': serializer.user.username,
'token': serializer.token
})
可逆的加密解密
# 可逆的加密解密
import base64
s = 'abc'
# 加密
s1 = base64.b64encode(s.encode())
print(s1)
# 解密
s2 = base64.b64decode(s1).decode()
print(s2)
import hashlib
# 盐
SECRET_KEY = 'm#mh1=hyrj=wgqgz)x%@!c9^jg%@sv9iqcsrh7k-*#1u)jg0un'
# 加密
s3 = hashlib.md5(s.encode())
s3.update(SECRET_KEY.encode())
print(s3.hexdigest())
n_s = 'abc'
n_s3 = hashlib.md5(n_s.encode())
n_s3.update(SECRET_KEY.encode())
print(s3.hexdigest() == n_s3.hexdigest())