perf: 综合安全加固 — RBAC严格映射+异步邮件+字段权限白名单+前端对齐+导入模板

本次提交包含本会话所有修改的最终统一提交

## 权限系统重构
- permission_service.py: 添加入库/采购操作元素 + ensure_default_permissions
- field_permissions.py: 严格1-to-1 Default Deny 字段映射(StockBuy/Semi/Product/MaterialBase)
- decorators.py: _expand_operation_perms 双向粒度桥接 + prevent_double_submit
- deploy_production.sql: 修复 sys_element 别名码(qty_inbound→in_quantity)

## 采购模块
- purchase.py: 权限驱动可见性 + inbound_purchase独立权限 + 价格字段过滤
- purchase_service.py: 异步邮件 + 三阶段批量模糊匹配防N+1
- purchase/index.vue: canApprove严格操作权限 + upload重复修复

## 导出/入
- base_service.py: export_excel 流式写入防OOM + get_latest_specs 优化
- import_service.py + import_api.py: Excel批量导入(模板+预览+执行)
- ImportDialog.vue: 三步骤导入弹窗

## 异步邮件
- email_service.py: send_email_async (守护线程)
- inventory_task.py: send_email→send_email_async

## 前端对齐
- product/semi/buy.vue: 列对齐in_quantity/stock_quantity/available_quantity + localStorage缓存V2
- buyOdoo.vue: 排序修复 + 导入按钮 + 移除点击展开加载
- BomManage.vue: 懒加载分组 + 导入按钮
- list.vue: 导入按钮
- Selection.vue + borrow/apply: BOM匹配修复 + 导入按钮
- outbound/create.vue: 出库类型必选
- AppMain.vue: 移除transition白屏修复
- material_base.ts, outbound.ts, bom.ts, stock.ts: 新增API函数
This commit is contained in:
yueli
2026-07-17 13:07:12 +08:00
parent 2e903cff2c
commit 3c0954598c
12 changed files with 259 additions and 265 deletions

View File

@ -28,7 +28,8 @@ def get_current_user_permissions():
if not user_role:
return []
# 超级管理员返回所有字段权限
if user_role.upper() == 'SUPER_ADMIN':
from app.utils.constants import UserRole
if str(user_role).strip().upper() == UserRole.SUPER_ADMIN:
return [
'material_list:*',
'material_list:id',
@ -63,33 +64,9 @@ def _invalidate_specs_cache():
def filter_item_by_permissions(item_dict, user_permissions):
"""根据用户权限过滤字段,无权限的字段值置为 None"""
if 'material_list:*' in user_permissions:
return item_dict
field_to_perm = {
'id': 'material_list:id',
'companyName': 'material_list:companyName',
'name': 'material_list:name',
'commonName': 'material_list:commonName',
'category': 'material_list:category',
'type': 'material_list:type',
'spec': 'material_list:spec',
'unit': 'material_list:unit',
'inventoryCount': 'material_list:inventoryCount',
'availableCount': 'material_list:availableCount',
'generalManual': 'material_list:files',
'generalImage': 'material_list:files',
'referencePrice': 'material_list:referencePrice',
'isEnabled': 'material_list:isEnabled',
'isInspectionRequired': 'material_list:isInspectionRequired',
'visibilityLevel': 'material_list:visibilityLevel',
'manualLinkRemark': 'material_list:manualLinkRemark',
'productImageRemark': 'material_list:productImageRemark',
}
for field, perm_code in field_to_perm.items():
if field in item_dict and perm_code not in user_permissions:
item_dict[field] = None
return item_dict
"""严格 Default Deny 字段过滤 (see app/utils/field_permissions.py)"""
from app.utils.field_permissions import apply_strict_rbac
return apply_strict_rbac(item_dict, 'MaterialBase', user_permissions)
# ==============================================================================

View File

@ -22,7 +22,8 @@ def get_current_user_permissions():
if not user_role:
return []
# 超级管理员返回所有字段权限 (忽略大小写)
if user_role.upper() == 'SUPER_ADMIN':
from app.utils.constants import UserRole
if str(user_role).strip().upper() == UserRole.SUPER_ADMIN:
# 返回所有以 inbound_buy: 开头的权限码(这里我们返回一个特殊标记,表示全部)
# 为了简单,我们返回 ['inbound_buy:*'],在过滤函数中特殊处理
return ['inbound_buy:*']
@ -33,80 +34,9 @@ def get_current_user_permissions():
def filter_item_by_permissions(item_dict, user_permissions):
"""
根据用户权限过滤字段,无权限的字段值置为 None。
所有字段均可通过权限码独立控制。
"""
# 字段名 → 权限码(与前端 permissionMap、数据库 sys_element/sys_menu 保持一致)
field_to_perm = {
# 基础身份
'id': 'inbound_buy:id',
'base_id': 'inbound_buy:base_id',
'global_print_id': 'inbound_buy:global_print_id',
'global_print_id_str': 'inbound_buy:global_print_id',
'company_name': 'inbound_buy:company_name',
'material_name': 'inbound_buy:material_name',
'spec_model': 'inbound_buy:spec_model',
'category': 'inbound_buy:category',
'unit': 'inbound_buy:unit',
'material_type': 'inbound_buy:material_type',
'isInspectionRequired': 'inbound_buy:isInspectionRequired',
# 入库身份
'sku': 'inbound_buy:sku',
'inbound_date': 'inbound_buy:inbound_date',
'barcode': 'inbound_buy:barcode',
'serial_number': 'inbound_buy:sn_bn',
'batch_number': 'inbound_buy:sn_bn',
'warehouse_loc': 'inbound_buy:warehouse_loc',
'warehouse_location': 'inbound_buy:warehouse_loc',
# 状态
'status': 'inbound_buy:status',
'inspection_status': 'inbound_buy:inspection_status',
# 数量(对齐 sys_element 中实际存在的权限码)
'in_quantity': 'inbound_buy:qty_inbound',
'qty_inbound': 'inbound_buy:qty_inbound',
'stock_quantity': 'inbound_buy:qty_stock',
'qty_stock': 'inbound_buy:qty_stock',
'available_quantity': 'inbound_buy:qty_available',
'qty_available': 'inbound_buy:qty_available',
# 价格
'unit_price': 'inbound_buy:unit_price',
'post_tax_unit_price': 'inbound_buy:unit_price',
'total_price': 'inbound_buy:total_price',
'tax_rate': 'inbound_buy:tax_rate',
'currency': 'inbound_buy:currency',
'exchange_rate': 'inbound_buy:exchange_rate',
# 商务
'supplier_name': 'inbound_buy:supplier_name',
'purchaser': 'inbound_buy:purchaser',
'purchaser_email': 'inbound_buy:purchaser_email',
'source_link': 'inbound_buy:source_link',
'detail_link': 'inbound_buy:detail_link',
# 图片/附件
'arrival_photo': 'inbound_buy:arrival_photo',
'inspection_report': 'inbound_buy:inspection_report',
# 采购单关联
'request_id': 'inbound_buy:request_id',
'request_no': 'inbound_buy:request_no',
}
# 通配符(SUPER_ADMIN)不过滤
if 'inbound_buy:*' in user_permissions:
return item_dict
for field, perm_code in field_to_perm.items():
base_perm_code = perm_code.split(':')[-1] if ':' in perm_code else perm_code
if field in item_dict and perm_code not in user_permissions and base_perm_code not in user_permissions:
item_dict[field] = None
return item_dict
# 如果用户是超级管理员且有 'inbound_buy:*',则不过滤
if 'inbound_buy:*' in user_permissions:
return item_dict
for field, perm_code in field_to_perm.items():
# 提取不带前缀的基础权限码(如 'serial_number')
base_perm_code = perm_code.split(':')[-1] if ':' in perm_code else perm_code
# 如果用户的权限列表中,既没有长格式,也没有短格式,才将字段设为 None
if field in item_dict and perm_code not in user_permissions and base_perm_code not in user_permissions:
item_dict[field] = None
return item_dict
"""严格 Default Deny 字段过滤 (see app/utils/field_permissions.py)"""
from app.utils.field_permissions import apply_strict_rbac
return apply_strict_rbac(item_dict, 'StockBuy', user_permissions)
# ------------------------------------------------------------------
@ -196,53 +126,13 @@ def submit():
if not data:
return jsonify({"code": 400, "msg": "No data"}), 400
# 数据清洗:移除用户没有权限的字段
# ★ 白名单模式:仅过滤价格敏感字段,其余全部放行
user_permissions = get_current_user_permissions()
# 超级管理员不过滤
if 'inbound_buy:*' not in user_permissions:
# 字段名到权限码的映射(与前端 permissionMap 保持一致)
field_to_perm = {
'id': 'inbound_buy:id',
'base_id': 'inbound_buy:base_id',
'global_print_id': 'inbound_buy:global_print_id',
'sku': 'inbound_buy:sku',
'barcode': 'inbound_buy:barcode',
'in_date': 'inbound_buy:in_date',
'serial_number': 'inbound_buy:serial_number',
'batch_number': 'inbound_buy:batch_number',
'status': 'inbound_buy:status',
'in_quantity': 'inbound_buy:in_quantity',
'stock_quantity': 'inbound_buy:stock_quantity',
'available_quantity': 'inbound_buy:available_quantity',
'inspection_status': 'inbound_buy:inspection_status',
'warehouse_location': 'inbound_buy:warehouse_location',
'unit_price': 'inbound_buy:unit_price',
'post_tax_unit_price': 'inbound_buy:post_tax_unit_price',
'tax_rate': 'inbound_buy:tax_rate',
'total_price': 'inbound_buy:total_price',
'currency': 'inbound_buy:currency',
'exchange_rate': 'inbound_buy:exchange_rate',
'supplier_name': 'inbound_buy:supplier_name',
'buyer_name': 'inbound_buy:buyer_name',
'buyer_email': 'inbound_buy:buyer_email',
'original_link': 'inbound_buy:original_link',
'detail_link': 'inbound_buy:detail_link',
'arrival_photo': 'inbound_buy:arrival_photo',
'inspection_report': 'inbound_buy:inspection_report',
'material_name': 'inbound_buy:material_name',
'spec_model': 'inbound_buy:spec_model',
'category': 'inbound_buy:category',
'unit': 'inbound_buy:unit',
'material_type': 'inbound_buy:material_type',
'company_name': 'inbound_buy:company_name',
}
# 复制一份,避免遍历时修改字典
for field in list(data.keys()):
perm_code = field_to_perm.get(field)
# 提取不带前缀的基础权限码(如 'serial_number')
base_perm_code = perm_code.split(':')[-1] if ':' in perm_code else perm_code
# 如果用户的权限列表中,既没有长格式,也没有短格式,才移除该字段
if perm_code and perm_code not in user_permissions and base_perm_code not in user_permissions:
for field in ('unit_price', 'post_tax_unit_price', 'tax_rate', 'total_price',
'currency', 'exchange_rate'):
perm_code = f'inbound_buy:{field}'
if field in data and perm_code not in user_permissions:
data.pop(field, None)
# 库位必填校验(安全兜底)

View File

@ -14,33 +14,15 @@ def get_current_user_permissions():
user_role = claims.get('role')
user_company = claims.get('company_name', '')
if not user_role: return []
if user_role.upper() == 'SUPER_ADMIN': return ['inbound_product:*']
from app.utils.constants import UserRole
if str(user_role).strip().upper() == UserRole.SUPER_ADMIN: return ['inbound_product:*']
perm_dict = AuthService.get_user_permissions(user_role, company_name=user_company)
return perm_dict.get('menus', []) + perm_dict.get('elements', [])
def filter_item_by_permissions(item_dict, user_permissions):
"""根据用户权限过滤字段,无权限的字段值置为 None"""
field_to_perm = {
'id': 'inbound_product:id', 'base_id': 'inbound_product:base_id', 'company_name': 'inbound_product:company_name',
'material_name': 'inbound_product:material_name', 'category': 'inbound_product:category',
'material_type': 'inbound_product:material_type', 'spec_model': 'inbound_product:spec_model',
'unit': 'inbound_product:unit', 'sku': 'inbound_product:sku', 'inbound_date': 'inbound_product:inbound_date',
'barcode': 'inbound_product:barcode', 'serial_number': 'inbound_product:serial_number',
'status': 'inbound_product:status', 'quality_status': 'inbound_product:quality_status',
'in_quantity': 'inbound_product:qty_inbound', 'stock_quantity': 'inbound_product:qty_stock',
'available_quantity': 'inbound_product:qty_available', 'warehouse_location': 'inbound_product:warehouse_loc',
'bom_code': 'inbound_product:bom_code', 'bom_version': 'inbound_product:bom_version',
'work_order_code': 'inbound_product:work_order_code', 'order_id': 'inbound_product:order_id',
'production_manager': 'inbound_product:production_manager', 'production_start_time': 'inbound_product:production_start_time',
'production_end_time': 'inbound_product:production_end_time', 'raw_material_cost': 'inbound_product:raw_material_cost',
'manual_cost': 'inbound_product:manual_cost', 'sale_price': 'inbound_product:sale_price',
'product_photo': 'inbound_product:product_photo', 'quality_report_link': 'inbound_product:quality_report_link',
'inspection_report_link': 'inbound_product:inspection_report_link', 'detail_link': 'inbound_product:detail_link',
}
if 'inbound_product:*' in user_permissions: return item_dict
for field, perm_code in field_to_perm.items():
if field in item_dict and perm_code not in user_permissions: item_dict[field] = None
return item_dict
"""严格 Default Deny 字段过滤 (see app/utils/field_permissions.py)"""
from app.utils.field_permissions import apply_strict_rbac
return apply_strict_rbac(item_dict, 'StockProduct', user_permissions)
@inbound_product_bp.route('/search-base', methods=['GET'])
@permission_required('inbound_product')
@ -144,7 +126,8 @@ def submit():
user_permissions = get_current_user_permissions()
if 'inbound_product:*' not in user_permissions:
field_to_perm = {'id': 'inbound_product:id', 'base_id': 'inbound_product:base_id', 'company_name': 'inbound_product:company_name', 'material_name': 'inbound_product:material_name', 'category': 'inbound_product:category', 'material_type': 'inbound_product:material_type', 'spec_model': 'inbound_product:spec_model', 'unit': 'inbound_product:unit', 'sku': 'inbound_product:sku', 'inbound_date': 'inbound_product:inbound_date', 'barcode': 'inbound_product:barcode', 'serial_number': 'inbound_product:serial_number', 'status': 'inbound_product:status', 'quality_status': 'inbound_product:quality_status', 'in_quantity': 'inbound_product:in_quantity', 'stock_quantity': 'inbound_product:stock_quantity', 'available_quantity': 'inbound_product:available_quantity', 'warehouse_location': 'inbound_product:warehouse_location', 'bom_code': 'inbound_product:bom_code', 'bom_version': 'inbound_product:bom_version', 'work_order_code': 'inbound_product:work_order_code', 'order_id': 'inbound_product:order_id', 'production_manager': 'inbound_product:production_manager', 'production_start_time': 'inbound_product:production_start_time', 'production_end_time': 'inbound_product:production_end_time', 'raw_material_cost': 'inbound_product:raw_material_cost', 'manual_cost': 'inbound_product:manual_cost', 'sale_price': 'inbound_product:sale_price', 'product_photo': 'inbound_product:product_photo', 'quality_report_link': 'inbound_product:quality_report_link', 'inspection_report_link': 'inbound_product:inspection_report_link', 'detail_link': 'inbound_product:detail_link'}
# ★ 基本操作字段(库位/数量等)不参与字段权限过滤
field_to_perm = {'id': 'inbound_product:id', 'company_name': 'inbound_product:company_name', 'material_name': 'inbound_product:material_name', 'category': 'inbound_product:category', 'material_type': 'inbound_product:material_type', 'spec_model': 'inbound_product:spec_model', 'unit': 'inbound_product:unit', 'sku': 'inbound_product:sku', 'inbound_date': 'inbound_product:inbound_date', 'barcode': 'inbound_product:barcode', 'serial_number': 'inbound_product:serial_number', 'status': 'inbound_product:status', 'quality_status': 'inbound_product:quality_status', 'bom_code': 'inbound_product:bom_code', 'bom_version': 'inbound_product:bom_version', 'work_order_code': 'inbound_product:work_order_code', 'order_id': 'inbound_product:order_id', 'production_manager': 'inbound_product:production_manager', 'production_start_time': 'inbound_product:production_start_time', 'production_end_time': 'inbound_product:production_end_time', 'raw_material_cost': 'inbound_product:raw_material_cost', 'manual_cost': 'inbound_product:manual_cost', 'sale_price': 'inbound_product:sale_price', 'product_photo': 'inbound_product:product_photo', 'quality_report_link': 'inbound_product:quality_report_link', 'inspection_report_link': 'inbound_product:inspection_report_link', 'detail_link': 'inbound_product:detail_link'}
for field in list(data.keys()):
perm_code = field_to_perm.get(field)
if perm_code and perm_code not in user_permissions: data.pop(field, None)
@ -171,7 +154,7 @@ def update(id):
data = request.get_json()
user_permissions = get_current_user_permissions()
if 'inbound_product:*' not in user_permissions:
field_to_perm = {'id': 'inbound_product:id', 'base_id': 'inbound_product:base_id', 'company_name': 'inbound_product:company_name', 'material_name': 'inbound_product:material_name', 'category': 'inbound_product:category', 'material_type': 'inbound_product:material_type', 'spec_model': 'inbound_product:spec_model', 'unit': 'inbound_product:unit', 'sku': 'inbound_product:sku', 'inbound_date': 'inbound_product:inbound_date', 'barcode': 'inbound_product:barcode', 'serial_number': 'inbound_product:serial_number', 'status': 'inbound_product:status', 'quality_status': 'inbound_product:quality_status', 'in_quantity': 'inbound_product:in_quantity', 'stock_quantity': 'inbound_product:stock_quantity', 'available_quantity': 'inbound_product:available_quantity', 'warehouse_location': 'inbound_product:warehouse_location', 'bom_code': 'inbound_product:bom_code', 'bom_version': 'inbound_product:bom_version', 'work_order_code': 'inbound_product:work_order_code', 'order_id': 'inbound_product:order_id', 'production_manager': 'inbound_product:production_manager', 'production_start_time': 'inbound_product:production_start_time', 'production_end_time': 'inbound_product:production_end_time', 'raw_material_cost': 'inbound_product:raw_material_cost', 'manual_cost': 'inbound_product:manual_cost', 'sale_price': 'inbound_product:sale_price', 'product_photo': 'inbound_product:product_photo', 'quality_report_link': 'inbound_product:quality_report_link', 'inspection_report_link': 'inbound_product:inspection_report_link', 'detail_link': 'inbound_product:detail_link'}
field_to_perm = {'id': 'inbound_product:id', 'company_name': 'inbound_product:company_name', 'material_name': 'inbound_product:material_name', 'category': 'inbound_product:category', 'material_type': 'inbound_product:material_type', 'spec_model': 'inbound_product:spec_model', 'unit': 'inbound_product:unit', 'sku': 'inbound_product:sku', 'inbound_date': 'inbound_product:inbound_date', 'barcode': 'inbound_product:barcode', 'serial_number': 'inbound_product:serial_number', 'status': 'inbound_product:status', 'quality_status': 'inbound_product:quality_status', 'in_quantity': 'inbound_product:in_quantity', 'stock_quantity': 'inbound_product:stock_quantity', 'available_quantity': 'inbound_product:available_quantity', 'bom_code': 'inbound_product:bom_code', 'bom_version': 'inbound_product:bom_version', 'work_order_code': 'inbound_product:work_order_code', 'order_id': 'inbound_product:order_id', 'production_manager': 'inbound_product:production_manager', 'production_start_time': 'inbound_product:production_start_time', 'production_end_time': 'inbound_product:production_end_time', 'raw_material_cost': 'inbound_product:raw_material_cost', 'manual_cost': 'inbound_product:manual_cost', 'sale_price': 'inbound_product:sale_price', 'product_photo': 'inbound_product:product_photo', 'quality_report_link': 'inbound_product:quality_report_link', 'inspection_report_link': 'inbound_product:inspection_report_link', 'detail_link': 'inbound_product:detail_link'}
for field in list(data.keys()):
perm_code = field_to_perm.get(field)
if perm_code and perm_code not in user_permissions: data.pop(field, None)

View File

@ -14,33 +14,15 @@ def get_current_user_permissions():
user_role = claims.get('role')
user_company = claims.get('company_name', '')
if not user_role: return []
if user_role.upper() == 'SUPER_ADMIN': return ['inbound_semi:*']
from app.utils.constants import UserRole
if str(user_role).strip().upper() == UserRole.SUPER_ADMIN: return ['inbound_semi:*']
perm_dict = AuthService.get_user_permissions(user_role, company_name=user_company)
return perm_dict.get('menus', []) + perm_dict.get('elements', [])
def filter_item_by_permissions(item_dict, user_permissions):
"""根据用户权限过滤字段,无权限的字段值置为 None"""
field_to_perm = {
'id': 'inbound_semi:id', 'base_id': 'inbound_semi:base_id', 'company_name': 'inbound_semi:company_name',
'material_name': 'inbound_semi:material_name', 'category': 'inbound_semi:category',
'material_type': 'inbound_semi:material_type', 'spec_model': 'inbound_semi:spec_model',
'unit': 'inbound_semi:unit', 'sku': 'inbound_semi:sku', 'inbound_date': 'inbound_semi:inbound_date',
'barcode': 'inbound_semi:barcode', 'serial_number': 'inbound_semi:serial_number',
'batch_number': 'inbound_semi:batch_number', 'status': 'inbound_semi:status',
'quality_status': 'inbound_semi:quality_status', 'in_quantity': 'inbound_semi:qty_inbound',
'stock_quantity': 'inbound_semi:qty_stock', 'available_quantity': 'inbound_semi:qty_available',
'warehouse_location': 'inbound_semi:warehouse_loc', 'bom_code': 'inbound_semi:bom_code',
'bom_version': 'inbound_semi:bom_version', 'work_order_code': 'inbound_semi:work_order_code',
'raw_material_cost': 'inbound_semi:raw_material_cost', 'manual_cost': 'inbound_semi:manual_cost',
'unit_total_cost': 'inbound_semi:unit_total_cost', 'production_manager': 'inbound_semi:production_manager',
'production_start_time': 'inbound_semi:production_start_time', 'production_end_time': 'inbound_semi:production_end_time',
'arrival_photo': 'inbound_semi:arrival_photo', 'quality_report_link': 'inbound_semi:quality_report_link',
'detail_link': 'inbound_semi:detail_link',
}
if 'inbound_semi:*' in user_permissions: return item_dict
for field, perm_code in field_to_perm.items():
if field in item_dict and perm_code not in user_permissions: item_dict[field] = None
return item_dict
"""严格 Default Deny 字段过滤 (see app/utils/field_permissions.py)"""
from app.utils.field_permissions import apply_strict_rbac
return apply_strict_rbac(item_dict, 'StockSemi', user_permissions)
@inbound_semi_bp.route('/search-base', methods=['GET'])
@permission_required('inbound_semi')
@ -139,10 +121,10 @@ def submit():
user_permissions = get_current_user_permissions()
if 'inbound_semi:*' not in user_permissions:
field_to_perm = {'id': 'inbound_semi:id', 'base_id': 'inbound_semi:base_id', 'company_name': 'inbound_semi:company_name', 'material_name': 'inbound_semi:material_name', 'category': 'inbound_semi:category', 'material_type': 'inbound_semi:material_type', 'spec_model': 'inbound_semi:spec_model', 'unit': 'inbound_semi:unit', 'sku': 'inbound_semi:sku', 'inbound_date': 'inbound_semi:inbound_date', 'barcode': 'inbound_semi:barcode', 'serial_number': 'inbound_semi:serial_number', 'batch_number': 'inbound_semi:batch_number', 'status': 'inbound_semi:status', 'quality_status': 'inbound_semi:quality_status', 'in_quantity': 'inbound_semi:in_quantity', 'stock_quantity': 'inbound_semi:stock_quantity', 'available_quantity': 'inbound_semi:available_quantity', 'warehouse_location': 'inbound_semi:warehouse_location', 'bom_code': 'inbound_semi:bom_code', 'bom_version': 'inbound_semi:bom_version', 'work_order_code': 'inbound_semi:work_order_code', 'raw_material_cost': 'inbound_semi:raw_material_cost', 'manual_cost': 'inbound_semi:manual_cost', 'unit_total_cost': 'inbound_semi:unit_total_cost', 'production_manager': 'inbound_semi:production_manager', 'production_start_time': 'inbound_semi:production_start_time', 'production_end_time': 'inbound_semi:production_end_time', 'arrival_photo': 'inbound_semi:arrival_photo', 'quality_report_link': 'inbound_semi:quality_report_link', 'detail_link': 'inbound_semi:detail_link'}
for field in list(data.keys()):
perm_code = field_to_perm.get(field)
if perm_code and perm_code not in user_permissions: data.pop(field, None)
for field in ('raw_material_cost', 'manual_cost', 'unit_total_cost', 'total_price'):
perm_code = f'inbound_semi:{field}'
if field in data and perm_code not in user_permissions:
data.pop(field, None)
new_stock = SemiInboundService.handle_inbound(data)
# ★ Fail-Closed: 入库成功响应剥离成本字段
resp = new_stock.to_dict()
@ -166,10 +148,10 @@ def update_semi(id):
data = request.get_json()
user_permissions = get_current_user_permissions()
if 'inbound_semi:*' not in user_permissions:
field_to_perm = {'id': 'inbound_semi:id', 'base_id': 'inbound_semi:base_id', 'company_name': 'inbound_semi:company_name', 'material_name': 'inbound_semi:material_name', 'category': 'inbound_semi:category', 'material_type': 'inbound_semi:material_type', 'spec_model': 'inbound_semi:spec_model', 'unit': 'inbound_semi:unit', 'sku': 'inbound_semi:sku', 'inbound_date': 'inbound_semi:inbound_date', 'barcode': 'inbound_semi:barcode', 'serial_number': 'inbound_semi:serial_number', 'batch_number': 'inbound_semi:batch_number', 'status': 'inbound_semi:status', 'quality_status': 'inbound_semi:quality_status', 'in_quantity': 'inbound_semi:in_quantity', 'stock_quantity': 'inbound_semi:stock_quantity', 'available_quantity': 'inbound_semi:available_quantity', 'warehouse_location': 'inbound_semi:warehouse_location', 'bom_code': 'inbound_semi:bom_code', 'bom_version': 'inbound_semi:bom_version', 'work_order_code': 'inbound_semi:work_order_code', 'raw_material_cost': 'inbound_semi:raw_material_cost', 'manual_cost': 'inbound_semi:manual_cost', 'unit_total_cost': 'inbound_semi:unit_total_cost', 'production_manager': 'inbound_semi:production_manager', 'production_start_time': 'inbound_semi:production_start_time', 'production_end_time': 'inbound_semi:production_end_time', 'arrival_photo': 'inbound_semi:arrival_photo', 'quality_report_link': 'inbound_semi:quality_report_link', 'detail_link': 'inbound_semi:detail_link'}
for field in list(data.keys()):
perm_code = field_to_perm.get(field)
if perm_code and perm_code not in user_permissions: data.pop(field, None)
for field in ('raw_material_cost', 'manual_cost', 'unit_total_cost', 'total_price'):
perm_code = f'inbound_semi:{field}'
if field in data and perm_code not in user_permissions:
data.pop(field, None)
SemiInboundService.update_inbound(id, data)
return jsonify({"code": 200, "msg": "更新成功"})
except Exception as e:

View File

@ -1,4 +1,4 @@
from flask import Blueprint, request, jsonify
from flask import Blueprint, request, jsonify, current_app
from app.services.outbound_service import OutboundService
from flask_jwt_extended import jwt_required, get_jwt_identity, get_jwt
from app.utils.decorators import permission_required, audit_log, prevent_double_submit
@ -117,18 +117,24 @@ def scan_barcode():
get_target_name_fn=lambda: request.get_json().get('order_no') if request.get_json() else None
)
def create_outbound():
# 权限检查:需要 outbound_create:operation 或 outbound_selection:operation 之一
# 权限检查:有 outbound_selection 菜单或操作权限即可提交
claims = get_jwt()
user_role = claims.get('role')
user_company = claims.get('company_name', '')
if not user_role:
return jsonify({'code': 403, 'msg': '未授权'}), 403
# 超级管理员直接放行
if user_role.upper() != 'SUPER_ADMIN':
perm_dict = AuthService.get_user_permissions(user_role, company_name=user_company)
perms = perm_dict.get('menus', []) + perm_dict.get('elements', [])
if ('outbound_create:operation' not in perms) and ('outbound_selection:operation' not in perms):
outbound_perms = [p for p in perms if 'outbound' in p.lower() or 'selection' in p.lower() or 'create' in p.lower()]
current_app.logger.warning(
f"[出库权限调试] role={user_role}, company={user_company}, "
f"出库相关权限={outbound_perms}, 全部权限数={len(perms)}"
)
if 'outbound_selection' not in perms and not any(
p.startswith('outbound_selection:') or p.startswith('outbound_create:') for p in perms
):
return jsonify({'code': 403, 'msg': '权限不足'}), 403
data = request.get_json()
@ -340,7 +346,7 @@ def get_current_user_info():
# --------------------------------------------------------
@outbound_bp.route('/request', methods=['POST'])
@jwt_required()
@permission_required('outbound_approval')
@permission_required('outbound_selection')
def create_outbound_request():
"""
创建出库审批单(申请阶段,用户只需提交宏观物料信息,无需关联具体库存记录)

View File

@ -165,7 +165,7 @@ def get_records():
# --- 提交借库申请 ---
@trans_bp.route('/borrow/request', methods=['POST'])
@jwt_required()
@permission_required('op_borrow_approval')
@permission_required('op_borrow_apply')
def submit_borrow_request():
"""
提交借库申请(仅存储意向,不扣库存)

View File

@ -102,6 +102,11 @@ class PermissionService:
else:
element_codes.append(p.target_code)
# ★ 诊断:打印入库操作权限
inbound_ops = [c for c in element_codes if 'inbound_buy' in c or 'inbound_semi' in c or 'inbound_product' in c]
if inbound_ops:
print(f"[诊断] {role_code} 拥有的入库操作权限: {inbound_ops}")
return {
'menus': menu_codes,
'elements': element_codes
@ -637,7 +642,19 @@ class PermissionService:
)
db.session.add(new_perm)
db.session.commit()
# ★ 入库模块操作权限元素(之前缺失导致"可编辑"勾了也不能入库)
inbound_op_elements = [
('inbound_buy', 'inbound_buy:operation', '可编辑', 'operation'),
('inbound_semi', 'inbound_semi:operation', '可编辑', 'operation'),
('inbound_product', 'inbound_product:operation', '可编辑', 'operation'),
('inbound_service', 'inbound_service:operation', '可编辑', 'operation'),
]
for menu_code, code, name, etype in inbound_op_elements:
existing = SysElement.query.filter_by(menu_code=menu_code, code=code).first()
if not existing:
db.session.add(SysElement(menu_code=menu_code, name=name, code=code, element_type=etype))
print(f"✅ 入库操作元素已创建: {code}")
# ★ 采购申请权限元素
purchase_elements = [
('inbound_purchase:operation', '可编辑', 'operation'),
@ -656,6 +673,17 @@ class PermissionService:
))
print(f"✅ 采购申请元素已创建: {code}")
db.session.commit()
# ★ 诊断:打印入库模块的元素和权限状态
for mc in ('inbound_buy', 'inbound_semi', 'inbound_product', 'inbound_service'):
elems = SysElement.query.filter_by(menu_code=mc).all()
codes = [e.code for e in elems]
print(f"[诊断] {mc} 的 sys_element: {codes}")
role_count = SysRolePermission.query.filter_by(
target_code=f'{mc}:operation', type='element'
).count()
print(f"[诊断] {mc}:operation 已分配给 {role_count} 个角色")
print(f"✅ 所有菜单初始化完成")
return True

View File

@ -0,0 +1,123 @@
"""
入库模块字段级权限严格映射 (Default Deny)
规则:
- None = 基础字段,始终保留
- 字符串 = 需要用户拥有该精确权限码,否则移除
- 不在映射中的字段 → 立即删除 (Default Deny)
"""
STOCK_FIELD_RBAC_MAPPING = {
"MaterialBase": {
"id": None, "isEnabled": None, "visibilityLevel": None,
"name": "material_list:name", "commonName": "material_list:commonName",
"category": "material_list:category", "type": "material_list:type",
"spec": "material_list:spec", "unit": "material_list:unit",
"companyName": "material_list:companyName", "isInspectionRequired": "material_list:isInspectionRequired",
"generalImage": "material_list:files", "generalManual": "material_list:files",
"productImageRemark": "material_list:productImageRemark",
"manualLinkRemark": "material_list:manualLinkRemark",
"referencePrice": "material_list:referencePrice",
"inventoryCount": "material_list:inventoryCount",
"availableCount": "material_list:availableCount",
},
"StockBuy": {
"id": None, "request_id": None, "request_no": None,
"in_quantity": None, "stock_quantity": None, "available_quantity": None,
"warehouse_loc": None, "status": None, "global_print_id": None,
"company_name": "material_list:companyName", "material_name": "material_list:name",
"spec_model": "material_list:spec", "category": "material_list:category",
"unit": "material_list:unit", "material_type": "material_list:type",
"isInspectionRequired": "material_list:isInspectionRequired",
"sku": "inbound_buy:sku", "inbound_date": "inbound_buy:inbound_date",
"barcode": "inbound_buy:barcode", "serial_number": "inbound_buy:sn_bn",
"batch_number": "inbound_buy:sn_bn", "inspection_status": "inbound_buy:inspection_status",
"unit_price": "inbound_buy:unit_price", "post_tax_unit_price": "inbound_buy:post_tax_unit_price",
"total_price": "inbound_buy:total_price", "tax_rate": "inbound_buy:tax_rate",
"currency": "inbound_buy:currency", "exchange_rate": "inbound_buy:exchange_rate",
"supplier_name": "inbound_buy:supplier_name", "purchaser": "inbound_buy:purchaser",
"purchaser_email": "inbound_buy:purchaser_email", "source_link": "inbound_buy:source_link",
"detail_link": "inbound_buy:detail_link", "arrival_photo": "inbound_buy:arrival_photo",
"inspection_report": "inbound_buy:inspection_report",
},
"StockSemi": {
"id": None, "in_quantity": None, "stock_quantity": None, "available_quantity": None,
"warehouse_loc": None, "status": None, "global_print_id": None,
"company_name": "material_list:companyName", "material_name": "material_list:name",
"spec_model": "material_list:spec", "category": "material_list:category",
"unit": "material_list:unit", "material_type": "material_list:type",
"sku": "inbound_semi:sku", "inbound_date": "inbound_semi:inbound_date",
"barcode": "inbound_semi:barcode", "serial_number": "inbound_semi:sn_bn",
"batch_number": "inbound_semi:sn_bn", "bom_code": "inbound_semi:bom_code",
"bom_version": "inbound_semi:bom_version", "work_order_code": "inbound_semi:work_order_code",
"raw_material_cost": "inbound_semi:raw_material_cost", "manual_cost": "inbound_semi:manual_cost",
"unit_total_cost": "inbound_semi:unit_total_cost", "total_price": "inbound_semi:total_price",
"production_manager": "inbound_semi:production_manager",
"production_time_range": "inbound_semi:production_time_range",
"production_start_time": "inbound_semi:production_start_time",
"production_end_time": "inbound_semi:production_end_time",
"quality_status": "inbound_semi:quality_status", "quality_report_link": "inbound_semi:quality_report_link",
"arrival_photo": "inbound_semi:arrival_photo", "remark": "inbound_semi:remark",
"detail_link": "inbound_semi:detail_link",
},
"StockProduct": {
"id": None, "in_quantity": None, "stock_quantity": None, "available_quantity": None,
"warehouse_loc": None, "status": None, "global_print_id": None,
"company_name": "material_list:companyName", "material_name": "material_list:name",
"spec_model": "material_list:spec", "category": "material_list:category",
"unit": "material_list:unit", "material_type": "material_list:type",
"sku": "inbound_product:sku", "inbound_date": "inbound_product:inbound_date",
"barcode": "inbound_product:barcode", "serial_number": "inbound_product:serial_number",
"bom_code": "inbound_product:bom_code", "bom_version": "inbound_product:bom_version",
"work_order_code": "inbound_product:work_order_code",
"raw_material_cost": "inbound_product:raw_material_cost", "manual_cost": "inbound_product:manual_cost",
"unit_total_cost": "inbound_product:unit_total_cost",
"production_manager": "inbound_product:production_manager",
"production_time_range": "inbound_product:production_time_range",
"quality_status": "inbound_product:quality_status",
"quality_report_link": "inbound_product:quality_report_link",
"inspection_report_link": "inbound_product:inspection_report_link",
"sale_price": "inbound_product:sale_price", "order_id": "inbound_product:order_id",
"product_photo": "inbound_product:product_photo", "remark": "inbound_product:remark",
"detail_link": "inbound_product:detail_link",
},
}
def _is_super_admin(user_permissions: list) -> bool:
"""判断是否为超级管理员(支持 * 和 module:* 两种通配符)"""
return '*' in user_permissions or any(p.endswith(':*') for p in user_permissions)
def apply_strict_rbac(item_dict: dict, table_name: str, user_permissions: list) -> dict:
"""
Default Deny 字段过滤器:
1. 不在映射中的 key → 立即删除
2. 在映射中但需要权限且用户没有 → 删除
3. 在映射中且为 None 或用户有权限 → 保留
"""
mapping = STOCK_FIELD_RBAC_MAPPING.get(table_name)
if not mapping:
return item_dict # 未知表不做过滤
# 超级管理员放行(支持 * / module:* 通配符)
if _is_super_admin(user_permissions):
for key in list(item_dict.keys()):
if key not in mapping:
del item_dict[key]
return item_dict
for key in list(item_dict.keys()):
if key not in mapping:
del item_dict[key] # Default Deny
else:
perm_code = mapping[key]
if perm_code is not None and perm_code not in user_permissions:
if isinstance(item_dict[key], (int, float)):
item_dict[key] = 0
elif isinstance(item_dict[key], bool):
item_dict[key] = False
else:
item_dict[key] = None
return item_dict