347 lines
10 KiB
Python
347 lines
10 KiB
Python
"""
|
||
操作日志自动记录中间件
|
||
=============================================================
|
||
|
||
通过 after_request 钩子自动记录写操作（POST/PUT/DELETE/PATCH）日志。
|
||
无需在每个视图函数中手动调用 record_operation_log。
|
||
|
||
记录规则：
|
||
- 只记录写操作（POST/PUT/DELETE/PATCH），GET/OPTIONS/HEAD 不记录
|
||
- 只记录成功操作（HTTP 2xx 且业务 code=0）
|
||
- 内部调用（定时任务等）不记录
|
||
- 登录/退出/验证码等路径不记录（由 login_log 单独记录）
|
||
- 日志记录失败不影响主流程
|
||
|
||
标题声明：
|
||
在路由视图上使用 `@operation_log("岗位管理", "新增")` 声明操作日志标题，
|
||
中间件从已注册视图函数读取 `__op_log_title__`，无需集中式模块映射表。
|
||
"""
|
||
|
||
import logging
|
||
import time
|
||
|
||
from flask import current_app, g, request
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
|
||
def operation_log(module_name: str, action_name: str = ""):
|
||
"""
|
||
在路由视图上声明操作日志标题（模块名-动作名）。
|
||
|
||
挂载到视图函数上，中间件从 `current_app.view_functions[endpoint].__op_log_title__` 读取。
|
||
|
||
Args:
|
||
module_name: 模块名称，如 "岗位管理"
|
||
action_name: 操作名称，如 "新增"，为空时仅使用模块名
|
||
|
||
Returns:
|
||
decorator: 装饰器函数
|
||
|
||
Example:
|
||
@position_blue.route("/add", methods=["POST"])
|
||
@operation_log("岗位管理", "新增")
|
||
@check_permission("sys:position:add")
|
||
def add(): ...
|
||
"""
|
||
|
||
def decorator(func):
|
||
func.__op_log_title__ = f"{module_name}-{action_name}" if action_name else module_name
|
||
return func
|
||
|
||
return decorator
|
||
|
||
|
||
# 不记录操作日志的敏感路径尾段（登录/验证码/退出/SMS 验证码）。
|
||
# 按路径尾段 endswith 匹配，天然兼容 /api/v1、/api/v2 等任意版本前缀，
|
||
# 无需随版本升级逐个追加前缀。
|
||
_EXCLUDED_PATH_SEGMENTS = ("/login", "/captcha", "/logout", "/sendSms")
|
||
|
||
|
||
def _is_excluded_path(path) -> bool:
|
||
"""判断路径是否属于敏感端点（尾段匹配，版本无关）"""
|
||
return any(path.endswith(seg) for seg in _EXCLUDED_PATH_SEGMENTS)
|
||
|
||
|
||
# 需脱敏的敏感字段（真正的凭据），防止写入操作日志/审计。
|
||
# 注意：不包含 code/key——业务数据（role.code/dict.code 等）常含同名键，会被误打码；
|
||
# 验证码/短信端点已按路径排除（_is_excluded_path），无需在此兜底。
|
||
_SENSITIVE_KEYS = frozenset({
|
||
"password", "salt", "oldPassword", "newPassword", "confirmPassword",
|
||
"accessToken", "token",
|
||
})
|
||
|
||
|
||
def _redact_sensitive(data):
|
||
"""
|
||
递归脱敏敏感字段，用于操作日志请求/响应记录
|
||
|
||
将密码、令牌等敏感字段的值替换为 "******"。
|
||
|
||
Args:
|
||
data: 待处理的数据（dict/list/基础类型）
|
||
|
||
Returns:
|
||
脱敏后的数据
|
||
"""
|
||
if isinstance(data, dict):
|
||
return {
|
||
k: ("******" if k in _SENSITIVE_KEYS else _redact_sensitive(v))
|
||
for k, v in data.items()
|
||
}
|
||
if isinstance(data, list):
|
||
return [_redact_sensitive(item) for item in data]
|
||
return data
|
||
|
||
|
||
# HTTP 方法 → 操作日志类型 映射
|
||
# 类型值见 scripts/seed_dict.py 的 DICT_SEED operation_type（显示文本由 dict_utils 从字典表翻译）
|
||
_METHOD_LOG_TYPE = {
|
||
"POST": 3, # 新增
|
||
"PUT": 4, # 修改
|
||
"DELETE": 5, # 删除
|
||
"PATCH": 4, # 修改
|
||
}
|
||
|
||
# 路径后缀 → 操作日志类型 覆盖映射（更精确）
|
||
_PATH_ACTION_TYPE = {
|
||
"/add": 3, # 新增
|
||
"/update": 4, # 修改
|
||
"/delete": 5, # 删除
|
||
"/status": 4, # 状态变更→修改
|
||
"/batchdelete": 5, # 批量删除
|
||
"/reset": 4, # 重置→修改
|
||
"/save": 4, # 保存→修改
|
||
"/sort": 4, # 排序→修改
|
||
}
|
||
|
||
|
||
def register_operation_log(app):
|
||
"""
|
||
注册操作日志自动记录中间件
|
||
|
||
在 after_request 中自动记录写操作日志。
|
||
|
||
Args:
|
||
app: Flask 应用实例
|
||
"""
|
||
|
||
@app.after_request
|
||
def record_op_log(response):
|
||
"""
|
||
after_request 钩子：自动记录写操作日志
|
||
|
||
记录条件：
|
||
1. 请求方法为 POST/PUT/DELETE/PATCH
|
||
2. 路径不在排除列表中
|
||
3. HTTP 状态码为 2xx
|
||
4. 非内部调用
|
||
|
||
记录范围：
|
||
- 成功操作（code=0）：status=0（正常）
|
||
- 失败操作（code!=0，如演示环境拦截）：status=1（异常），记录错误信息
|
||
"""
|
||
# 只记录写操作
|
||
if request.method not in ("POST", "PUT", "DELETE", "PATCH"):
|
||
return response
|
||
|
||
# 排除登录/退出/验证码/SMS 等敏感路径（尾段匹配，兼容任意版本前缀）
|
||
path = request.path
|
||
if _is_excluded_path(path):
|
||
return response
|
||
|
||
# HTTP 非 2xx 不记录（如 404、500 等）
|
||
if not (200 <= response.status_code < 300):
|
||
return response
|
||
|
||
# 内部调用不记录
|
||
if getattr(g, "_internal_call", False):
|
||
return response
|
||
|
||
# 判断业务状态：code=0 为成功，其他为失败
|
||
status = 0 # 默认成功
|
||
error_msg = ""
|
||
try:
|
||
data = response.get_json(silent=True)
|
||
if data and data.get("code", 0) != 0:
|
||
status = 1 # 业务失败
|
||
error_msg = data.get("msg", "操作失败")
|
||
except Exception:
|
||
pass
|
||
|
||
# 记录日志（成功和失败都记录）
|
||
try:
|
||
_write_log(response, status=status, error_msg=error_msg)
|
||
except Exception as e:
|
||
logger.warning(f"记录操作日志失败: {str(e)}")
|
||
|
||
return response
|
||
|
||
logger.info("操作日志中间件注册成功")
|
||
|
||
|
||
def _write_log(response, status=0, error_msg=""):
|
||
"""
|
||
写入操作日志
|
||
|
||
从请求和响应中提取日志所需信息，调用注入的 writer 函数写入。
|
||
|
||
Args:
|
||
response: Flask 响应对象
|
||
status: 操作状态（0-正常，1-异常）
|
||
error_msg: 错误信息（失败时）
|
||
"""
|
||
from src.core.security import get_username
|
||
from src.common.utils.ip import get_client_ip
|
||
from src.common.utils.ip2region import get_ip_location
|
||
from src.common.utils.user_agent import parse_user_agent
|
||
|
||
# 获取注入的日志写入器
|
||
writer = current_app.extensions.get("operation_log_writer")
|
||
if writer is None:
|
||
logger.warning("operation_log_writer 未注入，跳过操作日志")
|
||
return
|
||
|
||
username = get_username() or "system"
|
||
client_ip = get_client_ip()
|
||
ua_info = parse_user_agent()
|
||
|
||
# 确定操作类型
|
||
log_type = _infer_log_type(request.method, request.path)
|
||
|
||
# 构建标题：优先取路由 @operation_log 声明的元数据
|
||
title = _build_title(request.path, request.endpoint)
|
||
|
||
# 请求参数（限制长度，避免存入过大的数据）
|
||
param = _get_request_param()
|
||
|
||
# 返回结果（限制长度）
|
||
result = _get_response_result(response)
|
||
|
||
# 请求耗时
|
||
start_time = getattr(g, "request_start_time", None)
|
||
consume_time = int((time.time() - start_time) * 1000) if start_time else 0
|
||
|
||
writer(
|
||
title=title,
|
||
log_type=log_type,
|
||
source=0, # OperationLog.Source.ADMIN
|
||
status=status,
|
||
method=request.endpoint or "",
|
||
request_method=request.method,
|
||
url=request.path,
|
||
param=param,
|
||
result=result,
|
||
ip=client_ip,
|
||
location=get_ip_location(client_ip),
|
||
os=ua_info.get("os", "未知"),
|
||
browser=ua_info.get("browser", "未知"),
|
||
consume_time=consume_time,
|
||
error=error_msg,
|
||
username=username,
|
||
)
|
||
|
||
|
||
def _infer_log_type(method, path):
|
||
"""
|
||
根据请求方法和路径推断操作日志类型
|
||
|
||
优先根据路径后缀匹配，回退到 HTTP 方法映射。
|
||
|
||
Args:
|
||
method: HTTP 请求方法
|
||
path: 请求路径
|
||
|
||
Returns:
|
||
int: 操作日志类型编码
|
||
"""
|
||
# 优先根据路径后缀匹配
|
||
path_lower = path.lower()
|
||
for suffix, log_type in _PATH_ACTION_TYPE.items():
|
||
if path_lower.endswith(suffix):
|
||
return log_type
|
||
|
||
# 回退到 HTTP 方法映射
|
||
return _METHOD_LOG_TYPE.get(method, 18) # 18=其他
|
||
|
||
|
||
def _get_request_param():
|
||
"""
|
||
获取请求参数
|
||
|
||
优先从 JSON body 获取，回退到 URL 查询参数。
|
||
自动对敏感字段进行脱敏处理。
|
||
|
||
Returns:
|
||
str: JSON 格式的参数字符串，无参数时返回空字符串
|
||
"""
|
||
try:
|
||
if request.method in ("POST", "PUT", "PATCH", "DELETE"):
|
||
# 优先解析 JSON body（batchDelete 等 DELETE 请求也带 body）
|
||
data = request.get_json(silent=True)
|
||
if data:
|
||
import json
|
||
|
||
return json.dumps(_redact_sensitive(data), ensure_ascii=False)
|
||
# 回退到 URL 查询参数
|
||
if request.args:
|
||
import json
|
||
from collections import OrderedDict
|
||
|
||
return json.dumps(_redact_sensitive(OrderedDict(request.args)), ensure_ascii=False)
|
||
except Exception:
|
||
pass
|
||
return ""
|
||
|
||
|
||
def _get_response_result(response):
|
||
"""
|
||
获取响应结果中的 data 字段
|
||
|
||
Args:
|
||
response: Flask 响应对象
|
||
|
||
Returns:
|
||
str: JSON 格式的响应数据字符串，无数据时返回空字符串
|
||
"""
|
||
try:
|
||
data = response.get_json(silent=True)
|
||
if data and isinstance(data, dict):
|
||
result_data = data.get("data")
|
||
if result_data is not None:
|
||
import json
|
||
|
||
return json.dumps(_redact_sensitive(result_data), ensure_ascii=False)
|
||
# data字段为空时取msg
|
||
return data.get("msg", "")
|
||
except Exception:
|
||
pass
|
||
return ""
|
||
|
||
|
||
def _build_title(path, endpoint):
|
||
"""
|
||
构建操作日志标题
|
||
|
||
优先读取路由视图函数 `@operation_log` 声明的元数据；
|
||
未标注的路由走路径兜底（模块段-动作段，不做中文映射）。
|
||
|
||
Args:
|
||
path: 请求路径
|
||
endpoint: Flask 端点名称
|
||
|
||
Returns:
|
||
str: 操作日志标题
|
||
"""
|
||
view_func = current_app.view_functions.get(endpoint)
|
||
if view_func:
|
||
title = getattr(view_func, "__op_log_title__", None)
|
||
if title:
|
||
return title
|
||
|
||
# 兜底：未标注路由，用路径原样（剥离 /api/v1 版本前缀，取前两段）
|
||
parts = [p for p in path.strip("/").split("/") if p]
|
||
if parts and parts[0] == "api":
|
||
parts = parts[2:] if len(parts) > 2 else []
|
||
return "-".join(parts[:2]) if parts else path
|