67d3113fe1
- PRIVILEGED_VIEWER_ROLES 收敛为 SUPER_ADMIN/SUPERVISOR/WAREHOUSE_MGR - 入库员(INBOUND)、出库员(OUTBOUND)按普通处理：借还/出库记录只看自己
374 lines
14 KiB
Python
374 lines
14 KiB
Python
# app/utils/decorators.py
|
||
from functools import wraps
|
||
from flask_jwt_extended import get_jwt, verify_jwt_in_request, get_jwt_identity
|
||
from flask import jsonify, g, request, current_app, has_request_context
|
||
import logging
|
||
import json
|
||
|
||
def _verify_user_active():
|
||
"""
|
||
JWT「幽灵令牌」安全漏洞修复：
|
||
在 Token 签名验证通过之后，进一步检查用户在数据库中是否仍然存在且未被禁用。
|
||
|
||
调用时机：login_required / permission_required 装饰器中，
|
||
在 verify_jwt_in_request() 成功之后立即调用。
|
||
|
||
返回 True → 用户正常，放行
|
||
返回 False → 用户已从数据库删除或被禁用，阻断请求
|
||
"""
|
||
try:
|
||
claims = get_jwt()
|
||
user_id = claims.get('sub')
|
||
if user_id is None:
|
||
return True
|
||
|
||
from app.models.system import SysUser
|
||
user = SysUser.query.get(user_id)
|
||
if user is None:
|
||
current_app.logger.warning(
|
||
f"🚫 [Ghost Token Blocked] user_id={user_id} not found in database (deleted account)"
|
||
)
|
||
return False
|
||
|
||
if user.status != 'active':
|
||
current_app.logger.warning(
|
||
f"🚫 [Token Blocked] user_id={user_id} status={user.status} (disabled account)"
|
||
)
|
||
return False
|
||
|
||
return True
|
||
except Exception as e:
|
||
current_app.logger.error(f"User active check error: {e}")
|
||
return True # 出错时 fail-open，避免数据库故障导致全站不可用
|
||
|
||
|
||
def _verify_token_in_redis():
|
||
"""
|
||
验证当前 Token 是否与 Redis 中存储的 Token 一致（单设备登录互踢）
|
||
"""
|
||
from app.extensions import redis_client
|
||
from flask import current_app
|
||
|
||
if redis_client is None:
|
||
return True
|
||
|
||
try:
|
||
auth_header = request.headers.get('Authorization', '')
|
||
if not auth_header.startswith('Bearer '):
|
||
return True
|
||
|
||
request_token = auth_header[7:]
|
||
claims = get_jwt()
|
||
user_id = claims.get('sub')
|
||
if user_id is None:
|
||
return True
|
||
|
||
stored_token = redis_client.get(f"user_token_{user_id}")
|
||
if stored_token is None:
|
||
return True
|
||
|
||
if request_token != stored_token:
|
||
current_app.logger.warning(f"Token mismatch for user {user_id}: request token != stored token")
|
||
return False
|
||
|
||
return True
|
||
except Exception as e:
|
||
current_app.logger.error(f"Redis token verification error: {e}")
|
||
return True
|
||
|
||
def _raise_token_mismatch_error():
|
||
"""抛出 Token 不一致的错误"""
|
||
return jsonify({
|
||
'msg': '您的账号已在其他设备登录，请重新登录',
|
||
'code': 401,
|
||
'reason': 'token_mismatch'
|
||
}), 401
|
||
|
||
def role_required(*roles):
|
||
"""自定义装饰器：检查用户角色"""
|
||
def wrapper(fn):
|
||
@wraps(fn)
|
||
def decorator(*args, **kwargs):
|
||
claims = get_jwt()
|
||
user_role = claims.get('role')
|
||
user_role_upper = user_role.upper() if user_role else None
|
||
|
||
if user_role_upper == 'SUPER_ADMIN':
|
||
return fn(*args, **kwargs)
|
||
|
||
if user_role_upper not in [r.upper() for r in roles]:
|
||
return jsonify(msg='权限不足：您没有访问此资源的权限'), 403
|
||
|
||
return fn(*args, **kwargs)
|
||
return decorator
|
||
return wrapper
|
||
|
||
def login_required(fn):
|
||
"""
|
||
验证 JWT 令牌是否存在且有效，并检查用户是否仍在数据库中且未被禁用。
|
||
双重防护：1) Token 签名验证 2) 数据库用户存在性 3) Redis 单设备互踢
|
||
"""
|
||
@wraps(fn)
|
||
def decorator(*args, **kwargs):
|
||
try:
|
||
verify_jwt_in_request()
|
||
except Exception as e:
|
||
logging.warning(f"JWT verification failed: {e}")
|
||
return jsonify(msg='登录已过期，请重新登录'), 401
|
||
|
||
# ★ 幽灵令牌漏洞修复：检查用户是否已从数据库删除或被禁用
|
||
if not _verify_user_active():
|
||
return jsonify(msg='账号已失效（已删除或已禁用），请重新登录'), 401
|
||
|
||
if not _verify_token_in_redis():
|
||
return _raise_token_mismatch_error()
|
||
|
||
return fn(*args, **kwargs)
|
||
return decorator
|
||
|
||
def _expand_operation_perms(permission_code, all_perms):
|
||
"""
|
||
操作权限自动展开映射器（双向粒度桥接）。
|
||
"""
|
||
if permission_code in all_perms:
|
||
logging.info(f"[权限展开] 精确匹配: {permission_code}")
|
||
return True
|
||
|
||
KNOWN_OPERATION_SUFFIXES = (
|
||
':operation', ':add', ':edit', ':delete', ':submit',
|
||
':approve', ':dispatch', ':write'
|
||
)
|
||
|
||
if ':' in permission_code:
|
||
prefix = permission_code.rsplit(':', 1)[0]
|
||
else:
|
||
prefix = permission_code
|
||
|
||
# 收集该模块下的所有用户权限，用于诊断日志
|
||
module_perms = [p for p in all_perms if p.startswith(prefix)]
|
||
|
||
for perm in all_perms:
|
||
if perm.startswith(prefix + ':') or perm == prefix:
|
||
for suffix in KNOWN_OPERATION_SUFFIXES:
|
||
if perm.endswith(suffix):
|
||
logging.info(f"[权限展开] {permission_code} ← 用户有 {perm} → 通过")
|
||
return True
|
||
if ':' not in permission_code and ':' in perm:
|
||
logging.info(f"[权限展开] {permission_code} ← 用户有 {perm}（下级权限）→ 通过")
|
||
return True
|
||
|
||
logging.warning(
|
||
f"[权限展开] 失败: 要求={permission_code}, "
|
||
f"该模块用户权限={module_perms}, 全部权限数={len(all_perms)}"
|
||
)
|
||
return False
|
||
|
||
|
||
def permission_required(permission_code):
|
||
"""检查当前用户是否拥有指定权限码，同时检查用户是否仍然有效"""
|
||
def wrapper(fn):
|
||
@wraps(fn)
|
||
def decorator(*args, **kwargs):
|
||
try:
|
||
verify_jwt_in_request()
|
||
except Exception as e:
|
||
logging.warning(f"JWT verification failed: {e}")
|
||
return jsonify(msg='登录已过期，请重新登录'), 401
|
||
|
||
# ★ 幽灵令牌漏洞修复：检查用户是否已从数据库删除或被禁用
|
||
if not _verify_user_active():
|
||
return jsonify(msg='账号已失效（已删除或已禁用），请重新登录'), 401
|
||
|
||
if not _verify_token_in_redis():
|
||
return _raise_token_mismatch_error()
|
||
|
||
claims = get_jwt()
|
||
user_role = claims.get('role')
|
||
if user_role and user_role.upper() == 'SUPER_ADMIN':
|
||
return fn(*args, **kwargs)
|
||
|
||
# 获取当前用户公司，用于权限隔离
|
||
user_company = claims.get('company_name', '')
|
||
|
||
try:
|
||
from app.services.auth_service import AuthService
|
||
perm_dict = AuthService.get_user_permissions(user_role, company_name=user_company)
|
||
except Exception as e:
|
||
logging.warning(f"Failed to fetch permissions for role {user_role}: {e}")
|
||
return jsonify(msg='权限查询失败'), 403
|
||
|
||
all_perms = perm_dict.get('menus', []) + perm_dict.get('elements', [])
|
||
|
||
# ★ 操作权限展开检查（粒度桥接）
|
||
if not _expand_operation_perms(permission_code, all_perms):
|
||
logging.warning(
|
||
f"权限检查失败: 角色={user_role}, 所需={permission_code}, "
|
||
f"拥有={[p for p in all_perms if permission_code.split(':')[0] in p]}"
|
||
)
|
||
return jsonify(msg='权限不足：您没有访问此资源的权限'), 403
|
||
return fn(*args, **kwargs)
|
||
return decorator
|
||
return wrapper
|
||
|
||
def _has_cross_domain_permission(user_role, user_company=''):
|
||
"""
|
||
检查指定角色是否拥有「全局跨域访问」权限。
|
||
复用 AuthService.get_user_permissions 统一查询。
|
||
"""
|
||
if not user_role:
|
||
return False
|
||
try:
|
||
from app.services.auth_service import AuthService
|
||
perm_dict = AuthService.get_user_permissions(user_role, company_name=user_company)
|
||
all_perms = perm_dict.get('menus', []) + perm_dict.get('elements', [])
|
||
return 'crossDomain' in all_perms
|
||
except Exception:
|
||
return False
|
||
|
||
|
||
# 能查看“他人借还/出库记录”的管理者角色（仓库管理员/主管/超管）。普通申请人(含入库员/出库员等)只能看自己的记录。
|
||
# 注意：入库员(INBOUND)、出库员(OUTBOUND)不是管理者视角，不应放此名单；如需让他们看全部请确认为库管角色后再加入。
|
||
PRIVILEGED_VIEWER_ROLES = ('SUPER_ADMIN', 'SUPERVISOR', 'WAREHOUSE_MGR')
|
||
|
||
|
||
def is_privileged_viewer():
|
||
"""
|
||
判断当前登录用户是否属于“管理者视角”（可查看他人借还/出库记录）。
|
||
规则：角色为 SUPER_ADMIN/SUPERVISOR/WAREHOUSE_MGR，或拥有 crossDomain 权限。
|
||
普通申请人（其余角色）只能看自己的记录（由各列表 handler 强制 applicant_id=当前用户）。
|
||
"""
|
||
from flask_jwt_extended import get_jwt
|
||
claims = get_jwt()
|
||
role = (claims.get('role') or '').upper()
|
||
if role in PRIVILEGED_VIEWER_ROLES:
|
||
return True
|
||
user_company = claims.get('company_name', '')
|
||
if _has_cross_domain_permission(role, user_company):
|
||
return True
|
||
return False
|
||
|
||
|
||
def get_current_company_filter():
|
||
"""
|
||
多租户数据权限隔离工具函数。
|
||
|
||
权限逻辑（基于权限码 crossDomain，支持动态分配）:
|
||
- SUPER_ADMIN → None（全量跨域）
|
||
- 拥有 crossDomain 权限 → None（全量跨域）
|
||
- 普通用户 → 强制隔离到 JWT 中记录的公司
|
||
|
||
返回值:
|
||
None → 不限制公司
|
||
str → 仅查询该公司数据
|
||
|
||
使用示例:
|
||
company_limit = get_current_company_filter()
|
||
if company_limit is not None:
|
||
query = query.filter(MaterialBase.company_name == company_limit)
|
||
"""
|
||
from flask import request
|
||
|
||
claims = get_jwt()
|
||
user_role = claims.get('role', '')
|
||
|
||
# 规范化角色为大写
|
||
if user_role:
|
||
user_role = user_role.upper()
|
||
|
||
# 从请求参数获取前端指定的公司（兼容 company_name 和 company 两个参数名）
|
||
req_company = request.args.get('company_name', '') or request.args.get('company', '')
|
||
|
||
# 超级管理员 → 全局跨域
|
||
if user_role == 'SUPER_ADMIN':
|
||
if req_company and req_company.strip().upper() != 'ALL':
|
||
return req_company.strip()
|
||
return None
|
||
|
||
# 动态权限码检查：拥有 crossDomain 权限 → 全局跨域
|
||
user_company = claims.get('company_name', '')
|
||
if user_role and _has_cross_domain_permission(user_role, user_company):
|
||
if req_company and req_company.strip().upper() != 'ALL':
|
||
return req_company.strip()
|
||
return None
|
||
|
||
# 普通用户 → 强制隔离到本公司
|
||
# 如果 JWT 中没有 company_name，返回哨兵值确保不会匹配任何数据
|
||
if user_company:
|
||
return user_company
|
||
return '__NO_COMPANY__'
|
||
|
||
|
||
def audit_log(module: str = None, action: str = None, get_target_id_fn=None, get_target_name_fn=None, get_details_fn=None):
|
||
"""
|
||
已废弃！
|
||
由 SQLAlchemy 底层监听器（app/core/audit_listener.py）全面接管审计日志入库。
|
||
此装饰器保留空壳以防项目中其他文件 import 引用时报错。
|
||
"""
|
||
def wrapper(fn):
|
||
from functools import wraps
|
||
@wraps(fn)
|
||
def decorator(*inner_args, **inner_kwargs):
|
||
return fn(*inner_args, **inner_kwargs)
|
||
return decorator
|
||
return wrapper
|
||
|
||
|
||
def prevent_double_submit(lock_timeout=5):
|
||
"""
|
||
Redis 幂等锁装饰器 — 防止重复提交。
|
||
|
||
原理：
|
||
- 基于 (user_id, request.path, hash(body)) 生成唯一 Redis key
|
||
- 若 key 存在 → 返回 409 Conflict
|
||
- 若不存在 → setex(key, timeout, "1") → 执行业务 → delete(key)
|
||
|
||
用法:
|
||
@app.route('/submit', methods=['POST'])
|
||
@prevent_double_submit(lock_timeout=10)
|
||
def submit(): ...
|
||
"""
|
||
import hashlib
|
||
from app.extensions import redis_client
|
||
|
||
def wrapper(fn):
|
||
@wraps(fn)
|
||
def decorator(*args, **kwargs):
|
||
if redis_client is None:
|
||
return fn(*args, **kwargs) # Redis 不可用时 fail-open
|
||
|
||
try:
|
||
from flask_jwt_extended import get_jwt_identity
|
||
user_id = str(get_jwt_identity() or 'anonymous')
|
||
path = request.path
|
||
|
||
# 对请求体做确定性哈希（空 body = 空串 hash）
|
||
body_str = ''
|
||
if request.is_json:
|
||
body_str = json.dumps(request.get_json() or {}, sort_keys=True)
|
||
body_hash = hashlib.md5(body_str.encode()).hexdigest()[:12]
|
||
|
||
lock_key = f"idem:{user_id}:{path}:{body_hash}"
|
||
|
||
if redis_client.exists(lock_key):
|
||
logging.warning(f"[幂等锁] 重复请求被拦截: {lock_key}")
|
||
return jsonify({
|
||
'code': 409,
|
||
'msg': '请求正在处理中，请勿重复提交'
|
||
}), 409
|
||
|
||
redis_client.setex(lock_key, lock_timeout, "1")
|
||
|
||
try:
|
||
return fn(*args, **kwargs)
|
||
finally:
|
||
try:
|
||
redis_client.delete(lock_key)
|
||
except Exception:
|
||
pass # 清理失败不影响业务
|
||
|
||
except Exception as e:
|
||
logging.warning(f"[幂等锁] Redis 异常，降级放行: {e}")
|
||
return fn(*args, **kwargs) # fail-open
|
||
|
||
return decorator
|
||
return wrapper |