refactor(audit): 审计架构清理——复活白名单监听器、停用噪声监听器、清除僵尸装饰器

一、统一为单一监听器实现
  原先两套 SQLAlchemy 事件监听器并存:
    · app/utils/audit_events.py   —— 全局监听 db.Model、无白名单、无请求上下文守卫(实际在跑)
    · app/core/audit_listener.py  —— 白名单制、有守卫、有模型级开关(从未生效)
  后者失效的根因:注册代码写在 extensions.py 的 init_extensions() 内,
  而该函数全仓库只有定义、没有任何调用(create_app 直接内联调用 db.init_app 等)。

  现统一由 app/core/audit_listener.py 承担,并在 create_app() 中显式注册。
  extensions.py 的死函数 init_extensions 整体删除,避免后人误以为它是有效入口。

二、修复监听器三处致命缺陷(此前注册了也写不进数据)
  1. 事件回调第二个参数是 Connection,原代码却调用 Connection.add()(不存在),
     每次写日志都抛 AttributeError 并被 except 吞掉 → 改为 connection.execute()
  2. register_audit_listeners 从 app.models 批量 import 多个未导出的模型,
     ImportError 被上层 try/except 吞掉 → 改为按表名从 db.metadata 取模型
  3. 本项目有 31 处函数体内延迟导入模型(如 scrap.py 内部才 import ScrapApproval),
     一次性注册会静默漏表 → 增加 ensure_audit_listeners() 惰性补绑,
     并在模型预加载段补全审批单/BOM/采购等模型

三、强约束
  · WHITELIST_TABLES:仅 18 张核心业务表,系统表/草稿表/向量表不再自审
  · has_request_context() 守卫:系统初始化与后台定时任务不再产生 username=system 噪声
  · IGNORE_FIELDS 增加 password/password_hash/salt/token/secret/api_key(安全红线)
  · created_at 显式写 beijing_time(),与全系统时间口径一致

四、清除僵尸装饰器
  @audit_log 早已退化为直接透传的空壳(module/action 参数全被忽略,
  数据库中零星的中文 action 即其历史遗留产物),却仍挂在 38 处路由上。
  连同 13 个文件的 import 一并移除;audit_events.register_audit_events 改为空操作。

验证:应用上下文中的写操作不产生日志;HTTP 请求产生 5 条日志,
对象为业务单号(APR-SCRAP-... / SKU),模块中文,操作人真实,时间为北京时间。
This commit is contained in:
yueli
2026-09-10 14:16:27 +08:00
parent 1ef9ae4ad9
commit 4808a48594
18 changed files with 513 additions and 527 deletions

View File

@ -84,40 +84,12 @@ def revoke_all_tokens_for_user(user_id):
current_app.logger.error(f"Failed to revoke tokens for user_id={user_id}: {e}")
# 2. 定义初始化函数 (供工厂函数 create_app 调用)
def init_extensions(app):
"""
统一初始化所有 Flask 扩展
"""
global redis_client
# 初始化数据库
db.init_app(app)
# 初始化迁移工具
migrate.init_app(app, db)
# 初始化跨域设置 (允许 /api/* 路径被所有来源访问)
cors.init_app(app, resources={r"/api/*": {"origins": "*"}})
# 初始化 JWT (这一步至关重要,缺少它会导致 500 错误)
jwt.init_app(app)
# 初始化 Redis (单设备登录互踢)
redis_url = app.config.get('REDIS_URL', 'redis://localhost:6379/0')
try:
redis_client = redis.from_url(redis_url, decode_responses=True)
redis_client.ping()
app.logger.info("✅ Redis connected successfully")
except Exception as e:
app.logger.warning(f"⚠️ Redis connection failed: {e}, single-device login will be disabled")
# ★ 注册 SQLAlchemy 审计监听器
# 必须在 db.init_app 之后调用,确保所有模型已映射
try:
from app.core.audit_listener import register_audit_listeners
with app.app_context():
count = register_audit_listeners(db)
app.logger.info(f"✅ 审计监听器注册成功,共绑定 {count} 个模型")
except Exception as e:
app.logger.error(f"⚠️ 审计监听器注册失败: {e}")
# 2. [已移除] init_extensions(app)
#
# 该函数历史上用于统一初始化扩展并注册审计监听器,但 create_app() 始终是
# 直接内联调用 db.init_app / migrate.init_app / jwt.init_app / cors.init_app,
# 从未调用过本函数 —— 属于死代码。其内部的审计注册因此也从未执行,
# 造成「两套监听器并存、实际只有未设计的那套在跑」的长期隐患。
#
# 为避免后人误以为它是一个有效的初始化入口,已整体删除;
# 审计监听器现由 create_app() 显式注册(见 app/__init__.py 1.1 节)。