From f5ab023ab5694b190f9053aff28fee979349c0de Mon Sep 17 00:00:00 2001 From: zlwu Date: Sun, 28 Jun 2026 20:30:36 +0800 Subject: [PATCH] =?UTF-8?q?feat(auth):=20=E6=89=A9=E5=B1=95=E6=95=B0?= =?UTF-8?q?=E6=8D=AE=E6=9D=83=E9=99=90=E6=A0=B8=E5=BF=83=E8=83=BD=E5=8A=9B?= =?UTF-8?q?=E5=B9=B6=E6=8E=A5=E5=85=A5=E7=B3=BB=E7=BB=9F=E7=AE=A1=E7=90=86?= =?UTF-8?q?=E8=BE=85=E5=8A=A9=E6=8E=A5=E5=8F=A3?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 为 apply_data_scope 增加 owner_field 与行级校验，补充 DbUserContext.to_user_info() 及单元测试，并在用户/角色/字典下拉与用户详情接口生效。 Co-authored-by: Cursor --- backend/app/api/admin.py | 34 ++++-- backend/app/core/data_scope.py | 201 +++++++++++++++++++++---------- backend/app/core/db_auth.py | 10 ++ backend/app/crud/dict_data.py | 10 +- backend/tests/test_data_scope.py | 128 ++++++++++++++++++++ 5 files changed, 306 insertions(+), 77 deletions(-) create mode 100644 backend/tests/test_data_scope.py diff --git a/backend/app/api/admin.py b/backend/app/api/admin.py index 76428c8e4..9ce70fad4 100644 --- a/backend/app/api/admin.py +++ b/backend/app/api/admin.py @@ -3,13 +3,13 @@ import os import uuid from typing import Any, List, Optional -from fastapi import APIRouter, Depends, Query, UploadFile, File, Form +from fastapi import APIRouter, Depends, HTTPException, Query, UploadFile, File, Form from pydantic import BaseModel, ConfigDict, Field from sqlalchemy import and_, or_ from sqlalchemy.orm import Session from app.core.config import settings -from app.core.data_scope import apply_data_scope +from app.core.data_scope import apply_data_scope, assert_row_in_data_scope from app.core.database import get_db from app.core.menu_guard import require_admin_or_super_admin, require_menu_code from app.core.response import success_response, error_response @@ -72,14 +72,14 @@ async def list_roles_for_select( @router.get("/users/options") async def list_users_for_select( - _user: dict = Depends(require_admin_or_super_admin), + user: dict = Depends(require_admin_or_super_admin), db: Session = Depends(get_db), ): """返回用户下拉选项（label=昵称，value=用户ID，附带 nickName/userNo），供审批人选择等场景使用。""" - tenant_no = _user.get("tenant_no", "") + tenant_no = _get_tenant_no(db, user) q = db.query(SysUser).filter(SysUser.del_flag == 0, SysUser.status == "1") - if tenant_no: - q = q.filter(SysUser.tenant_no == tenant_no) + q = q.filter(SysUser.tenant_no == tenant_no) + q = apply_data_scope(q, SysUser, user, db, tenant_no) rows = q.order_by(SysUser.create_time.desc()).all() out = [ { @@ -95,15 +95,15 @@ async def list_users_for_select( @router.get("/roles/options") async def list_roles_for_designer( - _user: dict = Depends(require_admin_or_super_admin), + user: dict = Depends(require_admin_or_super_admin), db: Session = Depends(get_db), ): """返回角色下拉选项（label=角色名，value=角色ID UUID），供流程设计器审批人选择使用。 与 /roles 不同，此端点返回角色 UUID 作为 value（引擎解析审批人需要 UUID）。""" - tenant_no = _user.get("tenant_no", "") + tenant_no = _get_tenant_no(db, user) q = db.query(SysRole).filter(SysRole.del_flag == 0, SysRole.status == "1") - if tenant_no: - q = q.filter(SysRole.tenant_no == tenant_no) + q = q.filter(SysRole.tenant_no == tenant_no) + q = apply_data_scope(q, SysRole, user, db, tenant_no) rows = q.order_by(SysRole.create_time.desc()).all() out = [{"label": (r.role_name or "").strip(), "value": r.id} for r in rows if (r.role_name or "").strip()] return success_response(out) @@ -323,6 +323,10 @@ async def get_user_by_id( return error_response(404, "用户不存在") if row.tenant_no != tenant_no and user.get("role") != "super_admin": return error_response(403, "无权查看该用户") + try: + assert_row_in_data_scope(db, row, SysUser, user, tenant_no) + except HTTPException: + return error_response(403, "无权查看该用户") data = user_svc.get_user_with_relations(db, user_id) return success_response(data) @@ -626,7 +630,9 @@ async def dict_data_options( db: Session = Depends(get_db), ): tenant_no = _get_tenant_no(db, user) - items = dict_data_svc.list_dict_data_options(db, dict_type_code=dictType, tenant_no=tenant_no) + items = dict_data_svc.list_dict_data_options( + db, dict_type_code=dictType, tenant_no=tenant_no, user_info=user + ) return success_response(items) @@ -636,9 +642,11 @@ async def dict_data_all( db: Session = Depends(get_db), ): tenant_no = _get_tenant_no(db, user) - rows = db.query(SysDictData).filter( + q = db.query(SysDictData).filter( and_(SysDictData.tenant_no == tenant_no, SysDictData.del_flag == 0) - ).order_by(SysDictData.order_num.asc()).all() + ) + q = apply_data_scope(q, SysDictData, user, db, tenant_no) + rows = q.order_by(SysDictData.order_num.asc()).all() grouped = {} for r in rows: key = r.dict_type diff --git a/backend/app/core/data_scope.py b/backend/app/core/data_scope.py index b02df6c73..effbc2c14 100644 --- a/backend/app/core/data_scope.py +++ b/backend/app/core/data_scope.py @@ -9,8 +9,9 @@ data_scope 值： """ from __future__ import annotations -from typing import List, Set, Tuple +from typing import List, Optional, Set, Tuple +from fastapi import HTTPException from sqlalchemy import or_ from sqlalchemy.orm import Session @@ -37,7 +38,6 @@ def get_user_dept_id(db: Session, user_id: str) -> str | None: def _get_child_dept_ids(db: Session, dept_id: str, tenant_no: str) -> List[str]: """递归获取 dept_id 及其所有子部门 ID（含自身）。""" result: List[str] = [dept_id] - # 用循环逐层扩展，避免深度递归栈溢出 queue: List[str] = [dept_id] while queue: parents = queue @@ -85,7 +85,6 @@ def _resolve_user_role_scopes( result: List[dict] = [] role_ids = [r.id for r in roles] - # 批量查所有自定义权限明细 all_details = ( db.query(SysRoleDataScopeDetail) .filter( @@ -109,76 +108,31 @@ def _resolve_user_role_scopes( return result -def apply_data_scope(query, model, user_info: dict, db: Session, tenant_no: str = None): - """根据用户角色的数据权限过滤查询。 - - Args: - query: SQLAlchemy query 对象 - model: 查询的主模型类（需要有 create_dept_id / create_by 字段） - user_info: 当前用户信息字典，需含 role / user_id / dept_id 等 - db: 数据库 Session - tenant_no: 租户编号，默认使用 settings.DEFAULT_TENANT_NO - - Returns: - 应用数据权限过滤后的 query - """ - # 1. 无用户信息 → 不过滤 - if not user_info: - return query - - # 2. super_admin 直接放行 - if user_info.get("role") == "super_admin": - return query - - # 3. 获取 user_id - user_id = user_info.get("user_id") - if not user_id: - return query - - # 4. 确定 tenant_no - effective_tenant = tenant_no or settings.DEFAULT_TENANT_NO - - # 5. 检查 model 是否有可过滤字段 - has_dept = hasattr(model, "create_dept_id") - has_user = hasattr(model, "create_by") - if not has_dept and not has_user: - return query - - # 6. 获取用户所有角色的数据权限 - role_scopes = _resolve_user_role_scopes(db, user_id, effective_tenant) - if not role_scopes: - return query.filter(False) - - # 7. 任一角色拥有全部数据权限 → 放行（union 语义） - if any(rs["data_scope"] == "1" for rs in role_scopes): - return query - - # 8. 收集所有角色的 dept_ids 和 user_ids（union 合并） +def _collect_scope_ids( + db: Session, + user_info: dict, + user_id: str, + effective_tenant: str, + role_scopes: List[dict], +) -> Tuple[Set[str], Set[str]]: + """合并多角色数据范围为 dept_ids / user_ids 并集。""" dept_ids: Set[str] = set() user_ids: Set[str] = set() - - # 获取用户部门 - user_dept_id = user_info.get("dept_id") or user_info.get("region_id") or get_user_dept_id(db, user_id) + user_dept_id = user_info.get("dept_id") or user_info.get("region_id") or get_user_dept_id( + db, user_id + ) for rs in role_scopes: scope = rs["data_scope"] - if scope == "2": - # 本部门 if user_dept_id: dept_ids.add(user_dept_id) - elif scope == "3": - # 本部门及以下 if user_dept_id: dept_ids.update(_get_child_dept_ids(db, user_dept_id, effective_tenant)) - elif scope == "4": - # 仅本人 user_ids.add(user_id) - elif scope == "5": - # 自定义 for detail in rs["details"]: val = (detail.scope_value or "").strip() if not val: @@ -187,16 +141,137 @@ def apply_data_scope(query, model, user_info: dict, db: Session, tenant_no: str dept_ids.add(val) elif detail.scope_type == "user": user_ids.add(val) + return dept_ids, user_ids + + +def _append_user_scope_conditions( + conditions: list, + model, + user_ids: Set[str], + *, + has_user: bool, + owner_field: Optional[str], +) -> None: + if not user_ids: + return + user_conds = [] + if has_user: + user_conds.append(model.create_by.in_(user_ids)) + if owner_field and hasattr(model, owner_field): + user_conds.append(getattr(model, owner_field).in_(user_ids)) + if user_conds: + conditions.append(or_(*user_conds)) + + +def apply_data_scope( + query, + model, + user_info: dict, + db: Session, + tenant_no: str = None, + owner_field: Optional[str] = None, +): + """根据用户角色的数据权限过滤查询。 + + Args: + query: SQLAlchemy query 对象 + model: 查询的主模型类（需要有 create_dept_id / create_by 字段） + user_info: 当前用户信息字典，需含 role / user_id / dept_id 等 + db: 数据库 Session + tenant_no: 租户编号，默认使用 settings.DEFAULT_TENANT_NO + owner_field: 可选负责人字段名（如 CRM 的 owner_user_id），与 create_by 取 OR + + Returns: + 应用数据权限过滤后的 query + """ + if not user_info: + return query + + if user_info.get("role") == "super_admin": + return query + + user_id = user_info.get("user_id") + if not user_id: + return query + + effective_tenant = tenant_no or settings.DEFAULT_TENANT_NO + + has_dept = hasattr(model, "create_dept_id") + has_user = hasattr(model, "create_by") + if not has_dept and not has_user: + return query + + role_scopes = _resolve_user_role_scopes(db, user_id, effective_tenant) + if not role_scopes: + return query.filter(False) + + if any(rs["data_scope"] == "1" for rs in role_scopes): + return query + + dept_ids, user_ids = _collect_scope_ids(db, user_info, user_id, effective_tenant, role_scopes) - # 9. 构建 OR 条件并应用 conditions = [] if dept_ids and has_dept: conditions.append(model.create_dept_id.in_(dept_ids)) - if user_ids and has_user: - conditions.append(model.create_by.in_(user_ids)) + _append_user_scope_conditions( + conditions, model, user_ids, has_user=has_user, owner_field=owner_field + ) if not conditions: - # 没有任何可匹配条件 → 返回空结果 return query.filter(False) return query.filter(or_(*conditions)) + + +def row_in_data_scope( + db: Session, + row, + user_info: dict, + tenant_no: Optional[str] = None, + owner_field: Optional[str] = None, +) -> bool: + """判断单行数据是否在当前用户数据权限范围内。""" + if not user_info or row is None: + return False + if user_info.get("role") == "super_admin": + return True + + user_id = user_info.get("user_id") + if not user_id: + return False + + effective_tenant = tenant_no or settings.DEFAULT_TENANT_NO + role_scopes = _resolve_user_role_scopes(db, user_id, effective_tenant) + if not role_scopes: + return False + if any(rs["data_scope"] == "1" for rs in role_scopes): + return True + + dept_ids, user_ids = _collect_scope_ids(db, user_info, user_id, effective_tenant, role_scopes) + + row_dept = (getattr(row, "create_dept_id", None) or "").strip() + row_creator = (getattr(row, "create_by", None) or "").strip() + row_owner = ( + (getattr(row, owner_field, None) or "").strip() if owner_field else "" + ) + + if dept_ids and row_dept and row_dept in dept_ids: + return True + if user_ids and row_creator and row_creator in user_ids: + return True + if user_ids and row_owner and row_owner in user_ids: + return True + return False + + +def assert_row_in_data_scope( + db: Session, + row, + _model, + user_info: dict, + tenant_no: Optional[str] = None, + owner_field: Optional[str] = None, +) -> None: + """不在数据权限范围内时抛出 403。""" + if not row_in_data_scope(db, row, user_info, tenant_no, owner_field=owner_field): + raise HTTPException(status_code=403, detail="无权访问该数据") diff --git a/backend/app/core/db_auth.py b/backend/app/core/db_auth.py index c7168a5c1..c4f4773a4 100644 --- a/backend/app/core/db_auth.py +++ b/backend/app/core/db_auth.py @@ -35,6 +35,16 @@ class DbUserContext: "tenant_no": self.tenant_no, } + def to_user_info(self) -> Dict[str, Any]: + """与门户 JWT / apply_data_scope 所需的 user_info 字段对齐。""" + return { + "user_id": self.id, + "role": self.role, + "tenant_no": self.tenant_no, + "dept_id": self.dept_id, + "region_id": self.dept_id, + } + def get_db_current_user( user: dict = Depends(require_admin_or_super_admin), diff --git a/backend/app/crud/dict_data.py b/backend/app/crud/dict_data.py index 05ddc74e4..56282d908 100644 --- a/backend/app/crud/dict_data.py +++ b/backend/app/crud/dict_data.py @@ -150,13 +150,21 @@ def soft_delete_dict_data(db: Session, data_id: str) -> None: def list_dict_data_options( - db: Session, *, dict_type_code: str, tenant_no: Optional[str] = None + db: Session, + *, + dict_type_code: str, + tenant_no: Optional[str] = None, + user_info=None, ) -> List[Dict[str, str]]: q = db.query(SysDictData).filter( and_(SysDictData.dict_type == dict_type_code, SysDictData.del_flag == 0) ) if tenant_no: q = q.filter(SysDictData.tenant_no == tenant_no) + if user_info: + from app.core.data_scope import apply_data_scope + + q = apply_data_scope(q, SysDictData, user_info, db, tenant_no) rows = q.order_by(SysDictData.create_time.desc()).all() return [{"label": r.dict_label, "value": r.dict_value, "tagType": r.tag_type or ""} for r in rows] diff --git a/backend/tests/test_data_scope.py b/backend/tests/test_data_scope.py new file mode 100644 index 000000000..12d9377c7 --- /dev/null +++ b/backend/tests/test_data_scope.py @@ -0,0 +1,128 @@ +"""数据权限 apply_data_scope / row_in_data_scope 单元测试。""" +from __future__ import annotations + +from types import SimpleNamespace +from unittest.mock import MagicMock, patch + +import pytest +from fastapi import HTTPException +from sqlalchemy import true + +from app.core.data_scope import ( + apply_data_scope, + assert_row_in_data_scope, + row_in_data_scope, +) + + +class _FakeModel: + create_dept_id = MagicMock() + create_by = MagicMock() + owner_user_id = MagicMock() + + +def test_super_admin_bypass_apply(): + query = MagicMock() + result = apply_data_scope(query, _FakeModel, {"role": "super_admin"}, MagicMock(), "shanxia2.0") + assert result is query + query.filter.assert_not_called() + + +def test_scope4_invokes_filter_via_scope_collection(): + query = MagicMock() + user_info = {"user_id": "user-1", "role": "admin", "dept_id": "dept-a"} + with patch( + "app.core.data_scope._resolve_user_role_scopes", + return_value=[{"data_scope": "4", "details": []}], + ), patch( + "app.core.data_scope._collect_scope_ids", + return_value=(set(), {"user-1"}), + ), patch("app.core.data_scope._append_user_scope_conditions") as append_user: + def _fake_append(conditions, model, user_ids, *, has_user, owner_field): + conditions.append(true()) + + append_user.side_effect = _fake_append + apply_data_scope(query, _FakeModel, user_info, MagicMock(), "shanxia2.0") + assert query.filter.called + + +def test_scope4_with_owner_field_passes_owner_to_helper(): + query = MagicMock() + user_info = {"user_id": "user-1", "role": "admin", "dept_id": "dept-a"} + with patch( + "app.core.data_scope._resolve_user_role_scopes", + return_value=[{"data_scope": "4", "details": []}], + ), patch( + "app.core.data_scope._collect_scope_ids", + return_value=(set(), {"user-1"}), + ), patch("app.core.data_scope._append_user_scope_conditions") as append_user: + apply_data_scope( + query, + _FakeModel, + user_info, + MagicMock(), + "shanxia2.0", + owner_field="owner_user_id", + ) + append_user.assert_called_once() + assert append_user.call_args.kwargs["owner_field"] == "owner_user_id" + + +def test_no_role_scopes_returns_empty_query(): + query = MagicMock() + user_info = {"user_id": "user-1", "role": "admin"} + with patch("app.core.data_scope._resolve_user_role_scopes", return_value=[]): + apply_data_scope(query, _FakeModel, user_info, MagicMock(), "shanxia2.0") + query.filter.assert_called() + + +def test_row_in_data_scope_super_admin(): + row = SimpleNamespace(create_dept_id="x", create_by="y") + assert row_in_data_scope(MagicMock(), row, {"role": "super_admin"}, "shanxia2.0") is True + + +def test_row_in_data_scope_self_by_create_by(): + row = SimpleNamespace(create_dept_id="", create_by="user-1") + user_info = {"user_id": "user-1", "role": "admin", "dept_id": "dept-a"} + with patch( + "app.core.data_scope._resolve_user_role_scopes", + return_value=[{"data_scope": "4", "details": []}], + ): + assert row_in_data_scope(MagicMock(), row, user_info, "shanxia2.0") is True + + +def test_row_in_data_scope_self_by_owner_field(): + row = SimpleNamespace(create_dept_id="", create_by="other", owner_user_id="user-1") + user_info = {"user_id": "user-1", "role": "admin", "dept_id": "dept-a"} + with patch( + "app.core.data_scope._resolve_user_role_scopes", + return_value=[{"data_scope": "4", "details": []}], + ): + assert ( + row_in_data_scope( + MagicMock(), row, user_info, "shanxia2.0", owner_field="owner_user_id" + ) + is True + ) + + +def test_row_in_data_scope_denied(): + row = SimpleNamespace(create_dept_id="dept-z", create_by="other") + user_info = {"user_id": "user-1", "role": "admin", "dept_id": "dept-a"} + with patch( + "app.core.data_scope._resolve_user_role_scopes", + return_value=[{"data_scope": "4", "details": []}], + ): + assert row_in_data_scope(MagicMock(), row, user_info, "shanxia2.0") is False + + +def test_assert_row_in_data_scope_raises_403(): + row = SimpleNamespace(create_dept_id="dept-z", create_by="other") + user_info = {"user_id": "user-1", "role": "admin", "dept_id": "dept-a"} + with patch( + "app.core.data_scope._resolve_user_role_scopes", + return_value=[{"data_scope": "4", "details": []}], + ): + with pytest.raises(HTTPException) as exc: + assert_row_in_data_scope(MagicMock(), row, _FakeModel, user_info, "shanxia2.0") + assert exc.value.status_code == 403