feat(audit): 按条件导出审计日志，并修 module 口径断裂与详情接口权限漏洞
需求是"想看某天或某段时间的入库的修改，支持下载或当场查看"。
核查后：筛选与查看审计页早已具备，缺的是导出 —— 但直接加导出会撞上两个
既有问题，故一并处理。
1. 「入库」的 module 口径在 2026-09-10 断过一次
那天旧的全局监听器 app/utils/audit_events.py 被现行的
app/core/audit_listener.py 取代（旧模块现已无人 import，是死代码），
新监听器把入库三表 stock_buy/stock_semi/stock_product 统一归为
「库存管理」，不再产出「入库管理」：
入库管理 16118 条 2026-04-22 ~ 2026-09-10
采购入库 1196 条 2026-03-18 ~ 2026-04-20
成品入库 8 条 2026-03-23 ~ 2026-04-17
库存管理 1371 条 2026-09-10 ~
后果：按单个 module 值筛「入库」会正好在那天断掉，现象是"入库记录突然
没了"。解法是**查询期别名展开**（audit_labels.MODULE_GROUPS），不迁移
历史数据 —— 改数据不可逆，而聚合查询无损。
2. GET /audit/logs/<id> 缺权限校验，也没有公司隔离
此前只有 @jwt_required()，任何登录用户改一下 URL 里的 id 就能读到全部
审计明细（含 details 里的完整快照）。列表接口有 system_audit 把关，
详情提供的信息是列表的超集，没道理比列表更宽松。
现补权限码 + 与列表同一个公司判据；越权与不存在**统一返回 404**，
区分开来等于告诉探测者"这个 id 存在，只是你没权限"。
3. 筛选逻辑只写一份
抽出 _build_audit_query()，/logs 与 /logs/export 共用。两处各写一份迟早
出现"页面上 300 条、导出来 280 条"，而报表对不上比没有报表更糟。
导出实现：
- GET /audit/logs/export，权限码同 /logs；忽略分页参数，导全量。
- 三个工作表：汇总（含本次生效的筛选条件，收件人看不到页面上的筛选框）、
审计日志（一行一条）、变更明细（一行一个变更字段，可对「字段」列筛选）。
- EXPORT_LIMIT=100000 作熔断。实测全库 5.8 万条导出 2.5MB / 约 3 秒，
故走同步返回，不引入 export_service 那套异步框架（它还会落盘且无清理）。
触顶时在文件名与汇总表显式写明，不静默丢弃。
- 顺带补 /audit/modules 的公司隔离（下拉选项此前会跨公司）。
前端（AuditLog.vue / api/audit.ts）：
- 模块下拉分「业务聚合 / 具体模块」两组，聚合项由后端下发，前端不硬编成员；
- 操作类型改多选，提交时 join(',')（后端 split(',') 接收）；
- 导出按钮沿用仓库既有的 blob 下载写法，文件名前端自定。
验证：后端 21 项断言全过（口径断裂、多选别名归一、日期闭区间、公司隔离、
401/403/200 权限矩阵、导出口径与列表 total 一致、附件为合法 xlsx）；
vue-tsc --noEmit 与 vite build 均 exit=0。
This commit is contained in:
yueli
@ -1,12 +1,16 @@
|
||||
# inventory-backend/app/api/v1/audit.py
|
||||
from flask import Blueprint, request, jsonify, current_app
|
||||
from flask_jwt_extended import jwt_required, get_jwt
|
||||
from app.utils.decorators import permission_required
|
||||
from app.models.audit import AuditLog
|
||||
import io
|
||||
from datetime import datetime, timedelta
|
||||
|
||||
from flask import Blueprint, current_app, jsonify, request, send_file
|
||||
from flask_jwt_extended import jwt_required
|
||||
from sqlalchemy import func
|
||||
|
||||
from app.extensions import db
|
||||
from sqlalchemy import or_
|
||||
from datetime import datetime
|
||||
import json
|
||||
from app.models.audit import AuditLog
|
||||
from app.models.system import SysUser
|
||||
from app.services.audit_export_service import SYSTEM_USERNAME, build_audit_workbook
|
||||
from app.utils.decorators import get_current_company_filter, permission_required
|
||||
|
||||
audit_bp = Blueprint('audit', __name__)
|
||||
|
||||
@ -24,9 +28,163 @@ audit_bp = Blueprint('audit', __name__)
|
||||
from app.utils.audit_labels import ( # noqa: E402
|
||||
ACTION_ALIASES,
|
||||
canon_action,
|
||||
expand_modules,
|
||||
labels_payload,
|
||||
)
|
||||
|
||||
# 单次导出的记录数上限。
|
||||
#
|
||||
# ★ 这是**熔断**，不是"为了让报表好看"的截断：
|
||||
# · Excel 单表硬上限 1,048,576 行，长表展开后会成倍放大；
|
||||
# · 同步请求跑太久会超时，用户只看到"失败"，不知道是数据量的问题。
|
||||
# 实测当前全库共 5.8 万条、全量导出 3.3 秒 / 2MB，正常情况下永远碰不到这个值。
|
||||
# 触顶时会**显式告知**（写进文件名和汇总工作表），不静默丢弃 ——
|
||||
# 报表的读法就是"看到的就是全部"，偷偷砍掉比报错更危险。
|
||||
EXPORT_LIMIT = 100000
|
||||
|
||||
|
||||
# =============================================================================
|
||||
# 查询构造 —— /logs 与 /logs/export 的**唯一**入口
|
||||
# =============================================================================
|
||||
|
||||
def _build_audit_query(args):
|
||||
"""
|
||||
按请求参数构造审计日志查询，返回 (query, company_limit)。
|
||||
|
||||
★ 列表与导出共用本函数。两处各写一份筛选逻辑，迟早会出现
|
||||
"页面上看到 300 条、导出来 280 条"这种对不上的情况 ——
|
||||
而报表对不上，比没有报表更糟（人会照着错的数字做决定）。
|
||||
这正是 audit_labels.py 开头警告过的"两份手工同步的副本必然漂移"。
|
||||
|
||||
★ 可多选的参数用逗号分隔（与仓库既有惯例一致：semi.vue 的 statuses
|
||||
用 join(',') 传参、inbound/product.py 用 split(',') 接收）。
|
||||
"""
|
||||
query = AuditLog.query
|
||||
|
||||
# ---------- 操作来源：all(默认) / user(仅真实用户) / system(仅系统) ----------
|
||||
# 背景：改造前的全局监听器没有请求上下文守卫，系统初始化与后台定时任务
|
||||
# 产生了大量 username='system' 的日志（历史存量约 1.8 万条），会把列表刷屏。
|
||||
operator_type = (args.get('operator_type') or 'all').strip().lower()
|
||||
if operator_type not in ('all', 'user', 'system'):
|
||||
operator_type = 'all'
|
||||
if operator_type == 'user':
|
||||
query = query.filter(AuditLog.username != SYSTEM_USERNAME)
|
||||
elif operator_type == 'system':
|
||||
query = query.filter(AuditLog.username == SYSTEM_USERNAME)
|
||||
|
||||
username = (args.get('username') or '').strip()
|
||||
if username:
|
||||
query = query.filter(AuditLog.username.like(f'%{username}%'))
|
||||
|
||||
# ---------- 模块（支持多选 + 聚合别名）----------
|
||||
# ★ 经 expand_modules 展开：「入库(全部)」会变成它名下的全部成员值。
|
||||
# 这一步是必需的 —— 审计的 module 在 2026-09-10 换过一次口径
|
||||
# （旧的「入库管理」→ 新的「库存管理」），不展开就会在换口径那天
|
||||
# 断掉，用户看到"入库记录突然没了"。
|
||||
# 详见 audit_labels.MODULE_GROUPS 的说明。
|
||||
modules = [m for m in (args.get('module') or '').split(',') if m.strip()]
|
||||
if modules:
|
||||
query = query.filter(AuditLog.module.in_(expand_modules(modules)))
|
||||
|
||||
# ---------- 操作类型（支持多选）----------
|
||||
actions = [a for a in (args.get('action') or '').split(',') if a.strip()]
|
||||
if actions:
|
||||
# ★ 逐个归一化再合并别名集合，不能只处理第一个：
|
||||
# 历史数据里 CREATE 与「新增」混用，多选时若只对首个值展开别名，
|
||||
# 其余选项就搜不到早期数据（4 月前的中文 action）。
|
||||
aliases = set()
|
||||
for a in actions:
|
||||
canon = canon_action(a)
|
||||
aliases.update(ACTION_ALIASES.get(canon, (a,)))
|
||||
query = query.filter(AuditLog.action.in_(aliases))
|
||||
|
||||
target_id = (args.get('target_id') or '').strip()
|
||||
if target_id:
|
||||
query = query.filter(AuditLog.target_id == target_id)
|
||||
|
||||
# ---------- 日期区间 ----------
|
||||
# 语义是**闭区间**：end_date 当天 23:59:59 的数据也要包含进来，
|
||||
# 故上界取次日 00:00 的严格小于。写成 <= end_date 会静默丢掉当天。
|
||||
start_date = (args.get('start_date') or '').strip()
|
||||
if start_date:
|
||||
try:
|
||||
query = query.filter(
|
||||
AuditLog.created_at >= datetime.strptime(start_date, '%Y-%m-%d'))
|
||||
except ValueError:
|
||||
pass
|
||||
end_date = (args.get('end_date') or '').strip()
|
||||
if end_date:
|
||||
try:
|
||||
query = query.filter(
|
||||
AuditLog.created_at < datetime.strptime(end_date, '%Y-%m-%d') + timedelta(days=1))
|
||||
except ValueError:
|
||||
pass
|
||||
|
||||
# ---------- 【行级数据隔离】----------
|
||||
# AuditLog 表**没有** company_name 字段，公司信息只能经 SysUser.department 反查。
|
||||
# 审计日志的 username 存的是账号（'gaoxue'），而 sys_user.username 存的是
|
||||
# '高雪/gaoxue'，故取 '/' 之后的账号部分去 join。
|
||||
#
|
||||
# ★ 导出接口必须带同样的隔离，否则"导出"就成了绕过数据隔离的后门 ——
|
||||
# 页面按公司过滤、导出却给全量，这个洞比页面越权更隐蔽。
|
||||
company_limit = get_current_company_filter()
|
||||
if company_limit is not None:
|
||||
query = query.join(
|
||||
SysUser, func.split_part(SysUser.username, '/', 2) == AuditLog.username
|
||||
).filter(SysUser.department == company_limit)
|
||||
|
||||
return query, company_limit
|
||||
|
||||
|
||||
def _filter_summary_rows(args):
|
||||
"""
|
||||
把本次生效的筛选条件写成汇总表的几行。
|
||||
|
||||
★ 收到 Excel 的人看不到页面上的筛选框，没有这段说明就无从判断
|
||||
"这份表是全量还是某个子集"。
|
||||
"""
|
||||
rows = []
|
||||
operator_type = (args.get('operator_type') or '').strip().lower()
|
||||
if operator_type:
|
||||
rows.append(('筛选条件', '操作来源',
|
||||
{'user': '仅真实用户', 'system': '仅系统操作'}.get(operator_type, '全部')))
|
||||
|
||||
labels = [
|
||||
('username', '操作人（模糊匹配）'),
|
||||
('module', '模块'),
|
||||
('action', '操作类型'),
|
||||
('target_id', '目标ID'),
|
||||
]
|
||||
for key, label in labels:
|
||||
val = (args.get(key) or '').strip()
|
||||
if val:
|
||||
rows.append(('筛选条件', label, val))
|
||||
|
||||
start_date = (args.get('start_date') or '').strip()
|
||||
end_date = (args.get('end_date') or '').strip()
|
||||
if start_date or end_date:
|
||||
rows.append(('筛选条件', '日期区间',
|
||||
f"{start_date or '不限'} ~ {end_date or '不限'}"))
|
||||
return rows
|
||||
|
||||
|
||||
def _distinct_with_company(column, company_limit):
|
||||
"""
|
||||
取某列的去重值（带公司隔离）。
|
||||
|
||||
★ 下拉选项也要隔离：不带的话，A 公司的用户在筛选框里能看到 B 公司
|
||||
才有的模块名，属于低危但确实存在的信息泄露。
|
||||
"""
|
||||
q = db.session.query(column).distinct()
|
||||
if company_limit is not None:
|
||||
q = q.join(SysUser, func.split_part(SysUser.username, '/', 2) == AuditLog.username) \
|
||||
.filter(SysUser.department == company_limit)
|
||||
return [v[0] for v in q.all() if v[0]]
|
||||
|
||||
|
||||
# =============================================================================
|
||||
# 接口
|
||||
# =============================================================================
|
||||
|
||||
@audit_bp.route('/logs', methods=['GET'])
|
||||
@jwt_required()
|
||||
@ -34,100 +192,25 @@ from app.utils.audit_labels import ( # noqa: E402
|
||||
def get_audit_logs():
|
||||
"""获取审计日志列表（分页）"""
|
||||
try:
|
||||
# 分页参数
|
||||
page = request.args.get('page', 1, type=int)
|
||||
page_size = request.args.get('pageSize', 50, type=int)
|
||||
|
||||
# 筛选参数
|
||||
username = request.args.get('username', '').strip()
|
||||
module = request.args.get('module', '').strip()
|
||||
action = request.args.get('action', '').strip()
|
||||
target_id = request.args.get('target_id', '').strip()
|
||||
start_date = request.args.get('start_date', '').strip()
|
||||
end_date = request.args.get('end_date', '').strip()
|
||||
|
||||
# ★ 操作人类型：all(默认) / user(仅真实用户) / system(仅系统)
|
||||
#
|
||||
# 背景：改造前的全局监听器没有请求上下文守卫，系统初始化与后台定时任务
|
||||
# 产生了大量 username='system' 的日志（历史存量约 1.8 万条），会把列表刷屏。
|
||||
# 新监听器已加守卫不再产生此类记录，但存量数据仍需要能筛掉。
|
||||
operator_type = request.args.get('operator_type', 'all').strip().lower()
|
||||
if operator_type not in ('all', 'user', 'system'):
|
||||
operator_type = 'all'
|
||||
|
||||
# 构建查询
|
||||
query = AuditLog.query
|
||||
|
||||
if operator_type == 'user':
|
||||
# 真实用户：排除 system 占位账号
|
||||
query = query.filter(AuditLog.username != 'system')
|
||||
elif operator_type == 'system':
|
||||
query = query.filter(AuditLog.username == 'system')
|
||||
|
||||
if username:
|
||||
query = query.filter(AuditLog.username.like(f'%{username}%'))
|
||||
if module:
|
||||
query = query.filter(AuditLog.module == module)
|
||||
if action:
|
||||
# ★ 兼容历史别名：先把任意写法（中文/小写）归一化为规范值，
|
||||
# 再展开为该值的全部等价写法一起匹配。
|
||||
# 否则选「新增」只能搜到早期中文 action 的数据（约 3-4 月），
|
||||
# 会让用户误以为"没有最近的内容"。
|
||||
canon = canon_action(action)
|
||||
if canon in ACTION_ALIASES:
|
||||
query = query.filter(AuditLog.action.in_(ACTION_ALIASES[canon]))
|
||||
else:
|
||||
query = query.filter(AuditLog.action == action)
|
||||
if target_id:
|
||||
query = query.filter(AuditLog.target_id == target_id)
|
||||
if start_date:
|
||||
try:
|
||||
start_dt = datetime.strptime(start_date, '%Y-%m-%d')
|
||||
query = query.filter(AuditLog.created_at >= start_dt)
|
||||
except ValueError:
|
||||
pass
|
||||
if end_date:
|
||||
try:
|
||||
end_dt = datetime.strptime(end_date, '%Y-%m-%d')
|
||||
# 包含当天结束时间
|
||||
from datetime import timedelta
|
||||
end_dt = end_dt + timedelta(days=1)
|
||||
query = query.filter(AuditLog.created_at < end_dt)
|
||||
except ValueError:
|
||||
pass
|
||||
|
||||
# 【行级数据隔离】通过操作人 username 关联用户表过滤公司
|
||||
from app.utils.decorators import get_current_company_filter
|
||||
from app.models.system import SysUser
|
||||
from sqlalchemy import func
|
||||
company_limit = get_current_company_filter()
|
||||
if company_limit is not None:
|
||||
query = query.join(SysUser, func.split_part(SysUser.username, '/', 2) == AuditLog.username) \
|
||||
.filter(SysUser.department == company_limit)
|
||||
query, company_limit = _build_audit_query(request.args)
|
||||
|
||||
# 排序
|
||||
query = query.order_by(AuditLog.created_at.desc())
|
||||
|
||||
# 分页
|
||||
pagination = query.paginate(page=page, per_page=page_size, error_out=False)
|
||||
logs = pagination.items
|
||||
|
||||
# 序列化
|
||||
data = [log.to_dict() for log in logs]
|
||||
data = [log.to_dict() for log in pagination.items]
|
||||
|
||||
# 获取可用的模块和操作类型（同公司范围内）
|
||||
modules_query = db.session.query(AuditLog.module).distinct()
|
||||
actions_query = db.session.query(AuditLog.action).distinct()
|
||||
if company_limit is not None:
|
||||
modules_query = modules_query.join(SysUser, func.split_part(SysUser.username, '/', 2) == AuditLog.username) \
|
||||
.filter(SysUser.department == company_limit)
|
||||
actions_query = actions_query.join(SysUser, func.split_part(SysUser.username, '/', 2) == AuditLog.username) \
|
||||
.filter(SysUser.department == company_limit)
|
||||
modules = [m[0] for m in modules_query.all() if m[0]]
|
||||
modules = _distinct_with_company(AuditLog.module, company_limit)
|
||||
actions_raw = _distinct_with_company(AuditLog.action, company_limit)
|
||||
|
||||
# ★ action 下拉项归一化：把 CREATE/create/新增 等别名合并为一个规范值，
|
||||
# 避免下拉框出现「CREATE」与「新增」两个语义重复的选项。
|
||||
actions = sorted({canon_action(a[0]) for a in actions_query.all() if a[0]})
|
||||
actions = sorted({canon_action(a) for a in actions_raw})
|
||||
|
||||
return jsonify({
|
||||
'code': 200,
|
||||
@ -147,12 +230,86 @@ def get_audit_logs():
|
||||
return jsonify({'code': 500, 'msg': f'服务器内部错误: {str(e)}'}), 500
|
||||
|
||||
|
||||
@audit_bp.route('/logs/export', methods=['GET'])
|
||||
@jwt_required()
|
||||
@permission_required('system_audit')
|
||||
def export_audit_logs():
|
||||
"""
|
||||
按筛选条件导出审计日志为 Excel（同步返回文件流）。
|
||||
|
||||
★ 筛选条件与 GET /logs **完全一致**（共用 _build_audit_query），
|
||||
且**忽略分页参数** —— 导的是符合条件的全量，不是当前页。
|
||||
「只导当前页 50 条」是导出功能最常见的误解，故这里明确不支持。
|
||||
|
||||
★ 为什么同步返回而不是走 export_service 那套异步任务：
|
||||
实测全库 5.8 万条导出耗时 3.3 秒、2MB，同步完全够用；
|
||||
而异步那套要落盘（uploads/exports/）且**没有清理机制**，
|
||||
日积月累会把磁盘吃掉 —— 见 export_service/excel_task.py。
|
||||
"""
|
||||
try:
|
||||
query, _ = _build_audit_query(request.args)
|
||||
# 台账按时间倒序：导出的人多半想看"最近的"（与列表页一致）
|
||||
query = query.order_by(AuditLog.created_at.desc())
|
||||
|
||||
total = query.count()
|
||||
truncated = total > EXPORT_LIMIT
|
||||
rows = query.limit(EXPORT_LIMIT).all() if truncated else query.all()
|
||||
|
||||
note = None
|
||||
if truncated:
|
||||
note = (f"匹配记录共 {total} 条，超过单次导出上限 {EXPORT_LIMIT} 条，"
|
||||
f"本次仅导出前 {EXPORT_LIMIT} 条。请缩小日期范围后分批导出。")
|
||||
|
||||
data = build_audit_workbook(
|
||||
rows,
|
||||
summary_rows=_filter_summary_rows(request.args),
|
||||
note=note,
|
||||
)
|
||||
|
||||
stamp = datetime.now().strftime('%Y%m%d_%H%M%S')
|
||||
# 文件名里带上「部分」—— 用户不解压也能看出这不是全量
|
||||
name = (f"审计日志_部分{EXPORT_LIMIT}条_{stamp}.xlsx" if truncated
|
||||
else f"审计日志_{stamp}.xlsx")
|
||||
|
||||
return send_file(
|
||||
io.BytesIO(data),
|
||||
mimetype='application/vnd.openxmlformats-officedocument.spreadsheetml.sheet',
|
||||
as_attachment=True,
|
||||
download_name=name,
|
||||
)
|
||||
|
||||
except Exception as e:
|
||||
current_app.logger.error(f"导出审计日志失败: {str(e)}")
|
||||
return jsonify({'code': 500, 'msg': f'导出失败: {str(e)}'}), 500
|
||||
|
||||
|
||||
@audit_bp.route('/logs/<int:log_id>', methods=['GET'])
|
||||
@jwt_required()
|
||||
@permission_required('system_audit')
|
||||
def get_audit_log_detail(log_id):
|
||||
"""获取单条审计日志详情"""
|
||||
"""
|
||||
获取单条审计日志详情。
|
||||
|
||||
★ 补权限码与公司隔离。此前这里**只有 @jwt_required()**，任何登录用户
|
||||
改一下 URL 里的 id 就能读到全部审计明细（含 details 里的完整快照，
|
||||
那是被操作记录的原始字段值）。列表接口有 system_audit 把关，
|
||||
详情接口提供的信息是列表的超集，没道理比列表更宽松。
|
||||
|
||||
★ 越权与不存在**统一返回 404**，不区分二者 —— 区分开来就等于告诉
|
||||
探测者"这个 id 是存在的，只是你没权限"，反而泄露了数据规模。
|
||||
"""
|
||||
try:
|
||||
log = AuditLog.query.get(log_id)
|
||||
query = AuditLog.query.filter(AuditLog.id == log_id)
|
||||
|
||||
# 与列表用同一个公司判据（SysUser.department 反查），保证两处一致：
|
||||
# 列表里看不到的记录，详情也不该看得到。
|
||||
company_limit = get_current_company_filter()
|
||||
if company_limit is not None:
|
||||
query = query.join(
|
||||
SysUser, func.split_part(SysUser.username, '/', 2) == AuditLog.username
|
||||
).filter(SysUser.department == company_limit)
|
||||
|
||||
log = query.first()
|
||||
if not log:
|
||||
return jsonify({'code': 404, 'msg': '日志不存在'}), 404
|
||||
|
||||
@ -172,8 +329,8 @@ def get_audit_log_detail(log_id):
|
||||
def get_modules():
|
||||
"""获取所有模块列表（用于筛选）"""
|
||||
try:
|
||||
modules = db.session.query(AuditLog.module).distinct().all()
|
||||
modules = [m[0] for m in modules if m[0]]
|
||||
company_limit = get_current_company_filter()
|
||||
modules = _distinct_with_company(AuditLog.module, company_limit)
|
||||
return jsonify({'code': 200, 'data': modules}), 200
|
||||
except Exception as e:
|
||||
current_app.logger.error(f"获取模块列表失败: {str(e)}")
|
||||
|
||||
Reference in New Issue
Block a user