refactor: P2 core+common 建设(应用工厂/扩展/工具层迁入 src/)

【结构】
- src/core/:database/response/jwt/password/security/redis_client/mail/scheduler
- src/common/:base_model/base_db/forms(base_form+fields)/constants/utils
- src/models_registry.py:模型聚合器(Flask-Migrate 全量表识别)

【绞杀者 shim】extensions/、utils/、application/constants|forms|models 打兼容转发
【兼容修复】应用工厂抽取至 application/factory.py(P2 前置,P1 已完成)

【基线修复】application/models(25) + middleware(7) 原被嵌套 .gitignore(*)
误排除出版本库,已补入(base_model/base_db 现为 shim,原版在 src/common/)

【登记既有缺陷】
1. flask db CLI 因 .env FLASK_VERSION 与 --version 冲突报错 → Makefile 已加 FLASK_SKIP_DOTENV=1
2. 本机 Redis 未设密码但 REDIS_AUTH=True,/captcha 报 AUTH 错误(环境问题)

【验收】22 模型表全量识别;src/ flake8 零告警;pytest 5 passed;应用启动;
/login 表单校验、/user/page 鉴权拦截冒烟正常

Co-Authored-By: Claude <noreply@anthropic.com>
This commit is contained in:
DjangoAdmin Dev
2026-08-07 13:20:46 +08:00
co-authored by Claude
parent 92130ef070
commit 658ee6d596
96 changed files with 9926 additions and 4153 deletions
+4 -630
View File
@@ -1,634 +1,8 @@
# +======================================================================
# | DjangoAdmin 企业数字化开发平台 [ 赋能开发者,助力企业发展 ]
# +----------------------------------------------------------------------
# | Copyright (c) 2021-2026 珊瑚云团队 版权所有
# +----------------------------------------------------------------------
# | 开源协议: Apache-2.0
# | 重要声明: 未经书面许可,不得移除或修改本框架相关版权标识。
# +----------------------------------------------------------------------
# | 官方网站: https://www.djangoadmin.cn
# +----------------------------------------------------------------------
# | 研发团队: 珊瑚云团队 荣誉出品
# +======================================================================
# | 版权和免责声明:
# | 一、知识产权声明
# | 本团队对该软件框架产品(以下简称“本产品”)拥有完整知识产权,包括但不限于商标权、专利权、
# | 著作权、商业秘密等。前述权利均受中华人民共和国及相关国际法律法规保护。未经本团队书面授权,
# | 任何个人或单位不得基于本产品本身申请任何形式的知识产权。
# | 二、使用限制与禁止条款
# | 您理解并同意,禁止将本产品用于以下任何行为:
# | 1. 任何违反中华人民共和国法律法规的用途;
# | 2. 任何侵害他人合法权益(包括但不限于人身权、财产权、知识产权等)的行为;
# | 3. 任何危害国家安全、公共利益及社会道德风尚的行为。
# | 三、责任豁免
# | 因使用本产品而产生的任何损失或法律纠纷(包括但不限于直接损失、间接损失、附带损失、衍生损失
# | 以及因违约、侵权等引发的责任),本团队均不承担任何法律责任。若因违规使用本产品导致任何第三
# | 方提起索赔或行政机关进行调查,您应独立承担全部责任。
# | 四、违约与法律责任
# | 如发现任何违反本声明的行为,本团队有权:
# | 1. 无条件配合公安机关及其他有权机关的调查取证工作;
# | 2. 依法追究相关方的民事责任,并要求赔偿全部损失;
# | 3. 保留进一步采取刑事控告等法律措施的权利。
# | 五、许可范围说明
# | 1、被授权方应视本产品源代码为核心机密信息,负有最高标准的保密与保管义务。禁止以任何形式(包
# | 括但不限于直接或间接的分享、复制、传播、分发、转售、出租、开源、反向工程、反编译、反汇编或通
# | 过网络服务提供给任何第三方)泄露、披露或处置源代码的全部或部分内容。任何未经授权的复制、使用、
# | 传播行为均构成对本团队知识产权的严重侵权,本团队有权依法追究其全部法律责任,并要求赔偿因此造
# | 成的一切经济损失及商誉损害。
# | 2、本产品仅限于公司或个人在法律允许范围内,用于内部且合规的软件产品研发。具体权利义务以《软件
# | 许可协议》附件为准。
# +======================================================================
"""
JWT令牌处理模块
兼容 shim(绞杀者迁移期间保留)
=============================================================
该模块提供基于JWT(JSON Web Token)的令牌生成和验证功能。
主要用于用户登录认证,生成访问令牌,并在后续请求中验证令牌的有效性。
支持令牌过期机制,确保系统的安全性。
主要功能:
1. JWT令牌生成(create_token)- 用于用户登录成功后生成访问令牌
2. JWT令牌解析验证(parse_payload)- 验证请求中的令牌有效性
3. 统一的异常处理,返回标准格式的验证结果
4. 可配置的令牌过期时间
所属包: utils
所在文件: utils/jwts.py
依赖模块: jwt - JWT处理第三方库
datetime - 时间处理模块
os - 环境变量读取模块
secrets - 安全随机数生成模块
JWT 工具已迁移至 src/core/jwt.py。
本文件仅作兼容转发,最终在 P7(切尾)阶段删除。
"""
import datetime
import hashlib
import logging
import jwt
from flask import request
from jwt import exceptions
from config import JWT_SALT, JWT_EXPIRE_MINUTES
# 配置日志记录器
logger = logging.getLogger(__name__)
# =============================================================
# JWT密钥配置
# =============================================================
# 从 config.auth 统一读取,密钥配置在 .env 文件中管理
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:
raise RuntimeError(
f"JWT_SALT 长度不足:{key_bytes} 字节,至少需要 32 字节(256位)。\n"
"生成方式:python -c \"import secrets; print(secrets.token_urlsafe(48))\""
)
logger.info(f"JWT密钥校验通过,长度:{key_bytes}字节")
# 启动时校验密钥强度
_validate_key_strength(JWT_SALT)
# JWT算法配置
JWT_ALGORITHM = 'HS256'
# 默认令牌过期时间(分钟),从配置读取
DEFAULT_TIMEOUT_MINUTES = JWT_EXPIRE_MINUTES
# =============================================================
# 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
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():
"""
从请求头中提取 access_token
=============================================================
从 HTTP 请求的 Authorization 头中提取 JWT token,返回纯净的 token 字符串。
统一处理 token 提取逻辑,便于在其他模块中复用。
参数:
request: Django 请求对象
返回:
tuple: (success, token, message)
- success (bool): 是否成功获取 token
- token (str/None): token 字符串,失败时为 None
- message (str): 提示信息
使用示例:
>>> success, token, msg = get_access_token()
>>> if success:
... # 使用 token 进行后续处理
... parse_result = parse_payload(token)
... else:
... # 处理错误
... return R.failed(msg)
"""
try:
# 从请求头中获取 token 值
# 说明:Authorization 头格式通常为 "Bearer <token>"
auth_header = request.headers.get('Authorization', '')
if not auth_header:
return False, None, '缺少 Authorization 头'
# 处理 Bearer 前缀
# 说明:移除 "Bearer " 前缀并去除首尾空格,获取纯净 token
access_token = auth_header.replace('Bearer ', "").strip()
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)}'
# token解析工具类
def parse_token(request):
"""
解析请求头中的token,返回解析结果
=============================================================
从HTTP请求的Authorization头中提取JWT token,进行解析验证,
并返回包含用户信息的结构化数据。统一处理各种异常情况。
参数:
request: Django请求对象
返回:
dict: 包含解析结果和用户信息的字典
{
'code': 0, # 0-成功,-1-失败
'msg': 'success', # 提示信息
'data': {
'userId': 0, # 用户ID
'username': '', # 用户名
'realname': '' # 真实姓名
}
}
使用示例:
>>> result = parse_token(request)
>>> if result['code'] == 0:
... user_id = result['data']['userId']
... username = result['data']['username']
"""
# 初始化返回结果
# 说明:设置默认失败状态,确保返回结构完整性
result = {
'code': -1,
'msg': 'token无效',
'data': {
'userId': 0,
'userName': None
}
}
try:
# 获取 access_token
success, access_token, msg = get_access_token()
if not success:
result['msg'] = msg
return result
# JWT解密
# 函数:parse_payload(access_token)
# 说明:调用JWT解析工具验证token有效性
parse_result = parse_payload(access_token)
# 验证解析结果
# 说明:确保解析结果格式正确
if not isinstance(parse_result, dict):
result['msg'] = '解析结果格式错误'
return result
# 结果标识
# 说明:code为0表示验证通过
code = parse_result.get('code', -1)
if code != 0:
result['msg'] = parse_result.get('msg', 'token验证失败')
return result
# 获取数据部分
# 说明:提取token载荷中的用户数据
data = parse_result.get('data', {})
if not isinstance(data, dict):
result['msg'] = '数据格式错误'
return result
# 提取用户信息
# 说明:从载荷中获取具体用户字段
user_id = data.get('userId', 0)
username = data.get('username', '')
realname = data.get('realname', '')
# 更新返回结果
# 说明:设置成功状态并填充用户信息
result['code'] = 0
result['msg'] = 'success'
result['data']['userId'] = user_id
result['data']['username'] = username
result['data']['realname'] = realname
except Exception as e:
# 异常捕获
# 说明:防止未处理的异常导致系统崩溃
result['msg'] = f'token解析异常: {str(e)}'
# 可以在这里记录日志
# logger.error(f"token解析异常: {str(e)}")
return result
# =============================================================
# Token黑名单功能
# =============================================================
# 黑名单Key前缀,用于Redis中存储被注销的Token
_TOKEN_BLACKLIST_PREFIX = 'token:blacklist:'
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 utils.redis_client import redis
# 获取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', expire=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 utils.redis_client import redis
fingerprint = _token_fingerprint(token)
key = f'{_TOKEN_BLACKLIST_PREFIX}{fingerprint}'
return redis.exists(key) > 0
except Exception as e:
logger.error(f"Token黑名单检查异常: {str(e)}")
# Redis不可用时放行,避免阻断所有请求
return False
from src.core.jwt import * # noqa: F401,F403