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

feat(security): implement strict row-level data isolation based on user company

Browse Source
This commit is contained in:
DXC
2026-04-14 08:38:50 +08:00
parent 81bfb29b50
commit 0e8ddd0851
3 changed files with 60 additions and 11 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@ -1,5 +1,6 @@
# 文件路径: app/services/inbound/base_service.py
from flask_jwt_extended import get_jwt
from app.extensions import db
from app.models.base import MaterialBase, MaterialWarningSetting
from app.models.inbound.buy import StockBuy
@ -209,9 +210,26 @@ class MaterialBaseService:
MaterialBase.spec_model.ilike(kw)
))
company = filters.get('company')
if company is not None and company != '':
query = query.filter(MaterialBase.company_name.ilike(company.strip()))
# ============================================================
# 【行级数据隔离】基于 JWT 中的 company_name 进行过滤
# ============================================================
claims = get_jwt()
user_role = claims.get('role', '').upper() if claims.get('role') else ''
user_company = claims.get('company_name', '')
# 获取前端传的查询参数
req_company = filters.get('company') if filters else None
if user_role != 'SUPER_ADMIN':
# 普通用户：强制隔离！无视前端传的 company 参数
if user_company:
query = query.filter(MaterialBase.company_name == user_company)
# 如果用户没有所属公司字段，则只显示公司为空的记录（或不允许查看）
else:
# 超级管理员：允许跨公司视角
if req_company:
query = query.filter(MaterialBase.company_name == req_company)
# 超管没选公司则不加过滤，看到全量
category = filters.get('category')
if category is not None and category != '':
@ -618,9 +636,23 @@ class MaterialBaseService:
MaterialBase.spec_model.ilike(kw),
MaterialBase.company_name.ilike(kw)
))
company = filters.get('company')
if company is not None and company != '':
filter_conditions.append(MaterialBase.company_name.ilike(company.strip()))
# ============================================================
# 【行级数据隔离】基于 JWT 中的 company_name 进行过滤（高级筛选）
# ============================================================
claims = get_jwt()
user_role = claims.get('role', '').upper() if claims.get('role') else ''
user_company = claims.get('company_name', '')
req_company = filters.get('company') if filters else None
if user_role != 'SUPER_ADMIN':
# 普通用户：强制隔离
if user_company:
filter_conditions.append(MaterialBase.company_name == user_company)
else:
# 超级管理员：允许跨公司视角
if req_company:
filter_conditions.append(MaterialBase.company_name == req_company)
category = filters.get('category')
if category is not None and category != '':
filter_conditions.append(MaterialBase.category.ilike(category.strip()))