114 lines
3.8 KiB
Python
114 lines
3.8 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):
|
|||
|
|
# 直接放行所有请求
|
|||
|
|
return 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):
|
|||
|
|
return filters # 直接返回不进行任何权限过滤
|
|||
|
|
# 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
|
|||
|
|
|
|||
|
|
|
|||
|
|
|