fix: Fail-Closed 字段级安全加固 — 堵住15+端点价格泄露

## stock.py
- _make_price_stripper(): 工厂函数,选单前缀自动剥离所有价格/成本字段
- _do_get_stock_list(permission_prefix): 每个item.to_dict()后剥离价格
- /all 端点: 非AI模式自动剥离价格字段
- /list 端点: permission_prefix='outbound_selection' 传递

## outbound.py
- /bom-match-stock: 返回前按stock_type剥离全部价格/成本字段
- /scan: result.pop('price', None)

## scrap.py
- /scan: result.pop('price', None)
- /records: 每条记录剥离 cost_at_scrap, total_loss

## transactions.py
- filter_item_by_permissions: 从空字典恢复完整20字段映射
- /borrow/stock-list: permission_prefix='op_borrow_apply' 传递

## buy.py / semi.py / product.py
- submit 成功响应剥离所有价格/成本字段(不泄露给前端)

## bom.py
- /base/list: 剥离 referencePrice
This commit is contained in:
yueli
2026-07-16 11:26:05 +08:00
parent 128985af17
commit c07f25b646
8 changed files with 127 additions and 27 deletions

View File

@ -370,9 +370,14 @@ def get_material_base_list():
page=page, per_page=limit, error_out=False page=page, per_page=limit, error_out=False
) )
# 构建返回数据 # 构建返回数据 — ★ Fail-Closed: 剥离 referencePrice
items = []
for item in pagination.items:
d = item.to_dict()
d.pop('referencePrice', None)
items.append(d)
data = { data = {
'list': [item.to_dict() for item in pagination.items], 'list': items,
'total': pagination.total 'total': pagination.total
} }

View File

@ -252,10 +252,16 @@ def submit():
new_stock = BuyInboundService.handle_inbound(data) new_stock = BuyInboundService.handle_inbound(data)
# ★ Fail-Closed: 入库成功响应剥离价格字段
resp = new_stock.to_dict()
for k in ('unit_price', 'post_tax_unit_price', 'total_price',
'tax_rate', 'currency', 'exchange_rate', 'pre_tax_unit_price'):
resp.pop(k, None)
return jsonify({ return jsonify({
"code": 200, "code": 200,
"msg": "入库成功", "msg": "入库成功",
"data": new_stock.to_dict() "data": resp
}) })
except Exception as e: except Exception as e:
traceback.print_exc() traceback.print_exc()

View File

@ -149,7 +149,11 @@ def submit():
perm_code = field_to_perm.get(field) perm_code = field_to_perm.get(field)
if perm_code and perm_code not in user_permissions: data.pop(field, None) if perm_code and perm_code not in user_permissions: data.pop(field, None)
new_stock = ProductInboundService.handle_inbound(data) new_stock = ProductInboundService.handle_inbound(data)
return jsonify({"code": 200, "msg": "入库成功", "data": new_stock.to_dict()}) # ★ Fail-Closed: 入库成功响应剥离成本/售价字段
resp = new_stock.to_dict()
for k in ('raw_material_cost', 'manual_cost', 'unit_total_cost', 'sale_price'):
resp.pop(k, None)
return jsonify({"code": 200, "msg": "入库成功", "data": resp})
except Exception as e: except Exception as e:
traceback.print_exc() traceback.print_exc()
return jsonify({"code": 500, "msg": str(e)}), 500 return jsonify({"code": 500, "msg": str(e)}), 500

View File

@ -144,7 +144,11 @@ def submit():
perm_code = field_to_perm.get(field) perm_code = field_to_perm.get(field)
if perm_code and perm_code not in user_permissions: data.pop(field, None) if perm_code and perm_code not in user_permissions: data.pop(field, None)
new_stock = SemiInboundService.handle_inbound(data) new_stock = SemiInboundService.handle_inbound(data)
return jsonify({"code": 200, "msg": "入库成功", "data": new_stock.to_dict()}) # ★ Fail-Closed: 入库成功响应剥离成本字段
resp = new_stock.to_dict()
for k in ('raw_material_cost', 'manual_cost', 'unit_total_cost', 'total_price'):
resp.pop(k, None)
return jsonify({"code": 200, "msg": "入库成功", "data": resp})
except Exception as e: except Exception as e:
traceback.print_exc() traceback.print_exc()
return jsonify({"code": 500, "msg": str(e)}), 500 return jsonify({"code": 500, "msg": str(e)}), 500

View File

@ -150,8 +150,11 @@ def get_all_stock():
支持 AI 极简模式: ?ai_mode=true 支持 AI 极简模式: ?ai_mode=true
- 只返回 name / spec / availableQuantity 三个字段 - 只返回 name / spec / availableQuantity 三个字段
- 键名压缩为 n / s / c - 键名压缩为 n / s / c
★ Fail-Closed: 非 AI 模式自动剥离所有价格成本字段
""" """
ai_mode = request.args.get('ai_mode', '').lower() == 'true' ai_mode = request.args.get('ai_mode', '').lower() == 'true'
_strip = _make_price_stripper('inventory_stocktake')
try: try:
all_items = [] all_items = []
@ -170,7 +173,9 @@ def get_all_stock():
'c': float(item.available_quantity or 0) 'c': float(item.available_quantity or 0)
}) })
else: else:
all_items.append(item.to_dict()) d = item.to_dict()
_strip(d, 'buy')
all_items.append(d)
# 2. 半成品 # 2. 半成品
if StockSemi: if StockSemi:
@ -187,7 +192,9 @@ def get_all_stock():
'c': float(item.available_quantity or 0) 'c': float(item.available_quantity or 0)
}) })
else: else:
all_items.append(item.to_dict()) d = item.to_dict()
_strip(d, 'semi')
all_items.append(d)
except Exception: except Exception:
pass pass
@ -206,7 +213,9 @@ def get_all_stock():
'c': float(item.available_quantity or 0) 'c': float(item.available_quantity or 0)
}) })
else: else:
all_items.append(item.to_dict()) d = item.to_dict()
_strip(d, 'product')
all_items.append(d)
except Exception: except Exception:
pass pass
@ -219,9 +228,41 @@ def get_all_stock():
# ============================================================================== # ==============================================================================
# 分页库存查询接口(服务端分页,出库/盘点/借用模块共用) # 分页库存查询接口(服务端分页,出库/盘点/借用模块共用)
# ============================================================================== # ==============================================================================
def _do_get_stock_list(): def _make_price_stripper(permission_prefix=None):
"""
Fail-Closed 价格字段剥离器工厂。
选单/借用场景默认剥离所有价格字段;
仅当 permission_prefix 不在已知选单前缀列表中时放行(例如 material_list 场景由调用方自行处理)。
"""
# 已知的选单/借用前缀 → 无条件剥离所有价格成本
SELECTION_PREFIXES = {'outbound_selection', 'op_borrow_apply', 'inventory_stocktake'}
def stripper(d, stock_category):
if permission_prefix is None or permission_prefix in SELECTION_PREFIXES:
if stock_category == 'buy':
for k in ('unit_price', 'post_tax_unit_price', 'total_price',
'tax_rate', 'currency', 'exchange_rate',
'pre_tax_unit_price', 'qty_inbound', 'in_quantity'):
d.pop(k, None)
elif stock_category == 'semi':
for k in ('raw_material_cost', 'manual_cost', 'unit_total_cost',
'total_price', 'unit_price'):
d.pop(k, None)
elif stock_category == 'product':
for k in ('raw_material_cost', 'manual_cost', 'unit_total_cost',
'sale_price', 'unit_price'):
d.pop(k, None)
return stripper
def _do_get_stock_list(permission_prefix=None):
""" """
分页获取库存列表stock_quantity > 0 — 裸逻辑,供各模块复用 分页获取库存列表stock_quantity > 0 — 裸逻辑,供各模块复用
Args:
permission_prefix: 可选权限前缀(如 'outbound_selection' / 'op_borrow_apply')。
传入后自动剥离对应模块无权查看的价格/成本字段。
""" """
try: try:
page = request.args.get('page', 1, type=int) page = request.args.get('page', 1, type=int)
@ -233,6 +274,9 @@ def _do_get_stock_list():
if pageSize < 1 or pageSize > 200: if pageSize < 1 or pageSize > 200:
pageSize = 20 pageSize = 20
# ★ Fail-Closed: 选单/借用场景默认剥离所有价格成本字段
_strip_price_fields = _make_price_stripper(permission_prefix)
all_items = [] all_items = []
# 1. 采购件 # 1. 采购件
@ -255,6 +299,7 @@ def _do_get_stock_list():
d['name'] = d.get('material_name', d.get('name', '')) d['name'] = d.get('material_name', d.get('name', ''))
d['standard'] = d.get('spec_model', d.get('standard', '')) d['standard'] = d.get('spec_model', d.get('standard', ''))
d['available_quantity'] = d.get('qty_available', d.get('available_quantity', 0)) d['available_quantity'] = d.get('qty_available', d.get('available_quantity', 0))
_strip_price_fields(d, 'buy')
all_items.append(d) all_items.append(d)
# 2. 半成品 # 2. 半成品
@ -278,6 +323,7 @@ def _do_get_stock_list():
d['name'] = d.get('material_name', d.get('name', '')) d['name'] = d.get('material_name', d.get('name', ''))
d['standard'] = d.get('spec_model', d.get('standard', '')) d['standard'] = d.get('spec_model', d.get('standard', ''))
d['available_quantity'] = d.get('qty_available', d.get('available_quantity', 0)) d['available_quantity'] = d.get('qty_available', d.get('available_quantity', 0))
_strip_price_fields(d, 'semi')
all_items.append(d) all_items.append(d)
except Exception: except Exception:
pass pass
@ -304,6 +350,7 @@ def _do_get_stock_list():
d['name'] = d.get('material_name', d.get('name', '')) d['name'] = d.get('material_name', d.get('name', ''))
d['standard'] = d.get('spec_model', d.get('standard', '')) d['standard'] = d.get('spec_model', d.get('standard', ''))
d['available_quantity'] = d.get('qty_available', d.get('available_quantity', 0)) d['available_quantity'] = d.get('qty_available', d.get('available_quantity', 0))
_strip_price_fields(d, 'product')
all_items.append(d) all_items.append(d)
# ── 按规格+库位聚合(出库选单合并同类项)─────────────────────── # ── 按规格+库位聚合(出库选单合并同类项)───────────────────────
@ -360,8 +407,8 @@ def _do_get_stock_list():
@jwt_required() @jwt_required()
@permission_required('outbound_selection') @permission_required('outbound_selection')
def get_stock_list(): def get_stock_list():
"""出库选单专用库存列表""" """出库选单专用库存列表 — Fail-Closed: 剥离价格字段"""
return _do_get_stock_list() return _do_get_stock_list(permission_prefix='outbound_selection')
# --- 草稿箱接口 --- # --- 草稿箱接口 ---

View File

@ -86,6 +86,8 @@ def scan_barcode():
result = OutboundService.get_stock_by_barcode(barcode) result = OutboundService.get_stock_by_barcode(barcode)
if result: if result:
# ★ Fail-Closed: 扫码响应剥离价格字段
result.pop('price', None)
return jsonify({ return jsonify({
'code': 200, 'code': 200,
'msg': '扫描成功', 'msg': '扫描成功',
@ -279,6 +281,22 @@ def bom_match_stock():
except Exception: except Exception:
pass pass
# ★ Fail-Closed: 剥离所有价格成本字段BOM 匹配用于出库选单,无需价格)
for d in all_items:
stype = d.get('stock_type', '')
if stype == 'material':
for k in ('unit_price', 'post_tax_unit_price', 'total_price',
'tax_rate', 'currency', 'exchange_rate'):
d.pop(k, None)
elif stype == 'semi':
for k in ('raw_material_cost', 'manual_cost', 'unit_total_cost',
'total_price', 'unit_price'):
d.pop(k, None)
elif stype == 'product':
for k in ('raw_material_cost', 'manual_cost', 'unit_total_cost',
'sale_price', 'unit_price'):
d.pop(k, None)
return jsonify({'code': 200, 'msg': 'success', 'data': {'items': all_items}}) return jsonify({'code': 200, 'msg': 'success', 'data': {'items': all_items}})
except Exception as e: except Exception as e:

View File

@ -49,6 +49,8 @@ def scan_barcode():
try: try:
result = ScrapService.get_stock_by_barcode(barcode) result = ScrapService.get_stock_by_barcode(barcode)
if result: if result:
# ★ Fail-Closed: 扫码响应剥离价格字段
result.pop('price', None)
return jsonify({'code': 200, 'msg': '扫描成功', 'data': result}) return jsonify({'code': 200, 'msg': '扫描成功', 'data': result})
else: else:
return jsonify({'code': 404, 'msg': '未找到对应的库存记录'}), 404 return jsonify({'code': 404, 'msg': '未找到对应的库存记录'}), 404
@ -124,6 +126,10 @@ def get_scrap_records():
start_date=start_date, start_date=start_date,
end_date=end_date end_date=end_date
) )
# ★ Fail-Closed: 报废记录剥离成本字段
for item in (result.get('list') or []):
for k in ('cost_at_scrap', 'total_loss'):
item.pop(k, None)
return jsonify({'code': 200, 'msg': 'success', 'data': result}) return jsonify({'code': 200, 'msg': 'success', 'data': result})
except Exception as e: except Exception as e:
traceback.print_exc() traceback.print_exc()

View File

@ -44,23 +44,33 @@ def get_current_user_info():
def filter_item_by_permissions(item_dict, user_permissions, prefix='op_records'): def filter_item_by_permissions(item_dict, user_permissions, prefix='op_records'):
""" """
根据用户权限过滤 item 字典,无权限的字段值置为 None 根据用户权限过滤 item 字典,无权限的字段值置为 None
★ Fail-Closed: 字段映射默认为完整列表,不再为空字典。
""" """
# 字段名到权限码的映射(与前端 permissionMap 保持一致)
field_to_perm = { field_to_perm = {
# 'borrow_no': f'{prefix}:borrow_no', 'id': f'{prefix}:id',
# 'borrower_name': f'{prefix}:borrower_name', 'borrow_no': f'{prefix}:borrow_no',
# 'sku': f'{prefix}:sku', 'borrower_name': f'{prefix}:borrower_name',
# 'borrow_time': f'{prefix}:borrow_time', 'sku': f'{prefix}:sku',
# 'return_time': f'{prefix}:return_time', 'source_table': f'{prefix}:source_table',
# 'return_operator': f'{prefix}:return_operator', 'stock_id': f'{prefix}:stock_id',
# 'status': f'{prefix}:status', 'barcode': f'{prefix}:barcode',
# 'expected_return_time': f'{prefix}:expected_return_time', 'quantity': f'{prefix}:quantity',
# 'return_location': f'{prefix}:return_location', 'returned_quantity': f'{prefix}:returned_quantity',
# 'borrow_signature': f'{prefix}:borrow_signature', 'borrow_time': f'{prefix}:borrow_time',
# 'return_signature': f'{prefix}:return_signature', 'return_time': f'{prefix}:return_time',
'return_operator': f'{prefix}:return_operator',
'return_location': f'{prefix}:return_location',
'status': f'{prefix}:status',
'expected_return_time': f'{prefix}:expected_return_time',
'borrow_signature': f'{prefix}:borrow_signature',
'return_signature': f'{prefix}:return_signature',
'remark': f'{prefix}:remark',
'material_name': f'{prefix}:material_name',
'is_returned': f'{prefix}:is_returned',
'current_location': f'{prefix}:current_location',
} }
# 如果用户是超级管理员且有 '*',则不过滤 if '*' in user_permissions or f'{prefix}:*' in user_permissions:
if '*' in user_permissions:
return item_dict return item_dict
for field, perm_code in field_to_perm.items(): for field, perm_code in field_to_perm.items():
if field in item_dict and perm_code not in user_permissions: if field in item_dict and perm_code not in user_permissions:
@ -290,9 +300,9 @@ def get_borrow_request_list():
@jwt_required() @jwt_required()
@permission_required('op_borrow_apply') @permission_required('op_borrow_apply')
def get_borrow_stock_list(): def get_borrow_stock_list():
"""借库选单专用库存列表,与出库选单共享底层逻辑""" """借库选单专用库存列表 — Fail-Closed: 剥离价格字段"""
from app.api.v1.inbound.stock import _do_get_stock_list from app.api.v1.inbound.stock import _do_get_stock_list
return _do_get_stock_list() return _do_get_stock_list(permission_prefix='op_borrow_apply')
# --- 执行借库扣减(审批通过后调用)--- # --- 执行借库扣减(审批通过后调用)---