4808a48594
一、统一为单一监听器实现
原先两套 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），模块中文，操作人真实，时间为北京时间。
91 lines
3.6 KiB
Python
91 lines
3.6 KiB
Python
# inventory-backend/app/api/v1/permission.py
|
||
from flask import Blueprint, request, jsonify, current_app
|
||
from flask_jwt_extended import jwt_required, get_jwt
|
||
from app.services.permission_service import PermissionService
|
||
from app.utils.decorators import permission_required
|
||
|
||
permission_bp = Blueprint('permission', __name__)
|
||
|
||
|
||
def _get_operator_company():
|
||
"""从 JWT 获取当前操作者公司（None=超管，具体值=该角色所属公司）"""
|
||
claims = get_jwt()
|
||
role = claims.get('role', '')
|
||
if role and role.upper() == 'SUPER_ADMIN':
|
||
return None # 超管不限制公司
|
||
return claims.get('company_name', '')
|
||
|
||
|
||
def _has_system_permission(role_code):
|
||
"""检查角色是否有 system_permission"""
|
||
try:
|
||
from app.services.auth_service import AuthService
|
||
perm_dict = AuthService.get_user_permissions(role_code)
|
||
all_perms = perm_dict.get('menus', []) + perm_dict.get('elements', [])
|
||
return 'system_permission' in all_perms
|
||
except Exception:
|
||
return False
|
||
|
||
|
||
@permission_bp.route('/tree', methods=['GET'])
|
||
@jwt_required()
|
||
@permission_required('system_permission')
|
||
def get_tree():
|
||
"""获取权限树"""
|
||
try:
|
||
data = PermissionService.get_permission_tree()
|
||
return jsonify({'code': 200, 'msg': '获取成功', 'data': data}), 200
|
||
except Exception as e:
|
||
current_app.logger.error(f"Get Tree Failed: {str(e)}")
|
||
return jsonify({'code': 500, 'msg': f'服务器内部错误: {str(e)}'}), 500
|
||
|
||
|
||
@permission_bp.route('/role/<string:role_code>', methods=['GET'])
|
||
@jwt_required()
|
||
def get_role_perms(role_code):
|
||
"""获取某个角色的权限列表。
|
||
- 查自己角色：不需要额外权限
|
||
- 查其他角色：需要 system_permission
|
||
"""
|
||
try:
|
||
claims = get_jwt()
|
||
current_role = (claims.get('role') or '').upper()
|
||
|
||
# 非管理员查其他角色 → 拒绝
|
||
if current_role != role_code.upper() and current_role != 'SUPER_ADMIN':
|
||
if not _has_system_permission(current_role):
|
||
return jsonify({'code': 403, 'msg': '无权查看其他角色的权限'}), 403
|
||
|
||
company_name = _get_operator_company()
|
||
data = PermissionService.get_role_permissions(role_code, company_name=company_name)
|
||
return jsonify({'code': 200, 'msg': '获取成功', 'data': data}), 200
|
||
except Exception as e:
|
||
current_app.logger.error(f"Get Role Perms Failed: {str(e)}")
|
||
return jsonify({'code': 500, 'msg': str(e)}), 500
|
||
|
||
|
||
@permission_bp.route('/assign', methods=['POST'])
|
||
@jwt_required()
|
||
@permission_required('system_permission:operation')
|
||
def assign_perms():
|
||
"""保存权限分配（自动带上当前操作者的公司标识）"""
|
||
try:
|
||
data = request.get_json()
|
||
role_code = data.get('role_code')
|
||
permissions = data.get('permissions', [])
|
||
|
||
# ★ 防自锁：操作者修改自己角色时，强制保留权限管理的关键权限
|
||
claims = get_jwt()
|
||
operator_role = claims.get('role', '')
|
||
if operator_role and operator_role.upper() == role_code.upper():
|
||
for required in ('system_permission', 'system_permission:operation'):
|
||
if required not in permissions:
|
||
permissions.append(required)
|
||
|
||
company_name = _get_operator_company()
|
||
PermissionService.assign_permissions(role_code, permissions, company_name=company_name)
|
||
return jsonify({'code': 200, 'msg': '保存成功'}), 200
|
||
except Exception as e:
|
||
current_app.logger.error(f"Assign Perms Failed: {str(e)}")
|
||
return jsonify({'code': 500, 'msg': str(e)}), 500
|