duxingchen/KCGL
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(backend): apply global cross-company data isolation logic across all inbound, outbound, and stock services

Browse Source
This commit is contained in:
DXC
2026-04-17 09:57:00 +08:00
parent 6c0e13e52d
commit 8291a89898
5 changed files with 118 additions and 21 deletions
Download Patch File Download Diff File Expand all files Collapse all files

35
inventory-backend/app/services/inbound/base_service.py
View File

@ -213,14 +213,24 @@ class MaterialBaseService:
# ============================================================
# 【行级数据隔离】基于 JWT 中的 company_name 进行过滤
# ============================================================
from flask_jwt_extended import get_jwt
claims = get_jwt()
user_role = claims.get('role', '').upper() if claims.get('role') else ''
user_company = claims.get('company_name', '')
# 获取用户权限列表（用于检查 global:cross_company_op 特权）
from app.api.v1.inbound.base import get_current_user_permissions
user_perms = get_current_user_permissions() or []
normalized_perms = set(p.lower().replace('_', '').replace(':', '') for p in user_perms)
# 检查是否拥有全局特权或超管角色
has_cross_company = 'globalcrosscompanyop' in normalized_perms
# 获取前端传的查询参数
req_company = filters.get('company') if filters else None
if user_role != 'SUPER_ADMIN':
if user_role != 'SUPER_ADMIN' and not has_cross_company:
# 【显式拒绝越权】如果前端传了公司参数，且不是当前用户的公司，返回403
if req_company and req_company != user_company:
from flask import abort
@ -229,11 +239,11 @@ class MaterialBaseService:
if user_company:
query = query.filter(MaterialBase.company_name == user_company)
# 如果用户没有所属公司字段，则只显示公司为空的记录（或不允许查看）
else:
# 超级管理员：允许跨公司视角
elif user_role == 'SUPER_ADMIN' or has_cross_company:
# 超级管理员或有跨域特权：允许跨公司视角
if req_company:
query = query.filter(MaterialBase.company_name == req_company)
# 超管没选公司则不加过滤，看到全量
# 没选公司则不加过滤，看到全量
category = filters.get('category')
if category is not None and category != '':
@ -643,17 +653,28 @@ class MaterialBaseService:
# ============================================================
# 【行级数据隔离】基于 JWT 中的 company_name 进行过滤（高级筛选）
# ============================================================
from flask_jwt_extended import get_jwt
claims = get_jwt()
user_role = claims.get('role', '').upper() if claims.get('role') else ''
user_company = claims.get('company_name', '')
# 获取用户权限列表（用于检查 global:cross_company_op 特权）
from app.api.v1.inbound.base import get_current_user_permissions
user_perms = get_current_user_permissions() or []
normalized_perms = set(p.lower().replace('_', '').replace(':', '') for p in user_perms)
# 检查是否拥有全局特权或超管角色
has_cross_company = 'globalcrosscompanyop' in normalized_perms
req_company = filters.get('company') if filters else None
if user_role != 'SUPER_ADMIN':
if user_role != 'SUPER_ADMIN' and not has_cross_company:
# 普通用户：强制隔离
if user_company:
filter_conditions.append(MaterialBase.company_name == user_company)
else:
# 超级管理员：允许跨公司视角
elif user_role == 'SUPER_ADMIN' or has_cross_company:
# 超级管理员或有跨域特权：允许跨公司视角
if req_company:
filter_conditions.append(MaterialBase.company_name == req_company)