49 lines
1.7 KiB
Python
49 lines
1.7 KiB
Python
import time
|
|
from typing import Annotated, Optional
|
|
from redis import Redis
|
|
from sqlalchemy.orm import Session
|
|
from fastapi import Depends, Header
|
|
|
|
from api.core.exception import ResponseException
|
|
from database.connect import get_db, get_redis
|
|
from api.core.Auth import AlterState
|
|
import base64
|
|
import json
|
|
|
|
SessionDep = Annotated[Session, Depends(get_db)]
|
|
RedisDep = Annotated[Redis, Depends(get_redis)]
|
|
|
|
|
|
def alterToken(alter_token: Optional[str] = Header(None)):
|
|
if alter_token is not None:
|
|
try:
|
|
decoded_bytes = base64.b64decode(alter_token)
|
|
decoded_str = decoded_bytes.decode('utf-8')
|
|
alter = json.loads(decoded_str)
|
|
# 时间判断 后面再加
|
|
if (int(time.time() * 1000) - alter['time']) / (60 * 60 * 1000) > 6:
|
|
raise ResponseException('登录过期或无效Token，请刷新后重试')
|
|
return AlterState(alter['alter'], alter_token, {}, alter['equVersion'], alter['time'])
|
|
except (base64.binascii.Error, json.JSONDecodeError):
|
|
raise ResponseException('登录过期或无效Token，请刷新后重试')
|
|
else:
|
|
raise ResponseException('登录过期或无效Token，请刷新后重试')
|
|
pass
|
|
|
|
|
|
AltersDep = Annotated[AlterState, Depends(alterToken)]
|
|
|
|
# def authorize(redis:RedisDep, access_token: Optional[str] = Header(None)):
|
|
# if access_token is not None:
|
|
# info = readToken(access_token, redis)
|
|
# return info
|
|
# else:
|
|
# raise ResponseException('登录过期或无效Token，请刷新后重试')
|
|
|
|
|
|
# def unauthorize(redis:RedisDep, access_token: Optional[str] = Header(None)):
|
|
# if access_token is not None:
|
|
# return deleteToken(access_token, redis)
|
|
|
|
# TokenDep = Annotated[AlterState, Depends(authorize)]
|