624 lines
22 KiB
Python
624 lines
22 KiB
Python
# +======================================================================
|
||
# | DjangoAdmin 企业数字化开发平台 [ 赋能开发者，助力企业发展 ]
|
||
# +----------------------------------------------------------------------
|
||
# | Copyright (c) 2021-2026 小蚂蚁云团队 版权所有
|
||
# +----------------------------------------------------------------------
|
||
# | 开源协议: Apache-2.0
|
||
# | 重要声明: 未经书面许可，不得移除或修改本框架相关版权标识。
|
||
# +----------------------------------------------------------------------
|
||
# | 官方网站: https://www.djangoadmin.cn
|
||
# +----------------------------------------------------------------------
|
||
# | 研发团队: 小蚂蚁云团队 荣誉出品
|
||
# +======================================================================
|
||
# | 版权和免责声明:
|
||
# | 一、知识产权声明
|
||
# | 本团队对该软件框架产品（以下简称“本产品”）拥有完整知识产权，包括但不限于商标权、专利权、
|
||
# | 著作权、商业秘密等。前述权利均受中华人民共和国及相关国际法律法规保护。未经本团队书面授权，
|
||
# | 任何个人或单位不得基于本产品本身申请任何形式的知识产权。
|
||
# | 二、使用限制与禁止条款
|
||
# | 您理解并同意，禁止将本产品用于以下任何行为：
|
||
# | 1. 任何违反中华人民共和国法律法规的用途；
|
||
# | 2. 任何侵害他人合法权益（包括但不限于人身权、财产权、知识产权等）的行为；
|
||
# | 3. 任何危害国家安全、公共利益及社会道德风尚的行为。
|
||
# | 三、责任豁免
|
||
# | 因使用本产品而产生的任何损失或法律纠纷（包括但不限于直接损失、间接损失、附带损失、衍生损失
|
||
# | 以及因违约、侵权等引发的责任），本团队均不承担任何法律责任。若因违规使用本产品导致任何第三
|
||
# | 方提起索赔或行政机关进行调查，您应独立承担全部责任。
|
||
# | 四、违约与法律责任
|
||
# | 如发现任何违反本声明的行为，本团队有权：
|
||
# | 1. 无条件配合公安机关及其他有权机关的调查取证工作；
|
||
# | 2. 依法追究相关方的民事责任，并要求赔偿全部损失；
|
||
# | 3. 保留进一步采取刑事控告等法律措施的权利。
|
||
# | 五、许可范围说明
|
||
# | 1、被授权方应视本产品源代码为核心机密信息，负有最高标准的保密与保管义务。禁止以任何形式（包
|
||
# | 括但不限于直接或间接的分享、复制、传播、分发、转售、出租、开源、反向工程、反编译、反汇编或通
|
||
# | 过网络服务提供给任何第三方）泄露、披露或处置源代码的全部或部分内容。任何未经授权的复制、使用、
|
||
# | 传播行为均构成对本团队知识产权的严重侵权，本团队有权依法追究其全部法律责任，并要求赔偿因此造
|
||
# | 成的一切经济损失及商誉损害。
|
||
# | 2、本产品仅限于公司或个人在法律允许范围内，用于内部且合规的软件产品研发。具体权利义务以《软件
|
||
# | 许可协议》附件为准。
|
||
# +======================================================================
|
||
|
||
# +======================================================================
|
||
# | 模块: JWT令牌处理
|
||
# | 说明: 提供JWT令牌的生成、验证、刷新和黑名单管理功能
|
||
# +======================================================================
|
||
|
||
"""
|
||
JWT令牌处理模块
|
||
=============================================================
|
||
|
||
该模块提供基于JWT（JSON Web Token）的令牌生成和验证功能。
|
||
主要用于用户登录认证，生成访问令牌，并在后续请求中验证令牌的有效性。
|
||
支持令牌过期机制，确保系统的安全性。
|
||
|
||
主要功能：
|
||
1. JWT令牌生成（create_token）- 用于用户登录成功后生成访问令牌
|
||
2. JWT令牌解析验证（parse_payload）- 验证请求中的令牌有效性
|
||
3. 统一的异常处理，返回标准格式的验证结果
|
||
4. 可配置的令牌过期时间
|
||
|
||
所属包： utils
|
||
所在文件： utils/jwt.py
|
||
依赖模块： jwt - JWT处理第三方库
|
||
datetime - 时间处理模块
|
||
hashlib - 摘要算法模块
|
||
os - 环境变量读取模块
|
||
logging - 日志记录模块
|
||
"""
|
||
|
||
import datetime
|
||
import hashlib
|
||
import logging
|
||
import os
|
||
import time
|
||
|
||
import jwt
|
||
from jwt import exceptions
|
||
|
||
# 配置日志记录器
|
||
logger = logging.getLogger(__name__)
|
||
|
||
|
||
# =============================================================
|
||
# JWT密钥配置
|
||
# =============================================================
|
||
# 优先从环境变量读取JWT密钥
|
||
JWT_SALT = os.environ.get('JWT_SALT', None)
|
||
|
||
# 如果没有环境变量，使用默认密钥
|
||
if not JWT_SALT:
|
||
JWT_SALT = "ds()udsjo@jlsdosjf)wjd_#(#)$%^&*()_+=-0987654321"
|
||
|
||
# JWT算法配置
|
||
JWT_ALGORITHM = 'HS256'
|
||
|
||
# 默认令牌过期时间（分钟）
|
||
DEFAULT_TIMEOUT_MINUTES = 20
|
||
|
||
|
||
def _validate_key_strength(key: str) -> None:
|
||
"""
|
||
启动时校验 JWT 密钥强度
|
||
|
||
校验规则：
|
||
1. 密钥不能为空
|
||
2. 密钥长度至少 32 字节（256 位）
|
||
|
||
校验失败直接抛出 RuntimeError，拒绝启动应用。
|
||
"""
|
||
if not key:
|
||
raise RuntimeError(
|
||
"JWT_SALT 未配置！请在 .env 中设置至少32字节的强随机密钥。\n"
|
||
"生成方式：python -c \"import secrets; print(secrets.token_urlsafe(48))\""
|
||
)
|
||
|
||
key_bytes = len(key.encode('utf-8'))
|
||
if key_bytes < 32:
|
||
logger.warning(
|
||
f"JWT密钥长度不足：{key_bytes} 字节，建议至少 32 字节（256位）。"
|
||
)
|
||
|
||
logger.info(f"JWT密钥校验通过，长度：{key_bytes}字节")
|
||
|
||
|
||
# 启动时校验密钥强度
|
||
_validate_key_strength(JWT_SALT)
|
||
|
||
|
||
# =============================================================
|
||
# JWT核心功能函数
|
||
# =============================================================
|
||
|
||
def create_token(payload, timeout=None):
|
||
"""
|
||
生成JWT令牌
|
||
=============================================================
|
||
|
||
根据传入的载荷数据生成JWT访问令牌，自动添加过期时间。
|
||
使用强密钥确保令牌安全性，符合RFC 7518安全规范。
|
||
|
||
参数:
|
||
payload (dict): JWT载荷数据，通常包含用户ID、用户名等信息
|
||
timeout (int): 令牌过期时间（分钟），默认使用DEFAULT_TIMEOUT_MINUTES
|
||
|
||
返回:
|
||
str: 生成的JWT令牌字符串
|
||
|
||
使用示例:
|
||
>>> payload = {'userId': 1001, 'username': 'admin'}
|
||
>>> token = create_token(payload, timeout=30)
|
||
>>> print(token)
|
||
eyJ0eXAiOiJKV1QiLCJhbGciOiJIUzI1NiJ9...
|
||
|
||
注意事项:
|
||
1. 密钥长度必须至少32字节，否则会发出警告
|
||
2. 生产环境建议通过环境变量设置JWT_SALT
|
||
3. 令牌过期后需要重新登录获取新令牌
|
||
"""
|
||
try:
|
||
# 设置过期时间
|
||
# 说明：在载荷中添加exp字段，指定令牌的过期时间点
|
||
# 计算方式：当前UTC时间 + 指定的分钟数
|
||
if timeout is None:
|
||
timeout = DEFAULT_TIMEOUT_MINUTES
|
||
|
||
# 创建payload副本，避免修改原始数据
|
||
payload_copy = payload.copy()
|
||
|
||
# 添加过期时间（exp）和签发时间（iat）
|
||
# exp: 过期时间，超过此时间令牌无效
|
||
# iat: 签发时间，用于令牌生命周期计算
|
||
current_time = datetime.datetime.now(tz=datetime.timezone.utc)
|
||
payload_copy['exp'] = current_time + datetime.timedelta(minutes=timeout)
|
||
payload_copy['iat'] = current_time
|
||
|
||
# 声明类型和加密算法
|
||
# 说明：定义JWT头部信息，包含令牌类型和加密算法
|
||
headers = {
|
||
"typ": "JWT", # 令牌类型
|
||
"alg": JWT_ALGORITHM # 加密算法
|
||
}
|
||
|
||
# 生成JWT令牌
|
||
# 函数：jwt.encode()
|
||
# 参数说明：
|
||
# payload: JWT载荷数据
|
||
# key: 加密密钥（盐）
|
||
# algorithm: 加密算法（HS256）
|
||
# headers: JWT头部信息
|
||
token = jwt.encode(
|
||
payload=payload_copy,
|
||
key=JWT_SALT,
|
||
algorithm=JWT_ALGORITHM,
|
||
headers=headers
|
||
)
|
||
|
||
# 记录日志（可选，生产环境建议使用DEBUG级别）
|
||
logger.debug(f"生成JWT令牌成功，过期时间：{timeout}分钟")
|
||
|
||
return token
|
||
|
||
except Exception as e:
|
||
# 捕获并记录生成令牌时的异常
|
||
logger.error(f"生成JWT令牌失败: {str(e)}")
|
||
raise
|
||
|
||
|
||
def parse_payload(token):
|
||
"""
|
||
解析验证JWT令牌
|
||
=============================================================
|
||
|
||
验证传入的JWT令牌，检查签名有效性和过期状态。
|
||
统一处理各种异常情况，返回标准格式的结果对象。
|
||
|
||
参数:
|
||
token (str): 需要验证的JWT令牌字符串
|
||
|
||
返回:
|
||
dict: 验证结果，包含以下字段：
|
||
code (int): 状态码，0-成功，1-失败
|
||
data (dict/None): 验证成功时返回载荷数据，失败时为None
|
||
msg (str): 提示信息
|
||
|
||
使用示例:
|
||
>>> result = parse_payload(token)
|
||
>>> if result['code'] == 0:
|
||
... user_id = result['data']['userId']
|
||
... username = result['data']['username']
|
||
... else:
|
||
... error_msg = result['msg']
|
||
... # 处理令牌失效，跳转登录页
|
||
|
||
异常处理说明:
|
||
- ExpiredSignatureError: 令牌已过期，需要重新登录
|
||
- DecodeError: 解码失败，令牌格式错误或签名无效
|
||
- InvalidTokenError: 其他JWT相关异常
|
||
"""
|
||
# 初始化返回结果
|
||
# 说明：定义统一返回格式，默认成功状态
|
||
result = {
|
||
"code": 0,
|
||
"data": None,
|
||
"msg": "操作成功"
|
||
}
|
||
|
||
# 参数校验
|
||
if not token:
|
||
result['code'] = -1
|
||
result['msg'] = "token不能为空"
|
||
return result
|
||
|
||
# 检查Token是否在黑名单中
|
||
if is_token_blacklisted(token):
|
||
result['code'] = -1
|
||
result['msg'] = "token已失效，请重新登录"
|
||
return result
|
||
|
||
try:
|
||
# 解密验证JWT令牌
|
||
# 函数：jwt.decode()
|
||
# 参数说明：
|
||
# jwt: 令牌字符串
|
||
# key: 解密密钥（必须与加密时使用的密钥一致）
|
||
# algorithms: 允许的算法列表
|
||
#
|
||
# 说明：jwt.decode会自动完成以下验证：
|
||
# 1. 签名验证 - 确保令牌未被篡改
|
||
# 2. 过期时间验证 - 检查exp字段是否大于当前时间
|
||
# 3. 算法验证 - 确保使用正确的加密算法
|
||
verified_payload = jwt.decode(
|
||
token,
|
||
JWT_SALT,
|
||
algorithms=[JWT_ALGORITHM]
|
||
)
|
||
|
||
# 验证成功，将载荷数据存入返回结果
|
||
# 注意：返回的数据中不包含敏感信息（如密码等）
|
||
result['data'] = verified_payload
|
||
|
||
# 可选：记录验证成功的日志（生产环境建议使用DEBUG级别）
|
||
logger.debug(f"JWT令牌验证成功，用户: {verified_payload.get('username', 'unknown')}")
|
||
|
||
except exceptions.ExpiredSignatureError as e:
|
||
# 令牌已过期异常处理
|
||
# 说明：令牌超过设置的exp时间，需要重新登录获取新令牌
|
||
result['code'] = -1
|
||
result['msg'] = "token已失效，请重新登录"
|
||
logger.info(f"JWT令牌已过期: {str(e)}")
|
||
|
||
except exceptions.DecodeError as e:
|
||
# 令牌解码失败异常处理
|
||
# 说明：令牌格式错误或签名验证失败，可能被篡改
|
||
result['code'] = -1
|
||
result['msg'] = "token认证失败，无效的令牌格式"
|
||
logger.warning(f"JWT令牌解码失败: {str(e)}")
|
||
|
||
except exceptions.InvalidTokenError as e:
|
||
# 无效令牌异常处理
|
||
# 说明：捕获其他所有JWT相关异常，如算法不匹配等
|
||
result['code'] = -1
|
||
result['msg'] = "非法的token，请检查令牌有效性"
|
||
logger.warning(f"无效的JWT令牌: {str(e)}")
|
||
|
||
except Exception as e:
|
||
# 其他未预期的异常处理
|
||
result['code'] = -1
|
||
result['msg'] = "token验证失败，系统异常"
|
||
logger.error(f"JWT验证系统异常: {str(e)}")
|
||
|
||
return result
|
||
|
||
|
||
def refresh_token(token, extend_minutes=None):
|
||
"""
|
||
刷新JWT令牌
|
||
=============================================================
|
||
|
||
验证现有令牌有效性，如果有效则生成新的令牌，延长有效期。
|
||
|
||
参数:
|
||
token (str): 需要刷新的JWT令牌字符串
|
||
extend_minutes (int): 延长的分钟数，默认使用DEFAULT_TIMEOUT_MINUTES
|
||
|
||
返回:
|
||
dict: 刷新结果，包含以下字段：
|
||
code (int): 状态码，0-成功，-1-失败
|
||
data (dict/None): 成功时包含新令牌，失败时为None
|
||
msg (str): 提示信息
|
||
|
||
使用示例:
|
||
>>> result = refresh_token(old_token, extend_minutes=30)
|
||
>>> if result['code'] == 0:
|
||
... new_token = result['data']['access_token']
|
||
... else:
|
||
... # 令牌无效，需要重新登录
|
||
... error_msg = result['msg']
|
||
|
||
注意事项:
|
||
1. 令牌只有在有效期内才能刷新
|
||
2. 刷新后会生成全新的令牌，旧令牌立即失效
|
||
3. 建议在令牌剩余有效期少于一半时进行刷新
|
||
"""
|
||
if extend_minutes is None:
|
||
extend_minutes = DEFAULT_TIMEOUT_MINUTES
|
||
|
||
# 先验证原令牌
|
||
result = parse_payload(token)
|
||
|
||
# 如果验证失败，返回错误信息
|
||
if result['code'] != 0:
|
||
return {
|
||
"code": -1,
|
||
"data": None,
|
||
"msg": f"令牌无效，无法刷新: {result['msg']}"
|
||
}
|
||
|
||
try:
|
||
# 获取原载荷数据（不包含exp、iat等JWT标准字段）
|
||
payload = result['data']
|
||
|
||
# 移除JWT标准字段，避免重复添加
|
||
# 注意：保留原始业务数据，移除系统字段
|
||
clean_payload = {k: v for k, v in payload.items()
|
||
if k not in ['exp', 'iat', 'nbf']}
|
||
|
||
# 生成新令牌
|
||
new_token = create_token(clean_payload, timeout=extend_minutes)
|
||
|
||
logger.info(f"JWT令牌刷新成功，延长{extend_minutes}分钟")
|
||
|
||
return {
|
||
"code": 0,
|
||
"data": {
|
||
"access_token": new_token,
|
||
"expires_in": extend_minutes * 60 # 转换为秒
|
||
},
|
||
"msg": "令牌刷新成功"
|
||
}
|
||
|
||
except Exception as e:
|
||
logger.error(f"刷新JWT令牌失败: {str(e)}")
|
||
return {
|
||
"code": -1,
|
||
"data": None,
|
||
"msg": "令牌刷新失败，请重新登录"
|
||
}
|
||
|
||
|
||
def get_token_expiration(token):
|
||
"""
|
||
获取JWT令牌的过期时间
|
||
|
||
验证签名后读取过期时间，确保token未被篡改。
|
||
不过验证过期状态（verify_exp=False），以便调用方
|
||
读取已过期token的过期时间用于剩余时间计算等场景。
|
||
|
||
Args:
|
||
token (str): JWT令牌字符串
|
||
|
||
Returns:
|
||
datetime/None: 令牌过期时间（UTC），解析失败返回None
|
||
"""
|
||
try:
|
||
# 验证签名确保token未被篡改，但不验证过期状态
|
||
payload = jwt.decode(
|
||
token,
|
||
JWT_SALT,
|
||
algorithms=[JWT_ALGORITHM],
|
||
options={"verify_exp": False}
|
||
)
|
||
|
||
exp_timestamp = payload.get('exp')
|
||
if exp_timestamp:
|
||
return datetime.datetime.fromtimestamp(exp_timestamp, tz=datetime.timezone.utc)
|
||
return None
|
||
|
||
except Exception as e:
|
||
logger.warning(f"获取JWT过期时间失败: {str(e)}")
|
||
return None
|
||
|
||
|
||
def get_access_token(request):
|
||
"""
|
||
从请求头中提取 access_token
|
||
|
||
Args:
|
||
request: Django 请求对象
|
||
|
||
Returns:
|
||
tuple: (success, token, message)
|
||
"""
|
||
try:
|
||
auth_header = request.META.get('HTTP_AUTHORIZATION', '')
|
||
if not auth_header:
|
||
return False, None, '缺少 Authorization 头'
|
||
|
||
# Bearer 认证方案前缀大小写不敏感（RFC 7235），兼容 'Bearer'/'bearer'/'BEARER'
|
||
stripped = auth_header.strip()
|
||
if stripped.lower().startswith('bearer '):
|
||
access_token = stripped[7:].strip()
|
||
else:
|
||
access_token = stripped
|
||
if not access_token:
|
||
return False, None, 'token 为空'
|
||
|
||
return True, access_token, 'success'
|
||
|
||
except Exception as e:
|
||
logger.error(f"获取 access_token 异常: {str(e)}")
|
||
return False, None, f'token 提取异常: {str(e)}'
|
||
|
||
|
||
def parse_token(request):
|
||
"""
|
||
解析请求头中的token，返回解析结果
|
||
|
||
Args:
|
||
request: Django 请求对象
|
||
|
||
Returns:
|
||
dict: 包含解析结果和用户信息的字典
|
||
"""
|
||
result = {
|
||
'code': -1,
|
||
'msg': 'token无效',
|
||
'data': {
|
||
'userId': 0,
|
||
'username': None
|
||
}
|
||
}
|
||
|
||
try:
|
||
success, access_token, msg = get_access_token(request)
|
||
if not success:
|
||
result['msg'] = msg
|
||
return result
|
||
|
||
parse_result = parse_payload(access_token)
|
||
|
||
if not isinstance(parse_result, dict):
|
||
result['msg'] = '解析结果格式错误'
|
||
return result
|
||
|
||
code = parse_result.get('code', 1)
|
||
if code != 0:
|
||
result['msg'] = parse_result.get('msg', 'token验证失败')
|
||
return result
|
||
|
||
data = parse_result.get('data', {})
|
||
if not isinstance(data, dict):
|
||
result['msg'] = '数据格式错误'
|
||
return result
|
||
|
||
result['code'] = 0
|
||
result['msg'] = 'success'
|
||
result['data']['userId'] = data.get('userId', 0)
|
||
result['data']['username'] = data.get('username', '')
|
||
result['data']['realname'] = data.get('realname', '')
|
||
|
||
except Exception as e:
|
||
result['msg'] = f'token解析异常: {str(e)}'
|
||
|
||
return result
|
||
|
||
|
||
# =============================================================
|
||
# Token黑名单功能
|
||
# =============================================================
|
||
|
||
# 黑名单Key前缀，用于Redis中存储被注销的Token
|
||
_TOKEN_BLACKLIST_PREFIX = 'token:blacklist:'
|
||
|
||
# 黑名单检查告警节流：首次失败记 ERROR，随后每 _BLACKLIST_ALERT_INTERVAL 秒至多一条 ERROR，期间重复失败降为 DEBUG
|
||
_BLACKLIST_ALERT_INTERVAL = 60.0
|
||
_blacklist_check_last_alert = 0.0
|
||
|
||
|
||
def _log_blacklist_check_alert(error: Exception) -> None:
|
||
"""节流记录黑名单检查失败告警，避免 Redis 故障期间每个请求刷一条 ERROR"""
|
||
global _blacklist_check_last_alert
|
||
now = time.monotonic()
|
||
if now - _blacklist_check_last_alert >= _BLACKLIST_ALERT_INTERVAL:
|
||
_blacklist_check_last_alert = now
|
||
logger.error(
|
||
"Token黑名单检查异常，已按 fail-open 放行（已注销的令牌可能被误放行）: %s", error)
|
||
else:
|
||
logger.debug("Token黑名单检查异常（节流中，跳过 ERROR）: %s", error)
|
||
|
||
|
||
def _token_fingerprint(token: str) -> str:
|
||
"""
|
||
生成Token指纹（SHA256摘要）
|
||
|
||
避免将完整Token存入Redis，仅存储摘要值，减少泄露风险。
|
||
|
||
Args:
|
||
token: JWT令牌字符串
|
||
|
||
Returns:
|
||
str: SHA256十六进制摘要
|
||
"""
|
||
return hashlib.sha256(token.encode('utf-8')).hexdigest()
|
||
|
||
|
||
def add_token_to_blacklist(token: str) -> bool:
|
||
"""
|
||
将Token加入黑名单
|
||
|
||
退出登录时调用，使Token立即失效。黑名单条目的TTL与Token剩余有效期一致，
|
||
Token自然过期后黑名单条目自动清除，避免Redis内存无限增长。
|
||
|
||
Args:
|
||
token: JWT令牌字符串
|
||
|
||
Returns:
|
||
bool: 是否成功加入黑名单
|
||
"""
|
||
try:
|
||
from django_redis import get_redis_connection
|
||
redis = get_redis_connection("default")
|
||
|
||
# 获取Token剩余有效期
|
||
expiration = get_token_expiration(token)
|
||
if expiration is None:
|
||
# 无法解析过期时间，使用默认过期时间作为TTL
|
||
ttl_seconds = DEFAULT_TIMEOUT_MINUTES * 60
|
||
else:
|
||
ttl_seconds = int((expiration - datetime.datetime.now(tz=datetime.timezone.utc)).total_seconds())
|
||
if ttl_seconds <= 0:
|
||
# Token已自然过期，无需加入黑名单
|
||
return True
|
||
|
||
fingerprint = _token_fingerprint(token)
|
||
key = f'{_TOKEN_BLACKLIST_PREFIX}{fingerprint}'
|
||
redis.set(key, '1', ex=ttl_seconds)
|
||
|
||
logger.info(f"Token已加入黑名单，TTL={ttl_seconds}秒")
|
||
return True
|
||
|
||
except Exception as e:
|
||
logger.error(f"Token加入黑名单失败: {str(e)}")
|
||
return False
|
||
|
||
|
||
def is_token_blacklisted(token: str) -> bool:
|
||
"""
|
||
检查Token是否在黑名单中
|
||
|
||
在登录验证中间件中调用，拒绝已被注销的Token。
|
||
|
||
Args:
|
||
token: JWT令牌字符串
|
||
|
||
Returns:
|
||
bool: True-已被注销（在黑名单中），False-有效
|
||
"""
|
||
try:
|
||
from django_redis import get_redis_connection
|
||
redis = get_redis_connection("default")
|
||
|
||
fingerprint = _token_fingerprint(token)
|
||
key = f'{_TOKEN_BLACKLIST_PREFIX}{fingerprint}'
|
||
return redis.exists(key) > 0
|
||
|
||
except Exception as e:
|
||
# fail-open 权衡：Redis 不可用时放行，避免全部请求被拒；代价是已注销的令牌可能被放行。
|
||
# 属安全降级，需告警；节流记录避免 Redis 故障期间日志刷屏。
|
||
_log_blacklist_check_alert(e)
|
||
return False
|
||
|
||
|
||
def check_token_blacklist(token: str) -> bool:
|
||
"""
|
||
检查Token是否在黑名单中（用于中间件）
|
||
|
||
Args:
|
||
token: JWT令牌字符串
|
||
|
||
Returns:
|
||
bool: True-在黑名单中（应拒绝），False-不在黑名单中（可继续验证）
|
||
"""
|
||
return is_token_blacklisted(token)
|