111 lines
3.6 KiB
Python
111 lines
3.6 KiB
Python
# -*- coding: utf-8 -*-
|
||
# @Time : 2022/6/2 23:19
|
||
# @Author : 臧成龙
|
||
# @FileName: fu_auth.py
|
||
# @Software: PyCharm
|
||
# -*- coding: utf-8 -*-
|
||
|
||
import re
|
||
from datetime import datetime
|
||
|
||
# from django.core.cache import cache
|
||
from fuadmin.settings import DEMO, SECRET_KEY, WHITE_LIST
|
||
from ninja.security import HttpBearer
|
||
from system.models import MenuButton, Users
|
||
|
||
from .fu_jwt import FuJwt
|
||
from .fu_ninja import FuFilters
|
||
from .usual import get_dept, get_user_info_from_token
|
||
|
||
METHOD = {
|
||
'GET': 0,
|
||
'POST': 1,
|
||
'PUT': 2,
|
||
'DELETE': 3,
|
||
}
|
||
|
||
|
||
class GlobalAuth(HttpBearer):
|
||
def authenticate(self, request, token):
|
||
jwt = FuJwt(SECRET_KEY)
|
||
value = jwt.decode(SECRET_KEY, token)
|
||
time_now = int(datetime.now().timestamp())
|
||
# 判断token是否过期
|
||
if value.valid_to >= time_now:
|
||
token_user = value.payload
|
||
token_user_id = token_user['id']
|
||
user = Users.objects.get(id=token_user_id)
|
||
request_path = request.path
|
||
request_method = request.method
|
||
if DEMO:
|
||
# 判断是否在白名单中
|
||
if request_path in WHITE_LIST:
|
||
return token
|
||
if request_method == 'GET':
|
||
return token
|
||
else:
|
||
raise TimeoutError(403, '演示环境')
|
||
else:
|
||
# 判断是否是超级管理员
|
||
if not token_user['is_superuser']:
|
||
# 判断是path是否是‘/数字’结尾
|
||
result = re.search(r'/\d+$', request_path)
|
||
if result:
|
||
match_value = result.group()
|
||
# 将数字结尾的接口替换成.*? 因为接口中是/{id}
|
||
request_path = request_path.replace(match_value, '/*')
|
||
# 判断是否在白名单中
|
||
if request_path in WHITE_LIST:
|
||
return token
|
||
else:
|
||
menuIds = user.role.values_list('permission__id', flat=True)
|
||
queryset = MenuButton.objects.filter(id__in=menuIds, api__regex=request_path,
|
||
method=METHOD[request_method])
|
||
if queryset.exists():
|
||
return token
|
||
else:
|
||
raise TimeoutError(403, '没有权限')
|
||
# cache_token = cache.get(token_user_id)
|
||
# if token == cache_token:
|
||
return token
|
||
else:
|
||
raise TimeoutError(401, 'token时间过期')
|
||
|
||
|
||
def data_permission(request, filters: FuFilters):
|
||
user_info = get_user_info_from_token(request)
|
||
if user_info['is_superuser']:
|
||
return filters
|
||
user = Users.objects.get(id=user_info['id'])
|
||
data_range_qs = user.role.values_list('data_range', flat=True)
|
||
dept_ids = user.role.values_list('dept__id', flat=True)
|
||
|
||
# 如果有多个角色，取数据权限最大的角色
|
||
data_range = max(list(data_range_qs))
|
||
|
||
# 仅本人数据权限
|
||
if data_range == 0:
|
||
filters.creator_id = user_info['id']
|
||
|
||
# 本部门数据权限
|
||
if data_range == 1:
|
||
filters.belong_dept = user_info['dept']
|
||
|
||
# 本部门及以下数据权限
|
||
if data_range == 2:
|
||
dept_and_below_ids = get_dept(user_info['dept'])
|
||
filters.belong_dept__in = dept_and_below_ids
|
||
|
||
# 自定义数据权限
|
||
if data_range == 3:
|
||
filters.belong_dept__in = list(dept_ids)
|
||
|
||
# 所有数据权限
|
||
if data_range == 4:
|
||
pass
|
||
|
||
return filters
|
||
|
||
|
||
|