wsb1224
05c421091b
- parse-order和parse-customer接口require_roles添加wuliu - wuliu默认权限补全ai:parse-order Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
611 lines
24 KiB
Python
611 lines
24 KiB
Python
"""认证授权服务层。
|
||
|
||
负责用户登录、登出、Token 管理、权限菜单加载等认证授权业务逻辑。
|
||
依赖 SystemRepository 进行用户和角色数据查询，
|
||
依赖 security 模块进行密码验证和 JWT Token 生成/解析。
|
||
支持将已吊销的 Token 持久化到本地文件，实现跨进程的登出生效。
|
||
"""
|
||
|
||
import json
|
||
import time
|
||
from datetime import datetime, timezone
|
||
from pathlib import Path
|
||
from urllib import error, request as urllib_request
|
||
|
||
from sqlalchemy.exc import SQLAlchemyError
|
||
from sqlalchemy.orm import Session
|
||
|
||
from backend.app.core.config import get_settings
|
||
from backend.app.core.error_codes import ErrorCode
|
||
from backend.app.core.exceptions import AppException
|
||
from backend.app.core.security import create_access_token, decode_access_token, verify_password
|
||
from backend.app.repositories.system_repository import SystemRepository
|
||
|
||
|
||
class AuthService:
|
||
"""认证服务，封装登录、登出、Token 验证和权限加载逻辑。
|
||
|
||
依赖:
|
||
SystemRepository - 用户、角色、菜单数据访问
|
||
security 模块 - 密码验证、JWT Token 生成/解析
|
||
settings - Token 存储路径等配置
|
||
|
||
类属性:
|
||
DEFAULT_ROLE_MENUS - 各角色的默认菜单配置（数据库无菜单数据时的兜底）
|
||
DEFAULT_ROLE_PERMISSIONS - 各角色的默认权限编码配置（兜底）
|
||
"""
|
||
|
||
DEFAULT_ROLE_MENUS = {
|
||
"salesman": [
|
||
{"menu_name": "订单中心", "menu_path": "/orders"},
|
||
{"menu_name": "创建订单", "menu_path": "/orders/create"},
|
||
{"menu_name": "提醒中心", "menu_path": "/reminders"},
|
||
{"menu_name": "客户管理", "menu_path": "/customers"},
|
||
],
|
||
"manager": [
|
||
{"menu_name": "后台首页", "menu_path": "/dashboard"},
|
||
{"menu_name": "订单中心", "menu_path": "/orders"},
|
||
{"menu_name": "创建订单", "menu_path": "/orders/create"},
|
||
{"menu_name": "提醒中心", "menu_path": "/reminders"},
|
||
{"menu_name": "客户管理", "menu_path": "/customers"},
|
||
{"menu_name": "报表统计", "menu_path": "/reports/performance"},
|
||
{"menu_name": "AI识别", "menu_path": "/ai-recognition"},
|
||
],
|
||
"driver": [],
|
||
"admin": [{"menu_name": "后台首页", "menu_path": "/dashboard"}],
|
||
"kuaiji": [
|
||
{"menu_name": "后台首页", "menu_path": "/dashboard"},
|
||
{"menu_name": "订单中心", "menu_path": "/orders"},
|
||
{"menu_name": "客户管理", "menu_path": "/customers"},
|
||
{"menu_name": "报表统计", "menu_path": "/reports/performance"},
|
||
{"menu_name": "报表核对", "menu_path": "/reports/audit"},
|
||
{"menu_name": "配置管理", "menu_path": "/config"},
|
||
],
|
||
"wuliu": [
|
||
{"menu_name": "后台首页", "menu_path": "/dashboard"},
|
||
{"menu_name": "订单中心", "menu_path": "/orders"},
|
||
{"menu_name": "创建订单", "menu_path": "/orders/create"},
|
||
{"menu_name": "批量匹配物流", "menu_path": "/orders/batch-match"},
|
||
{"menu_name": "客户管理", "menu_path": "/customers"},
|
||
{"menu_name": "提醒中心", "menu_path": "/reminders"},
|
||
{"menu_name": "配置管理", "menu_path": "/config"},
|
||
],
|
||
}
|
||
|
||
DEFAULT_ROLE_PERMISSIONS = {
|
||
"salesman": [
|
||
"order:list",
|
||
"order:create",
|
||
"order:submit",
|
||
"customer:list",
|
||
"reminder:list",
|
||
"ai:parse-order",
|
||
"logistics:trace:list",
|
||
],
|
||
"manager": [
|
||
"order:list",
|
||
"order:create",
|
||
"order:submit",
|
||
"order:cancel-approve",
|
||
"customer:list",
|
||
"reminder:list",
|
||
"report:performance:view",
|
||
"report:performance:export",
|
||
"ai:recognize",
|
||
"ai:correct",
|
||
"ai:parse-order",
|
||
],
|
||
"driver": [
|
||
"driver:task:list",
|
||
"driver:task:accept",
|
||
"driver:task:pickup",
|
||
"driver:task:deliver",
|
||
"logistics:trace:list",
|
||
],
|
||
"admin": [
|
||
"dashboard:view",
|
||
"customer:list",
|
||
"master-data:list",
|
||
"master-data:update",
|
||
"pricing-rules:view",
|
||
"order:list",
|
||
"order:create",
|
||
"order:submit",
|
||
"order:cancel-approve",
|
||
"order:batch-match",
|
||
"reminder:list",
|
||
"config:view",
|
||
"audit:view",
|
||
"system:view",
|
||
"system:user:list",
|
||
"system:user:create",
|
||
"system:user:update",
|
||
"system:user:reset-password",
|
||
"system:role:list",
|
||
"system:role:create",
|
||
"system:role:update",
|
||
"system:role:assign-menus",
|
||
"system:menu:list",
|
||
"system:menu:create",
|
||
"system:menu:update",
|
||
"report:performance:view",
|
||
"report:performance:export",
|
||
"report:audit:view",
|
||
"ai:recognize",
|
||
"ai:correct",
|
||
"ai:parse-order",
|
||
],
|
||
"kuaiji": [
|
||
"dashboard:view",
|
||
"order:list",
|
||
"customer:list",
|
||
"report:performance:view",
|
||
"report:performance:export",
|
||
"report:audit:view",
|
||
"config:view",
|
||
],
|
||
"wuliu": [
|
||
"dashboard:view",
|
||
"order:list",
|
||
"order:create",
|
||
"order:batch-match",
|
||
"customer:list",
|
||
"ai:parse-order",
|
||
"reminder:list",
|
||
"config:view",
|
||
],
|
||
}
|
||
|
||
def __init__(self) -> None:
|
||
self.repository = SystemRepository()
|
||
self.settings = get_settings()
|
||
self.token_store_path = Path(self.settings.auth_token_store_path)
|
||
self.token_store_path.parent.mkdir(parents=True, exist_ok=True)
|
||
self.revoked_tokens = self._load_revoked_tokens()
|
||
|
||
# 后台管理系统允许登录的角色列表
|
||
ADMIN_ALLOWED_ROLES = {"admin", "manager", "kuaiji", "wuliu", "salesman", "driver"}
|
||
|
||
def login(self, username: str, password: str, role_type: str, session: Session | None = None) -> dict | None:
|
||
"""用户登录。
|
||
|
||
验证用户名密码、校验角色匹配和账号状态，生成 JWT Token，
|
||
加载菜单和权限信息。
|
||
|
||
Args:
|
||
username: 用户名
|
||
password: 密码（明文）
|
||
role_type: 登录的角色类型（如 admin），为空时根据账号角色自动判断
|
||
session: 数据库会话
|
||
|
||
Returns:
|
||
用户信息字典（含 token、menus、permissions），登录失败抛出异常
|
||
|
||
被调用路由: auth.py - POST /auth/login
|
||
"""
|
||
if session is not None:
|
||
try:
|
||
user = self.repository.get_user_by_username(session, username)
|
||
if user is None or not verify_password(password, user.password_hash):
|
||
raise AppException(
|
||
code=ErrorCode.UNAUTHORIZED,
|
||
message="账号不存在或密码错误",
|
||
status_code=401,
|
||
)
|
||
|
||
role = self.repository.get_role(session, user.role_id)
|
||
if role is None:
|
||
raise AppException(code=ErrorCode.NOT_FOUND, message="角色不存在", status_code=404)
|
||
if user.status != 1:
|
||
raise AppException(code=ErrorCode.FORBIDDEN, message="账号已停用", status_code=403)
|
||
if role.status != 1:
|
||
raise AppException(code=ErrorCode.FORBIDDEN, message="角色已禁用", status_code=403)
|
||
|
||
# 检查角色是否允许登录后台
|
||
if role.role_code not in self.ADMIN_ALLOWED_ROLES:
|
||
raise AppException(
|
||
code=ErrorCode.FORBIDDEN,
|
||
message="该角色无权登录后台管理系统",
|
||
status_code=403,
|
||
)
|
||
|
||
# 如果指定了 role_type，检查是否匹配
|
||
if role_type and role.role_code != role_type:
|
||
raise AppException(code=ErrorCode.FORBIDDEN, message="登录角色与账号不匹配", status_code=403)
|
||
|
||
return self._build_profile(session, user, role, include_token=True)
|
||
except AppException:
|
||
raise
|
||
except SQLAlchemyError as exc:
|
||
raise AppException(code=ErrorCode.SYSTEM_ERROR, message="数据库异常", status_code=500) from exc
|
||
|
||
raise AppException(code=ErrorCode.SYSTEM_ERROR, message="数据库连接不可用", status_code=500)
|
||
|
||
def wechat_login(self, code: str, phone_code: str, session: Session | None = None) -> dict:
|
||
"""微信手机号一键登录。
|
||
|
||
用 code 换取 session_key 和 open_id，用 phone_code 换取手机号，
|
||
匹配系统用户并自动绑定 open_id。
|
||
|
||
Args:
|
||
code: wx.login() 获取的临时凭证
|
||
phone_code: wx.getPhoneNumber 返回的授权 code
|
||
session: 数据库会话
|
||
|
||
Returns:
|
||
用户信息字典（含 token），手机号未匹配时抛出异常
|
||
|
||
被调用路由: auth.py - POST /auth/wechat-login
|
||
"""
|
||
if session is None:
|
||
raise AppException(code=ErrorCode.SYSTEM_ERROR, message="数据库连接不可用", status_code=500)
|
||
|
||
settings = self.settings
|
||
if not settings.wechat_app_id or not settings.wechat_app_secret:
|
||
raise AppException(code=ErrorCode.SYSTEM_ERROR, message="微信配置未就绪，请联系管理员", status_code=500)
|
||
|
||
try:
|
||
# 1. 用 code 换取 session_key 和 open_id
|
||
session_key, open_id = self._code2session(code, settings)
|
||
if not session_key or not open_id:
|
||
raise AppException(code=ErrorCode.UNAUTHORIZED, message="微信登录凭证无效", status_code=401)
|
||
|
||
# 2. 用 phone_code 换取手机号
|
||
phone_number = self._get_phone_number(phone_code, settings)
|
||
if not phone_number:
|
||
raise AppException(code=ErrorCode.UNAUTHORIZED, message="手机号获取失败", status_code=401)
|
||
|
||
# 3. 用手机号匹配系统用户
|
||
user = self.repository.get_user_by_mobile(session, phone_number)
|
||
if user is None:
|
||
raise AppException(code=ErrorCode.NOT_FOUND, message=f"手机号 {phone_number} 未注册系统账号", status_code=404)
|
||
|
||
# 4. 自动绑定 open_id
|
||
user.open_id = open_id
|
||
session.add(user)
|
||
session.commit()
|
||
|
||
# 5. 加载角色并返回
|
||
role = self.repository.get_role(session, user.role_id)
|
||
if role is None:
|
||
raise AppException(code=ErrorCode.NOT_FOUND, message="角色不存在", status_code=404)
|
||
|
||
return self._build_profile(session, user, role, include_token=True)
|
||
except AppException:
|
||
raise
|
||
except SQLAlchemyError as exc:
|
||
session.rollback()
|
||
raise AppException(code=ErrorCode.SYSTEM_ERROR, message="数据库异常", status_code=500) from exc
|
||
except Exception as exc:
|
||
raise AppException(code=ErrorCode.SYSTEM_ERROR, message=f"微信登录失败: {exc}", status_code=500) from exc
|
||
|
||
def _code2session(self, code: str, settings) -> tuple[str, str]:
|
||
"""用 code 调用微信 code2Session 接口，获取 session_key 和 open_id。"""
|
||
url = (
|
||
f"https://api.weixin.qq.com/sns/jscode2session"
|
||
f"?appid={settings.wechat_app_id}"
|
||
f"&secret={settings.wechat_app_secret}"
|
||
f"&js_code={code}"
|
||
f"&grant_type=authorization_code"
|
||
)
|
||
req = urllib_request.Request(url=url, method="GET")
|
||
try:
|
||
with urllib_request.urlopen(req, timeout=10) as response:
|
||
result = json.loads(response.read().decode("utf-8") or "{}")
|
||
return result.get("session_key", ""), result.get("openid", "")
|
||
except (error.HTTPError, error.URLError, json.JSONDecodeError):
|
||
return "", ""
|
||
|
||
def _get_phone_number(self, phone_code: str, settings) -> str:
|
||
"""用 phone_code 调用微信 getuserphonenumber 接口获取手机号。"""
|
||
access_token = self._get_access_token(settings)
|
||
if not access_token:
|
||
return ""
|
||
url = f"https://api.weixin.qq.com/wxa/business/getuserphonenumber?access_token={access_token}"
|
||
body = json.dumps({"code": phone_code}).encode("utf-8")
|
||
req = urllib_request.Request(url=url, data=body, headers={"Content-Type": "application/json"}, method="POST")
|
||
try:
|
||
with urllib_request.urlopen(req, timeout=10) as response:
|
||
result = json.loads(response.read().decode("utf-8") or "{}")
|
||
if result.get("errcode") == 0:
|
||
return result.get("phone_info", {}).get("purePhoneNumber", "")
|
||
except (error.HTTPError, error.URLError, json.JSONDecodeError):
|
||
pass
|
||
return ""
|
||
|
||
def _get_access_token(self, settings) -> str:
|
||
"""获取微信 access_token（简单缓存）。"""
|
||
now = time.time()
|
||
if not hasattr(self, "_token_cache"):
|
||
self._token_cache = {"token": "", "expires_at": 0}
|
||
if self._token_cache["token"] and self._token_cache["expires_at"] > now + 60:
|
||
return self._token_cache["token"]
|
||
url = (
|
||
f"https://api.weixin.qq.com/cgi-bin/token"
|
||
f"?grant_type=client_credential"
|
||
f"&appid={settings.wechat_app_id}"
|
||
f"&secret={settings.wechat_app_secret}"
|
||
)
|
||
req = urllib_request.Request(url=url, method="GET")
|
||
try:
|
||
with urllib_request.urlopen(req, timeout=10) as response:
|
||
result = json.loads(response.read().decode("utf-8") or "{}")
|
||
token = result.get("access_token", "")
|
||
expires_in = int(result.get("expires_in", 0))
|
||
if token and expires_in > 0:
|
||
self._token_cache["token"] = token
|
||
self._token_cache["expires_at"] = now + expires_in
|
||
return token
|
||
except (error.HTTPError, error.URLError, json.JSONDecodeError):
|
||
pass
|
||
return ""
|
||
|
||
def bind_openid_by_code(self, user_id: int, code: str, session: Session | None = None) -> dict:
|
||
"""通过 wx.login code 绑定 open_id 到指定用户。
|
||
|
||
用于密码登录后自动绑定 open_id，为微信订阅消息通知做准备。
|
||
|
||
Args:
|
||
user_id: 用户 ID
|
||
code: wx.login() 获取的临时凭证
|
||
session: 数据库会话
|
||
|
||
Returns:
|
||
{"success": True}
|
||
|
||
被调用路由: auth.py - POST /auth/bind-openid-by-code
|
||
"""
|
||
if session is None:
|
||
raise AppException(code=ErrorCode.SYSTEM_ERROR, message="数据库连接不可用", status_code=500)
|
||
|
||
settings = self.settings
|
||
if not settings.wechat_app_id or not settings.wechat_app_secret:
|
||
return {"success": True, "skipped": True}
|
||
|
||
_, open_id = self._code2session(code, settings)
|
||
if not open_id:
|
||
return {"success": True, "skipped": True}
|
||
|
||
user = self.repository.get_user(session, user_id)
|
||
if user is not None:
|
||
user.open_id = open_id
|
||
session.add(user)
|
||
session.commit()
|
||
|
||
return {"success": True}
|
||
|
||
def get_me(self, token: str, session: Session | None = None) -> dict | None:
|
||
"""根据 Token 获取当前用户信息（无需重新登录）。
|
||
|
||
解析 JWT Token 获取用户 ID 和角色，验证 Token 未被吊销，
|
||
重新加载最新的用户信息和权限。
|
||
|
||
Args:
|
||
token: JWT Token 字符串
|
||
session: 数据库会话
|
||
|
||
Returns:
|
||
用户信息字典（不含 token），验证失败返回 None
|
||
|
||
被调用路由: deps.py - get_current_user 依赖注入
|
||
"""
|
||
if self._is_token_revoked(token):
|
||
return None
|
||
|
||
if session is not None:
|
||
try:
|
||
payload = decode_access_token(token)
|
||
if payload is not None:
|
||
user_id = int(payload["user_id"])
|
||
role_code = str(payload["role_code"])
|
||
user = self.repository.get_user(session, user_id)
|
||
if user is None or user.status != 1:
|
||
return None
|
||
role = self.repository.get_role(session, user.role_id)
|
||
if role is None or role.status != 1 or role.role_code != role_code:
|
||
return None
|
||
return self._build_profile(session, user, role, include_token=False)
|
||
except (SQLAlchemyError, ValueError, KeyError, TypeError):
|
||
pass
|
||
|
||
return None
|
||
|
||
def logout(self, token: str) -> dict:
|
||
"""用户登出。
|
||
|
||
将当前 Token 加入吊销列表并持久化到本地文件，确保跨进程生效。
|
||
|
||
Args:
|
||
token: 当前 JWT Token
|
||
|
||
Returns:
|
||
{"success": True}
|
||
|
||
被调用路由: auth.py - POST /auth/logout
|
||
"""
|
||
if token:
|
||
# token 失效需要跨进程持久化，避免服务重启后已退出的 token 又重新可用。
|
||
self._revoke_token(token)
|
||
return {"success": True}
|
||
|
||
def _build_profile(self, session: Session, user, role, include_token: bool) -> dict:
|
||
"""构建用户信息字典（含菜单和权限）。
|
||
|
||
Args:
|
||
session: 数据库会话
|
||
user: 用户 ORM 对象
|
||
role: 角色 ORM 对象
|
||
include_token: 是否生成 JWT Token（登录时为 True，get_me 时为 False）
|
||
|
||
Returns:
|
||
用户信息字典
|
||
"""
|
||
menus = self._load_role_menus(session, role.role_code)
|
||
permissions = self._load_role_permissions(session, role.role_code)
|
||
token = None
|
||
token_expires_in = None
|
||
if include_token:
|
||
token = create_access_token({"user_id": user.id, "role_code": role.role_code})
|
||
token_expires_in = self.settings.jwt_expire_minutes * 60 # 转换为秒
|
||
return {
|
||
"user_id": user.id,
|
||
"username": user.username,
|
||
"real_name": user.real_name,
|
||
"mobile": user.mobile,
|
||
"role_id": role.id,
|
||
"role_name": role.role_name,
|
||
"role_code": role.role_code,
|
||
"token": token,
|
||
"token_expires_in": token_expires_in,
|
||
"menus": menus,
|
||
"permissions": permissions,
|
||
}
|
||
|
||
def _load_role_menus(self, session: Session, role_code: str) -> list[dict]:
|
||
"""加载角色对应的菜单列表。
|
||
|
||
所有角色都通过 role_menu 关系表过滤，只返回被授权的菜单。
|
||
若数据库无菜单数据则使用 DEFAULT_ROLE_MENUS 兜底。
|
||
|
||
Args:
|
||
session: 数据库会话
|
||
role_code: 角色编码
|
||
|
||
Returns:
|
||
菜单列表，每项包含 menu_name 和 menu_path
|
||
"""
|
||
menus = self.repository.list_menus(session, {"status": 1})
|
||
if not menus:
|
||
return self.DEFAULT_ROLE_MENUS.get(role_code, [])
|
||
|
||
role = self.repository.get_role_by_code(session, role_code)
|
||
if role is not None:
|
||
relations = self.repository.list_role_menu_relations(session, role.id)
|
||
if relations:
|
||
allowed_menu_ids = {item.menu_id for item in relations}
|
||
filtered = [
|
||
{"menu_name": menu.menu_name, "menu_path": menu.menu_path}
|
||
for menu in menus
|
||
if menu.id in allowed_menu_ids and menu.menu_path
|
||
]
|
||
if filtered:
|
||
return filtered
|
||
|
||
# 如果没有找到角色-菜单关联关系，使用默认配置
|
||
allowed_paths = {item["menu_path"] for item in self.DEFAULT_ROLE_MENUS.get(role_code, [])}
|
||
filtered = [
|
||
{"menu_name": menu.menu_name, "menu_path": menu.menu_path}
|
||
for menu in menus
|
||
if menu.menu_path in allowed_paths
|
||
]
|
||
return filtered or self.DEFAULT_ROLE_MENUS.get(role_code, [])
|
||
|
||
def _load_role_permissions(self, session: Session, role_code: str) -> list[str]:
|
||
"""加载角色对应的权限编码列表。
|
||
|
||
admin 角色返回所有权限，其他角色通过 role_menu 关系表过滤。
|
||
若数据库无菜单数据则使用 DEFAULT_ROLE_PERMISSIONS 兜底。
|
||
最终结果会与 DEFAULT_ROLE_PERMISSIONS 合并，确保基础权限不丢失。
|
||
|
||
Args:
|
||
session: 数据库会话
|
||
role_code: 角色编码
|
||
|
||
Returns:
|
||
权限编码字符串列表
|
||
"""
|
||
default_permissions = self.DEFAULT_ROLE_PERMISSIONS.get(role_code, [])
|
||
menus = self.repository.list_menus(session, {"status": 1})
|
||
if not menus:
|
||
return default_permissions
|
||
|
||
if role_code == "admin":
|
||
permissions = [menu.permission_code for menu in menus if menu.permission_code]
|
||
merged = list(dict.fromkeys([*default_permissions, *permissions]))
|
||
return merged or default_permissions
|
||
|
||
db_permissions = []
|
||
role = self.repository.get_role_by_code(session, role_code)
|
||
if role is not None:
|
||
relations = self.repository.list_role_menu_relations(session, role.id)
|
||
if relations:
|
||
allowed_menu_ids = {item.menu_id for item in relations}
|
||
db_permissions = [
|
||
menu.permission_code
|
||
for menu in menus
|
||
if menu.id in allowed_menu_ids and menu.permission_code
|
||
]
|
||
|
||
# 合并数据库权限和默认权限，确保基础权限不丢失
|
||
merged = list(dict.fromkeys([*default_permissions, *db_permissions]))
|
||
return merged
|
||
|
||
def _load_revoked_tokens(self) -> dict[str, int]:
|
||
"""从本地文件加载已吊销的 Token 列表。
|
||
|
||
自动清理已过期的 Token 条目。
|
||
|
||
Returns:
|
||
Token 到过期时间戳的映射字典
|
||
"""
|
||
if not self.token_store_path.exists():
|
||
return {}
|
||
try:
|
||
payload = json.loads(self.token_store_path.read_text(encoding="utf-8"))
|
||
except (OSError, json.JSONDecodeError):
|
||
return {}
|
||
now_ts = int(datetime.now(timezone.utc).timestamp())
|
||
tokens = {
|
||
token: int(exp)
|
||
for token, exp in payload.items()
|
||
if isinstance(token, str) and isinstance(exp, int) and exp > now_ts
|
||
}
|
||
if len(tokens) != len(payload):
|
||
self._save_revoked_tokens(tokens)
|
||
return tokens
|
||
|
||
def _save_revoked_tokens(self, tokens: dict[str, int]) -> None:
|
||
"""将已吊销的 Token 列表持久化到本地文件。
|
||
|
||
Args:
|
||
tokens: Token 到过期时间戳的映射字典
|
||
"""
|
||
self.token_store_path.write_text(json.dumps(tokens, ensure_ascii=False, indent=2), encoding="utf-8")
|
||
|
||
def _is_token_revoked(self, token: str) -> bool:
|
||
"""检查 Token 是否已被吊销。
|
||
|
||
同时清理已过期的吊销条目。
|
||
|
||
Args:
|
||
token: JWT Token 字符串
|
||
|
||
Returns:
|
||
True 表示已吊销
|
||
"""
|
||
if not token:
|
||
return False
|
||
now_ts = int(datetime.now(timezone.utc).timestamp())
|
||
expired_tokens = [item for item, exp in self.revoked_tokens.items() if exp <= now_ts]
|
||
if expired_tokens:
|
||
for item in expired_tokens:
|
||
self.revoked_tokens.pop(item, None)
|
||
self._save_revoked_tokens(self.revoked_tokens)
|
||
return token in self.revoked_tokens
|
||
|
||
def _revoke_token(self, token: str) -> None:
|
||
"""吊销指定 Token 并持久化。
|
||
|
||
解析 Token 的过期时间，存入吊销列表并写入本地文件。
|
||
|
||
Args:
|
||
token: 需要吊销的 JWT Token
|
||
"""
|
||
payload = decode_access_token(token)
|
||
expire_at = int(payload.get("exp", 0)) if payload else int(datetime.now(timezone.utc).timestamp()) + 60
|
||
self.revoked_tokens[token] = expire_at
|
||
self._save_revoked_tokens(self.revoked_tokens)
|
||
|
||
|
||
auth_service = AuthService()
|