目录python
""" 系统:session认证 rest_framework.authentication.SessionAuthentication ajax请求经过认证: cookie中要携带 sessionid、csrftoken,请求头中要携带 x-csrftoken 第三方:jwt认证 rest_framework_jwt.authentication.JSONWebTokenAuthentication ajax请求经过认证: 请求头中要携带 authorization,值为 jwt空格token 自定义:基于jwt、其它 1)自定义认证类,继承BaseAuthentication(或其子类),重写authenticate 2)authenticate中完成 拿到认证标识 auth 反解析出用户 user 前两步操做失败 返回None => 游客 前两步操做成功 返回user,auth => 登陆用户 注:若是在某个分支抛出异常,直接定义失败 => 非法用户 """
from rest_framework.exceptions import AuthenticationFailed import jwt from rest_framework_jwt.authentication import BaseJSONWebTokenAuthentication from rest_framework_jwt.authentication import jwt_decode_handler class JWTAuthentication(BaseJSONWebTokenAuthentication): # 自定义认证类,重写authenticate方法 def authenticate(self, request): # 认证经过,返回user,auth # 认证失败,返回None auth = request.META.get('HTTP_AUTH') # 前台用auth携带token if not auth: return None try: payload = jwt_decode_handler(auth) # 出现jwt解析异常,直接抛出异常,表明非法用户,也能够返回None,做为游客处理 except jwt.ExpiredSignature: raise AuthenticationFailed('token已过时') except: raise AuthenticationFailed('token非法') user = self.authenticate_credentials(payload) return (user, auth)
from rest_framework.authentication import BaseAuthentication def authenticate(self, request): auth = 从request中获得 user = 从auth中获得 if not user: return None return user, auth
""" 系统: 1)AllowAny:容许全部用户,校验方法直接返回True 2)IsAuthenticated:只容许登陆用户 必须request.user和request.user.is_authenticated都经过 3)IsAuthenticatedOrReadOnly:游客只读,登陆用户无限制 get、option、head 请求无限制 前台请求必须校验 request.user和request.user.is_authenticated 4)IsAdminUser:是不是后台用户 校验 request.user和request.user.is_staff is_staff(能够登陆后台管理系统的用户) 自定义:基于auth的Group与Permission表 1)自定义权限类,继承BasePermission,重写has_permission 2)has_permission中完成 拿到登陆用户 user <= request.user 校验user的分组或是权限 前两步操做失败 返回False => 无权限 前两步操做成功 返回True => 有权限 """
from rest_framework.permissions import BasePermission class AdminPermission(BasePermission): # 继承BasePermission,重写has_permission def has_permission(self, request, view): # 有权限,返回True # 无权限,返回False user = request.user if not user: return False # 用户是 管理员 分组 (管理员分组是Group表中的一条自定义记录) if not user.groups.filter(name='管理员'): return False # 登陆的用户必须是自定义管理员分组成员 return True
""" 系统: 1)AnonRateThrottle:对同一IP游客的限制 2)UserRateThrottle:对同一IP登陆用户的限制 必须在settings.py中 'DEFAULT_THROTTLE_RATES': { 'user': '10/min', # 登陆的用户一分钟能够访问10次 'anon': '3/min', # 游客一分钟能够访问3次 } 在视图类中: class TempAPIView(APIView): ... throttle_classes = [AnonRateThrottle, UserRateThrottle] 自定义:基于auth的Group与Permission表 1)自定义频率类,继承SimpleRateThrottle,重写get_cache_key,明确scope SimpleRateThrottle已经帮咱们实现了 allow_request、wait 2)scope与settings.py的DEFAULT_THROTTLE_RATES配合使用 3)get_cache_key中完成 拿到限制信息 ident <= request中获取 没有限制信息 返回None => 不限制 有限制信息 返回限制信息字符串 => 有限制 """
from rest_framework.throttling import SimpleRateThrottle class ThreeMinRateThrottle(SimpleRateThrottle): scope = 'sms' def get_cache_key(self, request, view): # 对手机号频率限制 ident = request.data.get('mobile') if not ident: # 为发现限制条件,返回None表明不进行频率限制 return None return self.cache_format % { 'scope': self.scope, 'ident': ident } # settings.py 配置文件 'DEFAULT_THROTTLE_RATES': { 'user': '10/min', # 登陆的用户一分钟能够访问10次 'anon': '3/min', # 游客一分钟能够访问3次 'sms': '1/min', }
from rest_framework.views import APIView from django.contrib import auth # views.py # 登陆成功后产生cookie,cookie中携带 sessionid、csrftoken class LoginSessionAPIView(APIView): # 登陆要禁用认证与权限 authentication_classes = [] permission_classes = [] def post(self, request, *args, **kwargs): username = request.data.get('username') password = request.data.get('password') if not (username and password): return Response('信息有误') # user = models.User.objects.filter(username=username).first() # type: models.User # user.check_password(password) user = auth.authenticate(request, username=username, password=password) if not user: return Response('登陆失败') auth.login(request, user=user) return Response('登陆成功') # postman 请求头中要携带 x-csrftoken from rest_framework.permissions import IsAuthenticatedOrReadOnly from rest_framework.authentication import SessionAuthentication class CarModelViewSet(ModelViewSet): queryset = models.Car.objects.filter(is_delete=False) serializer_class = serializers.CarModelSerializer permission_classes = [IsAuthenticatedOrReadOnly] authentication_classes = [SessionAuthentication] def create(self, request, *args, **kwargs): return super().create(request, *args, **kwargs) def destroy(self, request, *args, **kwargs): obj = self.get_object() obj.is_delete = True obj.save() return Response('删除成功') # urls.py # 游客只能够查看,登陆后能够增删改 url(r'^cars/$', views.CarModelViewSet.as_view({ 'get': 'list', 'post': 'create' }))
import re from utils.response import APIResponse from rest_framework_jwt.serializers import jwt_encode_handler, jwt_payload_handler class LoginJWTAPIView(APIView): authentication_classes = () permission_classes = () def post(self, request, *args, **kwargs): # username可能携带的不止是用户名,可能仍是用户的其它惟一标识 手机号 邮箱 username = request.data.get('username') password = request.data.get('password') # 若是username匹配上手机号正则 => 多是手机登陆 if re.match(r'1[3-9][0-9]{9}', username): try: # 手动经过 user 签发 jwt-token user = models.User.objects.get(mobile=username) except: return APIResponse(1, '该手机未注册') # 邮箱登陆 等 # 帐号登陆 else: try: # 手动经过 user 签发 jwt-token user = models.User.objects.get(username=username) except: return APIResponse(1, '该帐号未注册') # 得到用户后,校验密码并签发token if not user.check_password(password): return APIResponse(1, '密码错误') payload = jwt_payload_handler(user) token = jwt_encode_handler(payload) return APIResponse(0, 'ok', results={ 'username': user.username, 'mobile': user.mobile, 'token': token }) # 自定义认证类 JWTAuthentication from rest_framework.permissions import IsAuthenticatedOrReadOnly from api.authentications import JWTAuthentication class CarV3ModelViewSet(ModelViewSet): authentication_classes = [JWTAuthentication] permission_classes = [IsAuthenticatedOrReadOnly] queryset = models.Car.objects.filter(is_delete=False) serializer_class = serializers.CarModelSerializer def destroy(self, request, *args, **kwargs): obj = self.get_object() obj.is_delete = True obj.save() return Response('删除成功') # urls.py # 游客只能够查看,登陆后能够增删改 url(r'^v3/cars/$', views.CarV3ModelViewSet.as_view({ 'get': 'list', 'post': 'create' }))