fix: 修复 JWT 幽灵令牌漏洞，新增 Dify 权限过滤服务
This commit is contained in:
284
inventory-backend/app/services/dify_permission_service.py
Normal file
284
inventory-backend/app/services/dify_permission_service.py
Normal file
@ -0,0 +1,284 @@
|
||||
# app/services/dify_permission_service.py
|
||||
"""
|
||||
Dify 智能客服权限服务层
|
||||
|
||||
职责：
|
||||
1. 从 JWT Token / g.dify_user_role 解析用户角色
|
||||
2. 根据角色查询 sys_role_permission 表，获取用户拥有的 target_code
|
||||
3. AI 专属的动态脱敏策略：
|
||||
- SUPER_ADMIN：无条件放行所有数据
|
||||
- SALES / INBOUND：剔除 price, cost, supplier 等敏感字段
|
||||
- 跨模块越权查询：直接阻断，返回角色专属的错误信息给大模型
|
||||
"""
|
||||
|
||||
from flask import g, current_app
|
||||
from flask_jwt_extended import decode_token
|
||||
from app.models.system import SysRolePermission
|
||||
from app.services.auth_service import AuthService
|
||||
from app.utils.constants import UserRole
|
||||
from sqlalchemy import func
|
||||
|
||||
|
||||
# ==============================================================================
|
||||
# 角色敏感字段定义（按角色黑名单）
|
||||
# ==============================================================================
|
||||
# 每个角色在 AI 对话场景下，禁止查看的字段集合
|
||||
ROLE_SENSITIVE_FIELDS = {
|
||||
# 入库员：禁止查看采购相关金额
|
||||
'INBOUND': frozenset([
|
||||
'purchase_price',
|
||||
'cost',
|
||||
'price',
|
||||
'supplier',
|
||||
'supplier_id',
|
||||
'unit_price',
|
||||
'total_price',
|
||||
'order_price',
|
||||
'last_purchase_price',
|
||||
'avg_purchase_price',
|
||||
'supplier_name',
|
||||
]),
|
||||
# 销售员：禁止查看成本/采购相关数据
|
||||
'SALES': frozenset([
|
||||
'purchase_price',
|
||||
'cost',
|
||||
'unit_cost',
|
||||
'supplier',
|
||||
'supplier_id',
|
||||
'unit_price',
|
||||
'last_purchase_price',
|
||||
'avg_purchase_price',
|
||||
'supplier_name',
|
||||
'stock_cost',
|
||||
'total_cost',
|
||||
'supply_price',
|
||||
]),
|
||||
# 采购员：禁止查看销售毛利相关数据
|
||||
'PURCHASER': frozenset([
|
||||
'sale_price',
|
||||
'retail_price',
|
||||
'suggested_price',
|
||||
'margin',
|
||||
'profit',
|
||||
'profit_rate',
|
||||
]),
|
||||
}
|
||||
|
||||
# 跨模块越权查询拦截配置
|
||||
# key: 角色, value: { 尝试查询的字段: 给 AI 的回复 }
|
||||
ROLE_FORBIDDEN_QUERIES = {
|
||||
'INBOUND': {
|
||||
'purchase_price': '抱歉，您当前的角色（入库员）无权查看采购金额数据。',
|
||||
'cost': '抱歉，您当前的角色（入库员）无权查看采购成本数据。',
|
||||
'supplier': '抱歉，您当前的角色（入库员）无权查看供应商数据。',
|
||||
'unit_price': '抱歉，您当前的角色（入库员）无权查看采购单价数据。',
|
||||
'last_purchase_price': '抱歉，您当前的角色（入库员）无权查看历史采购价数据。',
|
||||
},
|
||||
'SALES': {
|
||||
'purchase_price': '抱歉，您当前的角色（销售员）无权查看采购成本数据。',
|
||||
'cost': '抱歉，您当前的角色（销售员）无权查看成本数据。',
|
||||
'supplier': '抱歉，您当前的角色（销售员）无权查看供应商信息。',
|
||||
'unit_cost': '抱歉，您当前的角色（销售员）无权查看单位成本数据。',
|
||||
'last_purchase_price': '抱歉，您当前的角色（销售员）无权查看历史采购价数据。',
|
||||
},
|
||||
'PURCHASER': {
|
||||
'sale_price': '抱歉，您当前的角色（采购员）无权查看销售价格数据。',
|
||||
'retail_price': '抱歉，您当前的角色（采购员）无权查看零售价数据。',
|
||||
'margin': '抱歉，您当前的角色（采购员）无权查看毛利数据。',
|
||||
'profit': '抱歉，您当前的角色（采购员）无权查看利润数据。',
|
||||
},
|
||||
}
|
||||
|
||||
|
||||
class DifyPermissionService:
|
||||
"""Dify AI 专属权限服务"""
|
||||
|
||||
@staticmethod
|
||||
def get_user_role(token: str = None) -> str:
|
||||
"""
|
||||
从 JWT Token 或 g.dify_user_role 解析用户角色
|
||||
返回标准化的大写角色码（如 'INBOUND', 'SALES', 'SUPER_ADMIN'）
|
||||
"""
|
||||
user_role = None
|
||||
|
||||
# 优先从 g 对象获取（由 dify_auth_required 存入）
|
||||
if hasattr(g, 'dify_user_role') and g.dify_user_role:
|
||||
return g.dify_user_role
|
||||
|
||||
# 从传入的 token 解码
|
||||
if token:
|
||||
try:
|
||||
claims = decode_token(token)
|
||||
user_role = claims.get('role')
|
||||
except Exception as e:
|
||||
current_app.logger.warning(f"[DifyPermission] token 解码失败: {e}")
|
||||
return ''
|
||||
|
||||
return (user_role or '').upper()
|
||||
|
||||
@staticmethod
|
||||
def is_super_admin(role: str = None) -> bool:
|
||||
"""判断是否为超级管理员"""
|
||||
if not role:
|
||||
role = DifyPermissionService.get_user_role()
|
||||
return role.upper() == UserRole.SUPER_ADMIN if role else False
|
||||
|
||||
@staticmethod
|
||||
def get_role_permissions(role: str = None) -> dict:
|
||||
"""
|
||||
获取指定角色的所有权限代码列表
|
||||
内部调用 AuthService.get_user_permissions()
|
||||
"""
|
||||
if not role:
|
||||
role = DifyPermissionService.get_user_role()
|
||||
return AuthService.get_user_permissions(role)
|
||||
|
||||
@staticmethod
|
||||
def get_target_codes(role: str = None) -> list:
|
||||
"""
|
||||
获取用户角色在 sys_role_permission 表中的所有 target_code
|
||||
包括 menu 和 element 两种类型
|
||||
"""
|
||||
if not role:
|
||||
role = DifyPermissionService.get_user_role()
|
||||
|
||||
if not role:
|
||||
return []
|
||||
|
||||
# 超级管理员拥有所有权限
|
||||
if DifyPermissionService.is_super_admin(role):
|
||||
return ['*']
|
||||
|
||||
# 从数据库查询
|
||||
try:
|
||||
perms = SysRolePermission.query.filter(
|
||||
func.upper(SysRolePermission.role_code) == role.upper()
|
||||
).all()
|
||||
return [p.target_code for p in perms]
|
||||
except Exception as e:
|
||||
current_app.logger.error(f"[DifyPermission] 查询 target_code 失败: {e}")
|
||||
return []
|
||||
|
||||
@staticmethod
|
||||
def has_permission(permission_code: str, role: str = None) -> bool:
|
||||
"""
|
||||
检查用户是否拥有指定权限码
|
||||
超级管理员永远返回 True
|
||||
"""
|
||||
if DifyPermissionService.is_super_admin(role):
|
||||
return True
|
||||
if not role:
|
||||
role = DifyPermissionService.get_user_role()
|
||||
perms = DifyPermissionService.get_role_permissions(role)
|
||||
all_perms = perms.get('menus', []) + perms.get('elements', [])
|
||||
return permission_code in all_perms
|
||||
|
||||
@staticmethod
|
||||
def check_forbidden_query(query_fields: list, role: str = None) -> dict:
|
||||
"""
|
||||
检查用户尝试查询的字段是否越权。
|
||||
|
||||
参数:
|
||||
query_fields: 用户查询涉及的字段名列表（可能包含敏感字段）
|
||||
role: 用户角色（可选，默认从 token/g 解析）
|
||||
|
||||
返回:
|
||||
{
|
||||
'blocked': bool, # 是否被拦截
|
||||
'message': str | None, # AI 应返回给用户的错误信息（如果有）
|
||||
}
|
||||
"""
|
||||
if DifyPermissionService.is_super_admin(role):
|
||||
return {'blocked': False, 'message': None}
|
||||
|
||||
if not role:
|
||||
role = DifyPermissionService.get_user_role()
|
||||
|
||||
if not role:
|
||||
return {
|
||||
'blocked': True,
|
||||
'message': '无法识别您的身份，请重新登录后再试。'
|
||||
}
|
||||
|
||||
# 获取该角色的越权拦截配置
|
||||
forbidden_map = ROLE_FORBIDDEN_QUERIES.get(role.upper(), {})
|
||||
if not forbidden_map:
|
||||
return {'blocked': False, 'message': None}
|
||||
|
||||
# 规范化字段名（小写比较）
|
||||
query_fields_lower = {f.lower() for f in query_fields}
|
||||
|
||||
# 检查是否命中越权字段
|
||||
for forbidden_field, msg in forbidden_map.items():
|
||||
if forbidden_field.lower() in query_fields_lower:
|
||||
current_app.logger.warning(
|
||||
f"[DifyPermission] 越权查询拦截: role={role}, field={forbidden_field}"
|
||||
)
|
||||
return {'blocked': True, 'message': msg}
|
||||
|
||||
return {'blocked': False, 'message': None}
|
||||
|
||||
@staticmethod
|
||||
def filter_sensitive_fields(data: dict, role: str = None) -> dict:
|
||||
"""
|
||||
根据用户角色对字典数据进行敏感字段脱敏。
|
||||
|
||||
- SUPER_ADMIN：不过滤，返回原数据
|
||||
- SALES / INBOUND / PURCHASER：按角色黑名单剔除敏感字段
|
||||
|
||||
参数:
|
||||
data: 原始数据字典
|
||||
role: 用户角色
|
||||
|
||||
返回:
|
||||
脱敏后的数据字典（敏感字段被置为 None）
|
||||
"""
|
||||
if DifyPermissionService.is_super_admin(role):
|
||||
return data
|
||||
|
||||
if not role:
|
||||
role = DifyPermissionService.get_user_role()
|
||||
|
||||
if not role:
|
||||
return data
|
||||
|
||||
# 获取该角色的敏感字段黑名单
|
||||
sensitive = ROLE_SENSITIVE_FIELDS.get(role.upper(), frozenset())
|
||||
|
||||
# 如果没有敏感字段定义，不过滤
|
||||
if not sensitive:
|
||||
return data
|
||||
|
||||
# 深拷贝，避免修改原数据
|
||||
import copy
|
||||
result = copy.deepcopy(data)
|
||||
|
||||
# 将敏感字段置为 None
|
||||
for field in sensitive:
|
||||
if field in result:
|
||||
result[field] = None
|
||||
|
||||
# 如果有 children 数组（子件列表），递归脱敏
|
||||
if isinstance(result.get('children'), list):
|
||||
result['children'] = [
|
||||
DifyPermissionService.filter_sensitive_fields(child, role)
|
||||
for child in result['children']
|
||||
]
|
||||
|
||||
return result
|
||||
|
||||
@staticmethod
|
||||
def filter_sensitive_fields_in_list(data_list: list, role: str = None) -> list:
|
||||
"""
|
||||
对列表数据批量应用敏感字段脱敏
|
||||
"""
|
||||
if DifyPermissionService.is_super_admin(role):
|
||||
return data_list
|
||||
|
||||
if not role:
|
||||
role = DifyPermissionService.get_user_role()
|
||||
|
||||
return [
|
||||
DifyPermissionService.filter_sensitive_fields(item, role)
|
||||
for item in data_list
|
||||
]
|
||||
Reference in New Issue
Block a user