#!/usr/bin/env python # -*- coding:utf-8 -*- # project: 9月 # author: NinEveN # date: 2020/9/24 import base64 from django.http.cookie import parse_cookie from rest_framework.authentication import BaseAuthentication from rest_framework.exceptions import AuthenticationFailed from rest_framework.permissions import BasePermission from api.models import UserInfo from common.cache.storage import UserTokenCache def get_cookie_token(request): cookies = request.META.get('HTTP_COOKIE') if cookies: cookie_dict = parse_cookie(cookies) return cookie_dict.get('auth_token') def get_user_from_api_token(request): if request.method == "OPTIONS": return None api_token = request.META.get("HTTP_APIAUTHORIZATION", request.query_params.get("api_token", None)) user_obj = UserInfo.objects.filter(api_token=api_token).first() if not user_obj: raise AuthenticationFailed({"code": 1001, "error": "无效的api_token"}) if user_obj.is_active: return user_obj, api_token else: raise AuthenticationFailed({"code": 1001, "error": "用户被禁用"}) def get_user_from_request_auth(request): if request.method == "OPTIONS": return None request_token = request.META.get("HTTP_AUTHORIZATION", request.META.get("HTTP_X_TOKEN", request.query_params.get("token", get_cookie_token(request)))) if request_token: try: format_token = base64.b64decode(request_token).decode() auth_token = format_token.split(":")[0] user_name = format_token.split(":")[1] except Exception: raise AuthenticationFailed({"code": 1001, "error": "token不合法"}) if not auth_token: raise AuthenticationFailed({"code": 1001, "error": "缺少token"}) auth_cache = UserTokenCache(auth_token) userinfo = auth_cache.get_storage_cache() if not userinfo: raise AuthenticationFailed({"code": 1001, "error": "无效的token"}) if user_name != userinfo.get('username', None): raise AuthenticationFailed({"code": 1001, "error": "token校验失败"}) user_obj = UserInfo.objects.filter(uid=userinfo.get('uid', None), username=userinfo.get("username")).first() if not user_obj: raise AuthenticationFailed({"code": 1001, "error": "无效的token"}) if user_obj.is_active: auth_cache.set_storage_cache(userinfo, 3600 * 24 * 7) return user_obj, auth_token else: raise AuthenticationFailed({"code": 1001, "error": "用户被禁用"}) else: raise AuthenticationFailed({"code": 1001, "error": "无效的认证"}) class ExpiringTokenAuthentication(BaseAuthentication): def authenticate(self, request): return get_user_from_request_auth(request) class AdminTokenAuthentication(BaseAuthentication): def authenticate(self, request): result = get_user_from_request_auth(request) if result: user_obj, token = result if user_obj and user_obj.role == 3: return result else: raise AuthenticationFailed({"code": 1001, "error": "无效的认证"}) return result class ApiTokenAuthentication(BaseAuthentication): def authenticate(self, request): return get_user_from_api_token(request) class StoragePermission(BasePermission): message = "权限不足" def has_permission(self, request, view): results = get_user_from_request_auth(request) if results and results[0]: return results[0].storage_active return True class SuperSignPermission(BasePermission): message = "权限不足" def has_permission(self, request, view): results = get_user_from_request_auth(request) if results and results[0]: return results[0].supersign_active return True class DownloadQrPermission(BasePermission): message = "权限不足" def has_permission(self, request, view): domain_type = request.query_params.get('domain_type', -1) results = get_user_from_request_auth(request) if results and results[0]: if domain_type == '0' and results[0].role < 2: return False return True class UserAdInfoPermission(BasePermission): message = "权限不足" def has_permission(self, request, view): results = get_user_from_request_auth(request) if results and results[0]: if results[0].role < 2: return False return True