usermod/Utils/AuthUtils.py

119 lines
3.8 KiB
Python
Raw Normal View History

2023-02-10 13:54:44 +08:00
from typing import List
2023-02-21 13:56:47 +08:00
from jose import jwt
2023-02-07 08:48:41 +08:00
from datetime import datetime, timedelta
2023-02-21 13:56:47 +08:00
from fastapi import Header, HTTPException
2023-02-07 16:09:57 +08:00
from Schemas.UserSchemas import TokenData
2023-02-07 08:48:41 +08:00
2023-02-14 14:06:25 +08:00
SECRET_KEY = "MADASDZXC255f"
ALGORITHM = "HS256"
2023-02-07 08:48:41 +08:00
2023-02-14 14:06:25 +08:00
def create_token(data: dict, secret_key=SECRET_KEY, algorithm=ALGORITHM, expires_delta: timedelta = timedelta(days=3)):
2023-02-08 15:40:03 +08:00
# 设置加密数据
to_encode_body = dict()
to_encode_body.update(data.copy())
# 设置过期时间
2023-02-10 09:35:56 +08:00
if expires_delta:
2023-02-10 09:35:05 +08:00
expire = datetime.utcnow() + expires_delta
to_encode_body.update({"exp": expire})
2023-02-08 15:40:03 +08:00
# Token编码
encoded_jwt = jwt.encode(to_encode_body, secret_key, algorithm=algorithm)
return encoded_jwt
2023-02-10 13:54:44 +08:00
def decode_token(token: str, secret_key: str, algorithms: List[str] = ['HS256']):
payload = jwt.decode(token, secret_key, algorithms=algorithms)
2023-02-08 15:40:03 +08:00
return payload
2023-02-07 08:48:41 +08:00
class Token:
2023-02-14 14:06:25 +08:00
SECRET_KEY = SECRET_KEY
ALGORITHM = ALGORITHM
2023-02-07 08:48:41 +08:00
@classmethod
def create_token(cls, data: dict, expires_delta: timedelta = timedelta(days=3)):
2023-02-08 15:40:03 +08:00
return create_token(data, cls.SECRET_KEY, cls.ALGORITHM, expires_delta)
2023-02-07 08:48:41 +08:00
@classmethod
def decode_token(cls, token: str):
2023-02-08 15:40:03 +08:00
payload = decode_token(token, cls.SECRET_KEY, algorithms=[cls.ALGORITHM])
2023-02-07 08:48:41 +08:00
return payload
2023-02-21 13:56:47 +08:00
def token_data_depend(Authorization: str = Header(None)) -> TokenData:
2023-02-07 08:48:41 +08:00
try:
2023-02-10 13:54:44 +08:00
_, jwt_token = Authorization.split(" ")
2023-02-07 08:48:41 +08:00
payload = Token.decode_token(jwt_token)
2023-02-14 14:06:25 +08:00
print(payload)
2023-02-07 16:09:57 +08:00
token_data = TokenData(**payload)
return token_data
2023-02-07 08:48:41 +08:00
except Exception as e:
print(e)
2023-02-07 16:09:57 +08:00
raise HTTPException(status_code=403, detail="无权限的操作")
2023-02-08 08:48:22 +08:00
2023-02-07 16:09:57 +08:00
# 管理员token验证
2023-02-10 09:15:17 +08:00
def admin_auth_token_depend(Authorization: str = Header(None)) -> TokenData:
2023-02-10 13:54:44 +08:00
try:
_, token = Authorization.split(" ")
token_data: TokenData = token_data_depend(Authorization)
except Exception as e:
print(e)
raise HTTPException(status_code=403, detail="非管理员,无权限的操作")
2023-02-08 14:57:34 +08:00
if token_data.role != 'admin':
2023-02-10 16:39:06 +08:00
print(token_data)
2023-02-08 14:57:34 +08:00
raise HTTPException(status_code=403, detail="非管理员,无权限的操作")
return token_data
# 应用服务对用户模块的请求验证
2023-02-21 13:56:47 +08:00
# 接口需要的权限列表
def get_auth_rule_check_depend(contains_rules=[], requisite_rules=[]):
"""
接口需要的权限列表验证
:param requisite_rules: 必须的权限规则id
:param contains_rules: 权限只包含其中一个即可授权的权限列表
:return:
"""
def auth_rule_check_depend(Authorization: str = Header(None)):
try:
_, jwt_token = Authorization.split(" ")
payload = Token.decode_token(jwt_token)
print(payload)
token_data = TokenData(**payload)
if not check_auth(token_data):
raise HTTPException(status_code=403, detail="无权限的操作")
return token_data
except Exception as e:
print(e)
raise HTTPException(status_code=403, detail="无权限的操作")
return auth_rule_check_depend
def check_auth(auth_data, contains_rules, requisite_rules):
auth_rule_id_list = []
if auth_data:
auth_rule_id_list = [int(item) for item in auth_data.split(',')]
pass
if contains_rules:
contains_rules_checked = False
for item in auth_rule_id_list:
if item in contains_rules:
contains_rules_checked = True
break
else:
contains_rules_checked = True
if requisite_rules:
requisite_rules_checked = True
for item in requisite_rules:
if item not in auth_rule_id_list:
requisite_rules_checked = False
break
else:
requisite_rules_checked = True
return contains_rules_checked and requisite_rules_checked