标签:fail 返回值 def ges 无限 没有 pwd url mic
1.1、content_type表
"""
Course:
name、type、days、price、vip_type
基础 免费课 7 0
中级 学位课 180 69
究极 会员课 360 至尊会员
Course:
name、type、days、content_type_id
基础 免费课 7 null
中级 学位课 180 1
究极 会员课 360 2
app01_course_1
id、price
app01_course_2
id vip_type
content_type表(Django提供)
id、app_label、model
1 app01 course_1
2 app01 course_2
"""
!
content_type表作用: 作为两张表之间的关联第三张表
1.2 admin关联自定义
通过后台在浏览器中显示对象的字段或者对密码进行加密等操作
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)
2.1全局配置
# settings.py
# drf的配置
REST_FRAMEWORK = {
# 渲染模块
'DEFAULT_RENDERER_CLASSES': ['rest_framework.renderers.JSONRenderer'],
# 异常模块
# 'EXCEPTION_HANDLER': 'rest_framework.views.exception_handler',
'EXCEPTION_HANDLER': 'utilss.exception.exception_handler',
# 全局配置 认证
'DEFAULT_AUTHENTICATION_CLASSES': [
'rest_framework.authentication.SessionAuthentication',
'rest_framework.authentication.BasicAuthentication'
],
# 全局配置 权限
'DEFAULT_PERMISSION_CLASSES': [
'rest_framework.permissions.AllowAny',
],
}
2.2局部配置
from utilss.authentications import TokenAuthentication
from utilss.permissions import SuperUserPermission
class UserListAPIView(ListAPIView):
# 局部配置
authentication_classes = [TokenAuthentication]
permission_classes = [SuperUserPermission]
3.1 urls.py
urlpatterns = [
url(r'^user/', views.UserListAPIView.as_view()),
url(r'^login/', views.LoginAPIView.as_view()),
]
3.2 view.py
from rest_framework.generics import ListAPIView # 视图工具类
from api import models, serializers
from utilss.authentications import TokenAuthentication # 自定义验证
from utilss.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.UserSerializers
def get(self, request, *args, **kwargs):
response = self.list(request, *args, **kwargs)
return APIResponse(data=response.data)
# 登录接口:如果是超级管理员登录,返回一个可以校验出超级管理员的token字符串
# 只要有用户登录,就可以返回一个与登录用户相关的token字符串 => 返回给前台 => 签发token => user_obj -> token_str
class LoginAPIView(APIView):
# 登录接口一定要做局部禁用认证与权限的校验
authentication_classes = []
permission_classes = []
def post(self, request, *args, **kwargs):
login_serial = serializers.LoginModelSerializer(data=request.data)
login_serial.is_valid(raise_exception=True)
# 重点:校验成功后,就可以返回信息,一定不能调用save方法,因为该post方法只完成数据库查操作
# 所以校验会得到user对象,并且在校验过程中,会完成token签发(user_obj -> token_str)
return APIResponse(data={
"username": login_serial.user.username,
"token": login_serial.token
})
总结:
3.3 自定义序列化类
from . import models
from rest_framework import serializers
# 用户信息查看序列化接口
class UserSerializers(serializers.ModelSerializer):
class Meta:
model = models.User
fields = ("username", "mobile", "email")
from django.contrib import auth
from rest_framework.serializers import ValidationError
# 用户登录
class LoginModelSerializer(serializers.ModelSerializer):
user = serializers.CharField(max_length=20)
pwd = serializers.CharField(max_length=30)
class Meta:
model = models.User
fields = ("user", "pwd")
def validate(self, attrs):
"""
全局钩子校验用户是否存在,存在返回token
:param attrs:
:return:
"""
user = attrs.get("user")
pwd = attrs.get("pwd")
try:
user_obj = auth.authenticate(username=user, password=pwd)
except:
raise serializers.ValidationError({"user": "提供用户信息有误"})
# 将当前用户放到名称空间中
self.user = user_obj
# 将生成token放到名称空间中
self.token = _get_token(user_obj)
return attrs
# 签发算法:b64encode(用户名).b64encode(用户主键).md5(用户名+用户主键+服务器秘钥)
# 自定义签发token
# 分析:拿user得到token,后期还需要通过token得到user
# token:用户名(base64加密).用户主键(base64加密).用户名+用户主键+服务器秘钥(md5加密)
# eg: YWJj.Ao12bd.2c953ca5144a6c0a187a264ef08e1af1
# 校验算法(认证类)与签发算法配套
"""
拆封token:一段 二段 三段
用户名:b64decode(一段)
用户主键:b64decode(二段)
碰撞解密:md5(用户名+用户主键+服务器秘钥) == 三段
"""
import base64, json, hashlib
from d_pro import settings
# 获取token
def _get_token(obj):
"""
签发算法:b64encode(用户名).b64encode(用户主键).md5(用户名+用户主键+服务器秘钥)
自定义签发token
分析:拿user得到token,后期还需要通过token得到user
token:用户名(base64加密).用户主键(base64加密).用户名+用户主键+服务器秘钥(md5加密)
eg: YWJj.Ao12bd.2c953ca5144a6c0a187a264ef08e1af1
:param obj:
:return:
"""
user_base = base64.b64encode(json.dumps({"username": obj.username}).encode()).decode()
id_base = base64.b64encode(json.dumps({"pk": obj.id}).encode()).decode()
user_md5 = hashlib.md5(
json.dumps({
"username": obj.username,
"pk": obj.id,
"key": settings.SECRET_KEY
}).encode()
).hexdigest()
print(f"{user_base}.{id_base}.{user_md5}")
return f"{user_base}.{id_base}.{user_md5}"
总结:
3.4 自定义认证类.py
# 自定义认证类
"""
认证模块工作原理
1)继承BaseAuthentication类,重写authenticate方法
2)认证规则(authenticate方法实现体):
没有携带认证信息,直接返回None => 游客
有认证信息,校验失败,抛异常 => 非法用户
有认证信息,校验出User对象 => 合法用户
"""
from rest_framework.authentication import BasicAuthentication
from rest_framework.exceptions import AuthenticationFailed
class TokenAuthentication(BasicAuthentication):
prefix = "token"
def authenticate(self, request):
# 获取头携带的token
auth = request.META.get("HTTP_AUTHORIZATION")
# 如果为空返回None,进行校验
if not auth:
return None
# 获取 token和值
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)
import base64, hashlib, json
from d_pro import settings
from api.models import User
def _get_obj(token):
"""
校验算法(认证类)与签发算法配套
拆封token:一段 二段 三段
用户名:b64decode(一段)
用户主键:b64decode(二段)
碰撞解密:md5(用户名+用户主键+服务器秘钥) == 三段
"""
token_list = token.split(".")
# 判断是否为三分
if not len(token_list):
raise AuthenticationFailed("token异常")
# 获取用户名和pk
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
}
# 判断token是否相等
if token_list[2] != hashlib.md5(json.dumps(md5_dic).encode()).hexdigest():
raise AuthenticationFailed("token内容异常")
user_obj = User.objects.filter(pk=pk, username=username).first()
return user_obj
""" 认证类的认证核心规则
def authenticate(self, request):
token = get_token(request)
try:
user = get_user(token) # 校验算法
except:
raise AuthenticationFailed()
return (user, token)
"""
总结:
3.5 自定义权限类
# 自定义权限类
"""
权限模块工作原理
1)继承BasePermission类,重写has_permission方法
2)权限规则(has_permission方法实现体):
返回True,代表有权限
返回False,代表无权限
"""
class SuperUserPermission(BasePermission):
def has_permission(self, request, view):
return request.user and request.user.is_superuser
# 自定义权限类
"""
权限模块工作原理
1)继承BasePermission类,重写has_permission方法
2)权限规则(has_permission方法实现体):
返回True,代表有权限
返回False,代表无权限
"""
总结:
4.1认证模块源码分析
# 全局配置 认证
'DEFAULT_AUTHENTICATION_CLASSES': [
'rest_framework.authentication.SessionAuthentication',
'rest_framework.authentication.BasicAuthentication'
],
# 1.全局配置,SessionAuthentication,BasicAuthentication
# SessionAuthentication内容
def authenticate(self, request):
"""
Returns a `User` if the request session currently has a logged in user.
Otherwise returns `None`.
"""
# Get the session-based user from the underlying HttpRequest object
user = getattr(request._request, 'user', None)
# Unauthenticated, CSRF validation not required
if not user or not user.is_active:
return None
self.enforce_csrf(request)
# CSRF passed with authenticated user
return (user, None)
# 2. BasicAuthentication
def authenticate(self, request):
# 获取认证头
# 1)从请求头中获取 认证字符串(auth(后台) - token(前台))
auth = get_authorization_header(request).split()
# 2)如果没有auth,返回None =>游客 在认证模块中就是返回None
if not auth or auth[0].lower() != b'basic':
return None
# 3) 如果有认证字段一定校验
# 校验失败: 抛出校验失败的异常
# 校验成功: 返回长度为2的元组,且第一位是登录user对象,第二位none 空字符传auth
if len(auth) == 1:
msg = _('Invalid basic header. No credentials provided.')
raise exceptions.AuthenticationFailed(msg)
elif len(auth) > 2:
msg = _('Invalid basic header. Credentials string should not contain spaces.')
raise exceptions.AuthenticationFailed(msg)
try:
# 反向解密获取用户id和用户密码
auth_parts = base64.b64decode(auth[1]).decode(HTTP_HEADER_ENCODING).partition(':')
except (TypeError, UnicodeDecodeError, binascii.Error):
msg = _('Invalid basic header. Credentials not correctly base64 encoded.')
raise exceptions.AuthenticationFailed(msg)
# 4)校验成功
userid, password = auth_parts[0], auth_parts[2]
# 5)返回用户
return self.authenticate_credentials(userid, password, request)
# 查询用户是否存在,返回用户对象
def authenticate_credentials(self, userid, password, request=None):
credentials = {
get_user_model().USERNAME_FIELD: userid,
'password': password
}
# 校验用户
user = authenticate(request=request, **credentials)
if user is None:
raise exceptions.AuthenticationFailed(_('Invalid username/password.'))
if not user.is_active:
raise exceptions.AuthenticationFailed(_('User inactive or deleted.'))
return (user, None)
# 认证范围 当前api接口
def authenticate_header(self, request):
return 'Basic realm="%s"' % self.www_authenticate_realm
# 3重写认证模块
class TokenAuthentication(BasicAuthentication):
prefix = "token"
def authenticate(self, request):
# 获取头携带的token
auth = request.META.get("HTTP_AUTHORIZATION")
# 如果为空返回None,进行校验
if not auth:
return None
# 获取 token和值
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)
def _get_obj(token):
"""
校验算法(认证类)与签发算法配套 拆封token:一段 二段 三段 用户名:b64decode(一段)
用户主键:b64decode(二段) 碰撞解密:md5(用户名+用户主键+服务器秘钥) == 三段
"""
token_list = token.split(".")
# 判断是否为三分
if not len(token_list):
raise AuthenticationFailed("token异常")
# 获取用户名和pk
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
}
# 判断token是否相等
if token_list[2] != hashlib.md5(json.dumps(md5_dic).encode()).hexdigest():
raise AuthenticationFailed("token内容异常")
user_obj = User.objects.filter(pk=pk, username=username).first()
return user_obj
总结:
SessionAuthentication,BasicAuthentication这两个类都是继承BaseAuthentication类
SessionAuthentication类解析
BasicAuthentication类解析
认证模块工作原理
认证模块核心实现
# 重写函数
def authenticate(self, request):
token = get_token(request)
try:
user = get_user(token) # 校验算法
except:
raise AuthenticationFailed()
return (user, token)
4.2解析模块源码分析
# 全局配置 权限
'DEFAULT_PERMISSION_CLASSES': [
'rest_framework.permissions.AllowAny',
],
#
class BasePermission(metaclass=BasePermissionMetaclass):
"""
A base class from which all permission classes should inherit.
"""
def has_permission(self, request, view):
"""
Return `True` if permission is granted, `False` otherwise.
"""
return True
def has_object_permission(self, request, view, obj):
"""
Return `True` if permission is granted, `False` otherwise.
"""
return True
# 重写权限类
class SuperUserPermission(BasePermission):
def has_permission(self, request, view):
return request.user and request.user.is_superuser
总结
5.1 自定义频率类
from rest_framework.throttling import SimpleRateThrottle
"""
1. 定义类继承SimpleRateThrottle,重写get_cache_key方法,设置scope类属性
2. scope就是一个认证字符串,在配置文件中配置scope字符串对象的频率设置
3. get_cache_key的返回值是字符串,该字符串是缓存访问次数key
"""
class ThreeTimeUserThrottle(SimpleRateThrottle):
scope = "three"
# 当前用户缓存的key(key要唯一)
def get_cache_key(self, request, view):
return "throttle:user_%s" % request.user.id
5.2时间频率时间配置
# 频率设置
'DEFAULT_THROTTLE_RATES': {
'three': '3/min',
},
5.3使用
from rest_framework.permissions import IsAdminUser
from utilss.throttling import ThreeTimeUserThrottle
class JwtUserCenterAPIView(APIView):
# 权限认证
permission_classes = [IsAdminUser]
# 频率认证
throttle_classes = [ThreeTimeUserThrottle]
def get(self, request, *args, **kwargs):
user = request.user
serializer_obj = serializers.JwtUserModelSerializer(user)
return APIResponse(data=serializer_obj.data)
总结:
1. 作用:用于一个接口在一定时间范围内可以访问几次,访问接口的次数在设定的时间范围内是否过快,配置访问频率,每次访问都要缓存几次,超次后需要等待的时间
2. 使用:定义类继承SimpleRateThrottle,重写get_cache_key方法,设置scope类属性
3. scope就是一个认证字符串,在配置文件中配置scope字符串对象的频率设置
4. get_cache_key的返回值是字符串,该字符串是缓存访问次数key
5. 在django的settings.py文件中配置频率访问次数, 'three': '3/min',在get_cache_key中使用,还有一个参数可以在重写方法中规定频率次数,rate属性也可以设置访问的频率
1.登录接口总结:
2.用户登录校验序列化类总结:
3.自定义验证模块总结:
4.自定义权限类
5认证模块源码分析
SessionAuthentication,BasicAuthentication这两个类都是继承BaseAuthentication类
SessionAuthentication类解析
BasicAuthentication类解析
认证模块工作原理
6.认证模块核心实现
# 重写函数
def authenticate(self, request):
token = get_token(request)
try:
user = get_user(token) # 校验算法
except:
raise AuthenticationFailed()
return (user, token)
7.自定义权限源码分析
8.自定义频率配置
标签:fail 返回值 def ges 无限 没有 pwd url mic
原文地址:https://www.cnblogs.com/randysun/p/12292969.html