feat(project): enforce team-scoped selection with explicit all-project permission
Separate cross-team selection from attribution maintenance and apply the same scope to options, history filters and writes. Add an opt-in tenant permission without granting existing roles. Protect authorization mutation paths against self-grants, validate tenant targets and links, and include menu ancestors atomically. Cover role union, revocation, mobile writes and administration boundaries with regression tests.
This commit is contained in:
1 parent
b510d18439
commit
0bddf38e52
16 files changed
+555
-46
No files matched your search
+12
-1
@@ -57,10 +57,21 @@ SQL 位于 [docs/sql](../docs/sql/)。当前初始化基线不完整,部分历
|
||||
切换到项目统计版本前,先在已授权数据库执行增量结构脚本
|
||||
[2026-11-02-project_management.sql](../docs/sql/migrations/2026-11-02-project_management.sql),
|
||||
再依次执行[菜单种子](../docs/sql/migrations/2026-11-02-project_management_menu_seed.sql)与
|
||||
[项目一级菜单迁移](../docs/sql/migrations/2026-11-02-project_top_menu.sql)。
|
||||
[项目一级菜单迁移](../docs/sql/migrations/2026-11-02-project_top_menu.sql),
|
||||
最后执行[全部项目权限种子](../docs/sql/migrations/2026-11-02-project_all_permission.sql)。
|
||||
“项目”与“市场、生产、质检”平级,包含项目组管理、项目管理、业务统计、历史归属维护。
|
||||
菜单种子不会给角色新增业务权限,应在现有角色管理分配项目管理权限及祖先菜单;
|
||||
一级菜单迁移保留已有叶子 ID/授权,仅给已授权项目子菜单的角色补齐新祖先。
|
||||
“所属项目”默认只列出本人担任负责人或成员的项目组内项目(多组取并集)。
|
||||
需要跨组选择时,在“系统管理 → 角色管理 → 菜单权限 → 项目 → 全部项目”显式授予
|
||||
`project:project:all`;多角色与职位继承权限取有效角色并集。已有归属维护权限、普通管理员身份不会自动获得该权限。
|
||||
超级管理员沿用系统全部权限语义,但所有候选仍限当前租户。新建仅列启用且支持该业务的项目;
|
||||
历史筛选叠加原订单数据权限,可保留组内停用项目。全部项目权限不授予订单查看、项目资料编辑或归属维护权限。
|
||||
授权配置入口同时加固:用户、角色、职位和菜单变更分别需要已有的 `system:user:list`、
|
||||
`system:role:list`、`system:post:list`、`system:menu:list` 权限(超级管理员保留全权限),
|
||||
避免普通业务管理员自行绑定角色或修改权限码。质检金额专用角色授权的三个写入口也校验
|
||||
`biz:qcMaterialAmountAuth:list`,防止经专用角色绕过上述限制。配置保存校验当前租户,角色菜单自动补齐祖先,
|
||||
不自动勾选其他功能。升级前应核对实际配置人员已获对应系统管理权限。
|
||||
脚本不回填历史业务,不代表空库初始化;执行前备份并在隔离 MySQL 环境验证。
|
||||
|
||||
`PROJECT_ASSIGNMENT_REQUIRED` 默认 `false`,便于先配置项目组和项目、兼容历史客户端。
|
||||
|
||||
+44
-24
@@ -12,7 +12,7 @@ from sqlalchemy.orm import Session
|
||||
from app.core.config import settings
|
||||
from app.core.data_scope import apply_data_scope, _get_child_dept_ids
|
||||
from app.core.database import get_db
|
||||
from app.core.menu_guard import require_admin_or_super_admin, require_menu_code
|
||||
from app.core.menu_guard import require_admin_or_super_admin, require_menu_code, require_strict_menu_code
|
||||
from app.core.response import success_response, error_response
|
||||
from app.core.storage import storage
|
||||
from app.crud import tenant as tenant_svc
|
||||
@@ -27,7 +27,8 @@ from app.crud import menu as menu_svc
|
||||
from app.crud import role as role_svc
|
||||
from app.crud import user as user_svc
|
||||
from app.crud.collection_protocol import _lookup_attachment_by_url
|
||||
from app.models.models import BizAttachment, SysDept, SysRole, SysUser
|
||||
from app.models.models import BizAttachment, SysDept, SysRole, SysUser, SysMenu, SysPost
|
||||
from app.services.permission_administration import validate_tenant_references
|
||||
from app.services.web_auth_roles import (
|
||||
_map_role_name_to_web,
|
||||
list_user_ids_with_effective_role_name,
|
||||
@@ -238,12 +239,12 @@ def _get_tenant_no(db: Session, user: dict) -> str:
|
||||
|
||||
|
||||
def _assert_user_tenant_access(db: Session, user: dict, target_user_id: str):
|
||||
"""同租户或 super_admin 才可写操作用户;返回目标用户行。"""
|
||||
"""用户管理始终在当前认证租户内操作;返回目标用户行。"""
|
||||
row = user_svc.get_user(db, target_user_id)
|
||||
if not row:
|
||||
raise ValueError("用户不存在")
|
||||
tenant_no = _get_tenant_no(db, user)
|
||||
if row.tenant_no != tenant_no and user.get("role") != "super_admin":
|
||||
if row.tenant_no != tenant_no:
|
||||
raise ValueError("无权操作其他租户的用户")
|
||||
return row
|
||||
|
||||
@@ -383,11 +384,13 @@ async def get_user_by_id(
|
||||
@router.post("/users")
|
||||
async def create_user(
|
||||
body: UserCreateBody,
|
||||
user: dict = Depends(require_admin_or_super_admin),
|
||||
user: dict = Depends(require_strict_menu_code("system:user:list", allow_super_admin=True)),
|
||||
db: Session = Depends(get_db),
|
||||
):
|
||||
try:
|
||||
tenant_no = _get_tenant_no(db, user)
|
||||
validate_tenant_references(db, SysRole, body.role_ids, tenant_no)
|
||||
validate_tenant_references(db, SysPost, body.post_ids, tenant_no)
|
||||
if user_svc.user_no_exists(db, body.user_no, tenant_no):
|
||||
return error_response(400, "用户账号已存在")
|
||||
obj = user_svc.create_user(
|
||||
@@ -421,11 +424,14 @@ async def create_user(
|
||||
async def update_user(
|
||||
user_id: str,
|
||||
body: UserUpdateBody,
|
||||
user: dict = Depends(require_admin_or_super_admin),
|
||||
user: dict = Depends(require_strict_menu_code("system:user:list", allow_super_admin=True)),
|
||||
db: Session = Depends(get_db),
|
||||
):
|
||||
try:
|
||||
_assert_user_tenant_access(db, user, user_id)
|
||||
tenant_no = _get_tenant_no(db, user)
|
||||
validate_tenant_references(db, SysRole, body.role_ids, tenant_no)
|
||||
validate_tenant_references(db, SysPost, body.post_ids, tenant_no)
|
||||
payload: dict[str, Any] = body.model_dump(exclude_unset=True, by_alias=False)
|
||||
update_kwargs: dict[str, Any] = {
|
||||
"nick_name": payload.get("nick_name"),
|
||||
@@ -457,7 +463,7 @@ async def update_user(
|
||||
@router.delete("/users/{user_id}")
|
||||
async def delete_user(
|
||||
user_id: str,
|
||||
user: dict = Depends(require_admin_or_super_admin),
|
||||
user: dict = Depends(require_strict_menu_code("system:user:list", allow_super_admin=True)),
|
||||
db: Session = Depends(get_db),
|
||||
):
|
||||
try:
|
||||
@@ -471,7 +477,7 @@ async def delete_user(
|
||||
@router.post("/users/{user_id}/reset-password")
|
||||
async def reset_user_password(
|
||||
user_id: str,
|
||||
user: dict = Depends(require_admin_or_super_admin),
|
||||
user: dict = Depends(require_strict_menu_code("system:user:list", allow_super_admin=True)),
|
||||
db: Session = Depends(get_db),
|
||||
):
|
||||
try:
|
||||
@@ -971,7 +977,7 @@ async def list_posts(
|
||||
@router.post("/posts")
|
||||
async def create_post(
|
||||
body: PostCreateBody,
|
||||
user: dict = Depends(require_admin_or_super_admin),
|
||||
user: dict = Depends(require_strict_menu_code("system:post:list", allow_super_admin=True)),
|
||||
db: Session = Depends(get_db),
|
||||
):
|
||||
try:
|
||||
@@ -995,10 +1001,11 @@ async def create_post(
|
||||
async def update_post(
|
||||
post_id: str,
|
||||
body: PostUpdateBody,
|
||||
user: dict = Depends(require_admin_or_super_admin),
|
||||
user: dict = Depends(require_strict_menu_code("system:post:list", allow_super_admin=True)),
|
||||
db: Session = Depends(get_db),
|
||||
):
|
||||
try:
|
||||
validate_tenant_references(db, SysPost, [post_id], _get_tenant_no(db, user))
|
||||
payload: dict[str, Any] = body.model_dump(exclude_unset=True, by_alias=False)
|
||||
obj = post_svc.update_post(
|
||||
db,
|
||||
@@ -1018,10 +1025,11 @@ async def update_post(
|
||||
@router.delete("/posts/{post_id}")
|
||||
async def delete_post(
|
||||
post_id: str,
|
||||
_user: dict = Depends(require_admin_or_super_admin),
|
||||
_user: dict = Depends(require_strict_menu_code("system:post:list", allow_super_admin=True)),
|
||||
db: Session = Depends(get_db),
|
||||
):
|
||||
try:
|
||||
validate_tenant_references(db, SysPost, [post_id], _get_tenant_no(db, _user))
|
||||
post_svc.soft_delete_post(db, post_id)
|
||||
return success_response(None, message="删除成功")
|
||||
except ValueError as e:
|
||||
@@ -1052,7 +1060,7 @@ async def list_post_bound_users(
|
||||
async def update_post_bound_users(
|
||||
post_id: str,
|
||||
body: PostUsersUpdateBody,
|
||||
user: dict = Depends(require_admin_or_super_admin),
|
||||
user: dict = Depends(require_strict_menu_code("system:post:list", allow_super_admin=True)),
|
||||
db: Session = Depends(get_db),
|
||||
):
|
||||
tenant_no = _get_tenant_no(db, user)
|
||||
@@ -1092,7 +1100,7 @@ async def list_post_bound_roles(
|
||||
async def update_post_bound_roles(
|
||||
post_id: str,
|
||||
body: PostRolesUpdateBody,
|
||||
user: dict = Depends(require_admin_or_super_admin),
|
||||
user: dict = Depends(require_strict_menu_code("system:post:list", allow_super_admin=True)),
|
||||
db: Session = Depends(get_db),
|
||||
):
|
||||
tenant_no = _get_tenant_no(db, user)
|
||||
@@ -1353,11 +1361,13 @@ async def list_menus(
|
||||
@router.post("/menus")
|
||||
async def create_menu(
|
||||
body: MenuCreateBody,
|
||||
user: dict = Depends(require_admin_or_super_admin),
|
||||
user: dict = Depends(require_strict_menu_code("system:menu:list", allow_super_admin=True)),
|
||||
db: Session = Depends(get_db),
|
||||
):
|
||||
try:
|
||||
tenant_no = _get_tenant_no(db, user)
|
||||
if body.parent_id and body.parent_id != "0":
|
||||
validate_tenant_references(db, SysMenu, [body.parent_id], _get_tenant_no(db, user))
|
||||
obj = menu_svc.create_menu(
|
||||
db,
|
||||
menu_name=body.menu_name,
|
||||
@@ -1383,11 +1393,14 @@ async def create_menu(
|
||||
async def update_menu(
|
||||
menu_id: str,
|
||||
body: MenuUpdateBody,
|
||||
user: dict = Depends(require_admin_or_super_admin),
|
||||
user: dict = Depends(require_strict_menu_code("system:menu:list", allow_super_admin=True)),
|
||||
db: Session = Depends(get_db),
|
||||
):
|
||||
try:
|
||||
validate_tenant_references(db, SysMenu, [menu_id], _get_tenant_no(db, user))
|
||||
payload: dict[str, Any] = body.model_dump(exclude_unset=True, by_alias=False)
|
||||
if body.parent_id and body.parent_id != "0":
|
||||
validate_tenant_references(db, SysMenu, [body.parent_id], _get_tenant_no(db, user))
|
||||
obj = menu_svc.update_menu(
|
||||
db,
|
||||
menu_id,
|
||||
@@ -1412,10 +1425,11 @@ async def update_menu(
|
||||
@router.delete("/menus/{menu_id}")
|
||||
async def delete_menu(
|
||||
menu_id: str,
|
||||
_user: dict = Depends(require_admin_or_super_admin),
|
||||
_user: dict = Depends(require_strict_menu_code("system:menu:list", allow_super_admin=True)),
|
||||
db: Session = Depends(get_db),
|
||||
):
|
||||
try:
|
||||
validate_tenant_references(db, SysMenu, [menu_id], _get_tenant_no(db, _user))
|
||||
menu_svc.soft_delete_menu(db, menu_id)
|
||||
return success_response(None, message="删除成功")
|
||||
except ValueError as e:
|
||||
@@ -1475,7 +1489,7 @@ async def list_role_management(
|
||||
@router.post("/role-management")
|
||||
async def create_role_management(
|
||||
body: RoleCreateBody,
|
||||
user: dict = Depends(require_admin_or_super_admin),
|
||||
user: dict = Depends(require_strict_menu_code("system:role:list", allow_super_admin=True)),
|
||||
db: Session = Depends(get_db),
|
||||
):
|
||||
try:
|
||||
@@ -1499,10 +1513,11 @@ async def create_role_management(
|
||||
async def update_role_management(
|
||||
role_id: str,
|
||||
body: RoleUpdateBody,
|
||||
user: dict = Depends(require_admin_or_super_admin),
|
||||
user: dict = Depends(require_strict_menu_code("system:role:list", allow_super_admin=True)),
|
||||
db: Session = Depends(get_db),
|
||||
):
|
||||
try:
|
||||
validate_tenant_references(db, SysRole, [role_id], _get_tenant_no(db, user))
|
||||
payload: dict[str, Any] = body.model_dump(exclude_unset=True, by_alias=False)
|
||||
obj = role_svc.update_role(
|
||||
db,
|
||||
@@ -1522,10 +1537,11 @@ async def update_role_management(
|
||||
@router.delete("/role-management/{role_id}")
|
||||
async def delete_role_management(
|
||||
role_id: str,
|
||||
_user: dict = Depends(require_admin_or_super_admin),
|
||||
_user: dict = Depends(require_strict_menu_code("system:role:list", allow_super_admin=True)),
|
||||
db: Session = Depends(get_db),
|
||||
):
|
||||
try:
|
||||
validate_tenant_references(db, SysRole, [role_id], _get_tenant_no(db, _user))
|
||||
role_svc.soft_delete_role(db, role_id)
|
||||
return success_response(None, message="删除成功")
|
||||
except ValueError as e:
|
||||
@@ -1535,9 +1551,12 @@ async def delete_role_management(
|
||||
@router.get("/role-management/{role_id}/menus")
|
||||
async def get_role_menus(
|
||||
role_id: str,
|
||||
_user: dict = Depends(require_admin_or_super_admin),
|
||||
_user: dict = Depends(require_strict_menu_code("system:role:list", allow_super_admin=True)),
|
||||
db: Session = Depends(get_db),
|
||||
):
|
||||
role = role_svc.get_role(db, role_id)
|
||||
if not role or role.tenant_no != _get_tenant_no(db, _user):
|
||||
return error_response(404, "角色不存在")
|
||||
menu_ids = role_svc.get_role_menu_ids(db, role_id)
|
||||
return success_response({"menuIds": menu_ids})
|
||||
|
||||
@@ -1546,7 +1565,7 @@ async def get_role_menus(
|
||||
async def update_role_menus(
|
||||
role_id: str,
|
||||
body: RoleMenuBody,
|
||||
user: dict = Depends(require_admin_or_super_admin),
|
||||
user: dict = Depends(require_strict_menu_code("system:role:list", allow_super_admin=True)),
|
||||
db: Session = Depends(get_db),
|
||||
):
|
||||
try:
|
||||
@@ -1560,11 +1579,11 @@ async def update_role_menus(
|
||||
@router.get("/role-management/{role_id}/data-scope")
|
||||
async def get_role_data_scope(
|
||||
role_id: str,
|
||||
_user: dict = Depends(require_admin_or_super_admin),
|
||||
_user: dict = Depends(require_strict_menu_code("system:role:list", allow_super_admin=True)),
|
||||
db: Session = Depends(get_db),
|
||||
):
|
||||
role = role_svc.get_role(db, role_id)
|
||||
if not role:
|
||||
if not role or role.tenant_no != _get_tenant_no(db, _user):
|
||||
return error_response(404, "角色不存在")
|
||||
details = role_svc.get_role_data_scope(db, role_id)
|
||||
return success_response({"dataScope": role.data_scope or "1", "scopeDetails": details})
|
||||
@@ -1574,10 +1593,11 @@ async def get_role_data_scope(
|
||||
async def update_role_data_scope(
|
||||
role_id: str,
|
||||
body: RoleDataScopeBody,
|
||||
user: dict = Depends(require_admin_or_super_admin),
|
||||
user: dict = Depends(require_strict_menu_code("system:role:list", allow_super_admin=True)),
|
||||
db: Session = Depends(get_db),
|
||||
):
|
||||
try:
|
||||
validate_tenant_references(db, SysRole, [role_id], _get_tenant_no(db, user))
|
||||
tenant_no = _get_tenant_no(db, user)
|
||||
role_svc.update_role_data_scope(
|
||||
db,
|
||||
|
||||
@@ -13,6 +13,7 @@ from app.core.menu_guard import (
|
||||
require_admin_or_super_admin,
|
||||
require_any_menu_code,
|
||||
require_menu_code,
|
||||
require_strict_menu_code,
|
||||
strip_row_fields,
|
||||
user_has_menu_perm,
|
||||
)
|
||||
@@ -2650,7 +2651,7 @@ async def list_qc_material_amount_auth_users(
|
||||
async def grant_qc_material_amount_auth(
|
||||
body: QcMaterialAmountAuthGrantBody,
|
||||
db: Session = Depends(get_db),
|
||||
user: dict = Depends(require_menu_code(QC_MATERIAL_AMOUNT_AUTH_LIST_PERM)),
|
||||
user: dict = Depends(require_strict_menu_code(QC_MATERIAL_AMOUNT_AUTH_LIST_PERM, allow_super_admin=True)),
|
||||
):
|
||||
"""为用户绑定「质检物料金额查看」专用角色。"""
|
||||
try:
|
||||
@@ -2674,7 +2675,7 @@ async def grant_qc_material_amount_auth(
|
||||
async def revoke_qc_material_amount_auth(
|
||||
user_id: str,
|
||||
db: Session = Depends(get_db),
|
||||
user: dict = Depends(require_menu_code(QC_MATERIAL_AMOUNT_AUTH_LIST_PERM)),
|
||||
user: dict = Depends(require_strict_menu_code(QC_MATERIAL_AMOUNT_AUTH_LIST_PERM, allow_super_admin=True)),
|
||||
):
|
||||
"""解除用户的「质检物料金额查看」专用角色。"""
|
||||
try:
|
||||
@@ -2699,7 +2700,7 @@ async def revoke_qc_material_amount_auth(
|
||||
async def batch_revoke_qc_material_amount_auth(
|
||||
body: QcMaterialAmountAuthBatchBody,
|
||||
db: Session = Depends(get_db),
|
||||
user: dict = Depends(require_menu_code(QC_MATERIAL_AMOUNT_AUTH_LIST_PERM)),
|
||||
user: dict = Depends(require_strict_menu_code(QC_MATERIAL_AMOUNT_AUTH_LIST_PERM, allow_super_admin=True)),
|
||||
):
|
||||
"""批量解除「质检物料金额查看」授权。"""
|
||||
try:
|
||||
|
||||
@@ -113,7 +113,7 @@ def assert_strict_menu_code(
|
||||
raise HTTPException(status_code=403, detail="无对应菜单权限")
|
||||
|
||||
|
||||
def require_strict_menu_code(menu_code: str) -> Callable:
|
||||
def require_strict_menu_code(menu_code: str, *, allow_super_admin: bool = False) -> Callable:
|
||||
"""租户化的精确菜单权限依赖;与历史宽松守卫并存。"""
|
||||
required = str(menu_code or "").strip()
|
||||
|
||||
@@ -121,6 +121,8 @@ def require_strict_menu_code(menu_code: str) -> Callable:
|
||||
user: dict = Depends(get_current_user),
|
||||
db: Session = Depends(get_db),
|
||||
) -> dict:
|
||||
if allow_super_admin and user.get('role') == SUPER_ADMIN_ROLE:
|
||||
return user
|
||||
assert_strict_menu_code(db, user, required)
|
||||
return user
|
||||
|
||||
|
||||
@@ -10,6 +10,7 @@ from sqlalchemy.orm import Session
|
||||
|
||||
from app.core.utils import generate_uuid
|
||||
from app.crud.child_row_sync import sync_child_rows, sync_link_rows
|
||||
from app.services.permission_administration import menu_ids_with_ancestors, validate_tenant_references
|
||||
from app.models.models import (
|
||||
SysPost,
|
||||
SysPostRole,
|
||||
@@ -378,6 +379,8 @@ def update_role_menus(
|
||||
tenant_no: str,
|
||||
user_id: Optional[str] = None,
|
||||
) -> None:
|
||||
validate_tenant_references(db, SysRole, [role_id], tenant_no)
|
||||
menu_ids = menu_ids_with_ancestors(db, menu_ids, tenant_no)
|
||||
sync_link_rows(
|
||||
db,
|
||||
SysRoleMenu,
|
||||
|
||||
@@ -0,0 +1,35 @@
|
||||
"""Validate authorization targets before any role/menu relationship is changed."""
|
||||
from app.models.models import SysMenu
|
||||
|
||||
|
||||
def validate_tenant_references(db, model, ids, tenant_no):
|
||||
"""Reject deleted, foreign or missing targets without writing partial grants."""
|
||||
ids = list(dict.fromkeys(ids or []))
|
||||
if not ids:
|
||||
return
|
||||
found = {row.id for row in db.query(model.id).filter(
|
||||
model.id.in_(ids), model.tenant_no == tenant_no, model.del_flag == 0)}
|
||||
if found != set(ids):
|
||||
raise ValueError('关联记录不存在或不属于当前租户')
|
||||
|
||||
|
||||
def menu_ids_with_ancestors(db, menu_ids, tenant_no):
|
||||
"""Include only each selected node's same-tenant ancestors, up to its own root."""
|
||||
menus = {row.id: row for row in db.query(SysMenu).filter(
|
||||
SysMenu.tenant_no == tenant_no, SysMenu.del_flag == 0)}
|
||||
result, included = [], set()
|
||||
for selected in menu_ids:
|
||||
current, path = selected, set()
|
||||
while current and current != '0':
|
||||
if current in path:
|
||||
raise ValueError('菜单存在循环父级,不能授权')
|
||||
if current not in menus:
|
||||
raise ValueError('菜单或祖先不存在或不属于当前租户')
|
||||
path.add(current)
|
||||
if current not in included:
|
||||
result.append(current)
|
||||
included.add(current)
|
||||
current = menus[current].parent_id
|
||||
if not path:
|
||||
raise ValueError('菜单不存在或不属于当前租户')
|
||||
return result
|
||||
@@ -20,6 +20,7 @@ BUSINESS = {
|
||||
'basic_experiment': (BizPilotApplication, 'application_no', 'application_status'),
|
||||
}
|
||||
COUNT_TYPES = tuple(k for k in BUSINESS if k != 'presign')
|
||||
ALL_PROJECTS_PERMISSION = 'project:project:all'
|
||||
|
||||
|
||||
def scoped(db, model, tenant):
|
||||
@@ -71,9 +72,9 @@ def validate_assignment(db, tenant_no, user, project_id, biz_type, *, current_pr
|
||||
if effective_type not in row.biz_types.split(','):
|
||||
raise HTTPException(400, '项目不支持该业务类型')
|
||||
uid = user.get('user_id')
|
||||
member = uid in (row.leader_user_id, team.leader_user_id) or scoped(db, ProjectTeamMember, tenant_no).filter(ProjectTeamMember.team_id == team.id, ProjectTeamMember.user_id == uid).with_for_update().first()
|
||||
if not member and not has_permission(db, tenant_no, user, 'project:assignment:edit'):
|
||||
raise HTTPException(403, '仅项目负责人或成员可选择此项目')
|
||||
member = uid == team.leader_user_id or scoped(db, ProjectTeamMember, tenant_no).filter(ProjectTeamMember.team_id == team.id, ProjectTeamMember.user_id == uid).with_for_update().first()
|
||||
if not member and not has_permission(db, tenant_no, user, ALL_PROJECTS_PERMISSION):
|
||||
raise HTTPException(403, '仅可选择所在项目组的项目,跨组选择需全部项目权限')
|
||||
return project_id
|
||||
|
||||
|
||||
@@ -360,6 +361,20 @@ def delete_entity(db, tenant, user, model, row_id):
|
||||
return {'id': row.id}
|
||||
|
||||
|
||||
def apply_project_choice_scope(query, db, tenant, user):
|
||||
"""Apply team membership or the explicit tenant-wide project-choice permission.
|
||||
|
||||
The caller joins the tenant-scoped, non-deleted ProjectTeam table. Business
|
||||
data scope and project availability remain separate caller constraints.
|
||||
"""
|
||||
if has_permission(db, tenant, user, ALL_PROJECTS_PERMISSION):
|
||||
return query
|
||||
memberships = scoped(db, ProjectTeamMember, tenant).filter(
|
||||
ProjectTeamMember.user_id == user.get('user_id')).with_entities(ProjectTeamMember.team_id)
|
||||
return query.filter(or_(ProjectTeam.leader_user_id == user.get('user_id'),
|
||||
Project.team_id.in_(memberships)))
|
||||
|
||||
|
||||
def options(db, tenant, user, biz_type, keyword=None):
|
||||
if biz_type not in BUSINESS:
|
||||
raise HTTPException(400, '业务类型不支持')
|
||||
@@ -367,12 +382,7 @@ def options(db, tenant, user, biz_type, keyword=None):
|
||||
q = scoped(db, Project, tenant).join(ProjectTeam, and_(
|
||||
ProjectTeam.id == Project.team_id, ProjectTeam.tenant_no == tenant,
|
||||
ProjectTeam.del_flag == 0, ProjectTeam.enabled == 1)).filter(Project.status == 'active')
|
||||
if not has_permission(db, tenant, user, 'project:assignment:edit'):
|
||||
memberships = scoped(db, ProjectTeamMember, tenant).filter(
|
||||
ProjectTeamMember.user_id == user.get('user_id')).with_entities(ProjectTeamMember.team_id)
|
||||
q = q.filter(or_(Project.leader_user_id == user.get('user_id'),
|
||||
ProjectTeam.leader_user_id == user.get('user_id'),
|
||||
Project.team_id.in_(memberships)))
|
||||
q = apply_project_choice_scope(q, db, tenant, user)
|
||||
if keyword:
|
||||
q = q.filter(or_(Project.project_name.contains(keyword), Project.project_no.contains(keyword)))
|
||||
rows = q.with_entities(Project.id, Project.project_name, Project.project_no,
|
||||
@@ -386,15 +396,16 @@ def options(db, tenant, user, biz_type, keyword=None):
|
||||
|
||||
|
||||
def filter_options(db, tenant, user, biz_type, keyword=None, *, business_scope=None):
|
||||
"""Read filters reflect accessible business history, independent of membership/status."""
|
||||
"""Intersect project-choice scope with visible history; include stopped projects."""
|
||||
if biz_type not in BUSINESS:
|
||||
raise HTTPException(400, '业务类型不支持')
|
||||
model = BUSINESS[biz_type][0]
|
||||
visible = business_scope if business_scope is not None else business_query(db, tenant, user, biz_type)
|
||||
ids = visible.with_entities(model.project_id).filter(model.project_id.isnot(None)).distinct()
|
||||
q = scoped(db, Project, tenant).outerjoin(ProjectTeam, and_(
|
||||
q = scoped(db, Project, tenant).join(ProjectTeam, and_(
|
||||
ProjectTeam.id == Project.team_id, ProjectTeam.tenant_no == tenant,
|
||||
ProjectTeam.del_flag == 0)).filter(Project.id.in_(ids))
|
||||
q = apply_project_choice_scope(q, db, tenant, user)
|
||||
if keyword:
|
||||
q = q.filter(or_(Project.project_name.contains(keyword), Project.project_no.contains(keyword)))
|
||||
rows = q.with_entities(Project.id, Project.project_name, Project.project_no,
|
||||
|
||||
@@ -170,3 +170,32 @@ def test_mobile_history_filter_options_keep_field_scope(db, role):
|
||||
query = apply_presign_scope(db.query(BizPresignOrder), db, _ctx(role, user['user_id']), user_info=user)
|
||||
result = filter_options(db, 'tenant-a', user, 'presign', business_scope=query)
|
||||
assert [row['id'] for row in result['list']] == ['project']
|
||||
|
||||
@pytest.mark.parametrize('kind', ['presign', 'cell_storage', 'research_donation'])
|
||||
def test_mobile_cross_team_assignment_requires_explicit_all_permission(db, kind):
|
||||
from app.models.models import SysMenu, SysRole, SysRoleMenu, SysUserRole
|
||||
# Becoming a project leader does not make the user a member of its team.
|
||||
db.get(ProjectTeam, 'team').leader_user_id = 'u2'
|
||||
db.add_all([
|
||||
SysRole(id='assignment-role', tenant_no='tenant-a', role_name='归属维护', data_scope='4'),
|
||||
SysUserRole(id='assignment-user-role', tenant_no='tenant-a', user_id='u1', role_id='assignment-role'),
|
||||
SysMenu(id='assignment-menu', tenant_no='tenant-a', menu_name='归属维护', perms='project:assignment:edit'),
|
||||
SysRoleMenu(id='assignment-role-menu', tenant_no='tenant-a', role_id='assignment-role', menu_id='assignment-menu'),
|
||||
])
|
||||
db.commit()
|
||||
|
||||
def create():
|
||||
if kind == 'presign':
|
||||
return market_presign.create_presign_order(db, body=MarketPresignOrderBody(motherName='母亲', projectId='project'), ctx=_ctx('specialist'), user_info={**_user(), 'role': 'admin'})
|
||||
return market_protocol.create_protocol(db, body=_body(kind, projectId='project', samples=[], attachments={}), ctx=_ctx('specialist'), user_info={**_user(), 'role': 'admin'})
|
||||
|
||||
with pytest.raises(HTTPException) as caught:
|
||||
create()
|
||||
assert caught.value.status_code == 403
|
||||
db.add_all([
|
||||
SysMenu(id='all-menu', tenant_no='tenant-a', menu_name='全部项目', menu_type='F', visible='0', perms='project:project:all'),
|
||||
SysRoleMenu(id='all-role-menu', tenant_no='tenant-a', role_id='assignment-role', menu_id='all-menu'),
|
||||
])
|
||||
db.commit()
|
||||
result = create()
|
||||
assert (result if kind == 'presign' else result['basic'])['projectId'] == 'project'
|
||||
@@ -109,12 +109,12 @@ class ProjectAssignmentIntegrationTest(unittest.TestCase):
|
||||
self.assertEqual(list_entities(self.db, 't', ProjectTeam, keyword='', leaderUserId='', enabled=None)['total'], 1)
|
||||
self.assertEqual(list_entities(self.db, 't', ProjectTeam, enabled=False)['total'], 0)
|
||||
|
||||
def test_history_filters_follow_business_access_not_membership(self):
|
||||
def test_history_filters_keep_stopped_projects_within_membership_and_business_access(self):
|
||||
from app.models.project_management import Project, ProjectTeam
|
||||
from app.models.models import SysRole, SysUserRole
|
||||
from app.services.project_management import filter_options, options
|
||||
self.db.add_all([
|
||||
ProjectTeam(id='g', tenant_no='t', team_name='停用组', leader_user_id='other', enabled=0),
|
||||
ProjectTeam(id='g', tenant_no='t', team_name='停用组', leader_user_id='u', enabled=0),
|
||||
Project(id='p', tenant_no='t', project_no='P', project_name='历史项目', team_id='g', leader_user_id='other', biz_types='cell_storage', status='disabled'),
|
||||
Project(id='hidden', tenant_no='t', project_no='H', project_name='不可见项目', team_id='g', leader_user_id='other', biz_types='cell_storage', status='active'),
|
||||
SysRole(id='role', role_name='member', tenant_no='t', data_scope='4'),
|
||||
|
||||
@@ -154,7 +154,7 @@ class ProjectManagementTest(unittest.TestCase):
|
||||
SysRole(id='role2', role_name='assignment', tenant_no='t', data_scope='4'),
|
||||
SysUserRole(id='ur1', user_id='outsider', role_id='role1', tenant_no='t'),
|
||||
SysUserRole(id='ur2', user_id='outsider', role_id='role2', tenant_no='t'),
|
||||
SysMenu(id='menu1', tenant_no='t', menu_name='归属', perms='project:assignment:edit'),
|
||||
SysMenu(id='menu1', tenant_no='t', menu_name='全部项目', perms='project:project:all'),
|
||||
SysRoleMenu(id='rm1', tenant_no='t', role_id='role2', menu_id='menu1'),
|
||||
BizCollectionProtocol(id='mine', protocol_no='Mine', tenant_no='t', create_by='outsider', project_id='p'),
|
||||
BizCollectionProtocol(id='foreign', protocol_no='Other', tenant_no='t', create_by='u', project_id='p')])
|
||||
|
||||
@@ -0,0 +1,180 @@
|
||||
"""Users cannot self-grant project access through the permission administration APIs."""
|
||||
import pytest
|
||||
from fastapi import FastAPI
|
||||
from fastapi.testclient import TestClient
|
||||
from sqlalchemy import create_engine
|
||||
from sqlalchemy.orm import Session
|
||||
from sqlalchemy.pool import StaticPool
|
||||
|
||||
from app.api.admin import router
|
||||
from app.core.auth import get_current_user
|
||||
from app.core.database import get_db
|
||||
from app.models.base import Base
|
||||
from app.models.models import SysMenu, SysPost, SysRole, SysRoleMenu, SysUser, SysUserRole
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def env():
|
||||
engine = create_engine('sqlite://', connect_args={'check_same_thread': False}, poolclass=StaticPool)
|
||||
Base.metadata.create_all(engine)
|
||||
db = Session(engine)
|
||||
user = dict(user_id='u', role='admin', tenant_no='t')
|
||||
db.add_all([
|
||||
SysUser(id='u', tenant_no='t', user_no='member', nick_name='Member'),
|
||||
SysUser(id='foreign-user', tenant_no='foreign', user_no='foreign', nick_name='Foreign'),
|
||||
SysRole(id='role', tenant_no='t', role_name='Member', data_scope='4'),
|
||||
SysRole(id='foreign-role', tenant_no='foreign', role_name='Foreign'),
|
||||
SysUserRole(id='ur', tenant_no='t', user_id='u', role_id='role'),
|
||||
SysPost(id='post', tenant_no='t', post_name='Post', post_code='POST'),
|
||||
SysPost(id='foreign-post', tenant_no='foreign', post_name='Foreign', post_code='FOREIGN'),
|
||||
SysMenu(id='root', tenant_no='t', parent_id='0', menu_name='Web'),
|
||||
SysMenu(id='project', tenant_no='t', parent_id='root', menu_name='项目', perms='menu:project'),
|
||||
SysMenu(id='all', tenant_no='t', parent_id='project', menu_name='全部项目', menu_type='F', perms='project:project:all'),
|
||||
SysMenu(id='foreign-menu', tenant_no='foreign', parent_id='0', menu_name='Foreign'),
|
||||
])
|
||||
db.commit()
|
||||
app = FastAPI()
|
||||
app.include_router(router)
|
||||
app.dependency_overrides[get_current_user] = lambda: user
|
||||
app.dependency_overrides[get_db] = lambda: db
|
||||
with TestClient(app) as client:
|
||||
yield db, user, client
|
||||
db.close()
|
||||
engine.dispose()
|
||||
|
||||
|
||||
def grant(db, code):
|
||||
db.add_all([SysMenu(id=code, tenant_no='t', menu_name=code, perms=code),
|
||||
SysRoleMenu(id=code, tenant_no='t', role_id='role', menu_id=code)])
|
||||
db.commit()
|
||||
|
||||
|
||||
@pytest.mark.parametrize('method,path,body', [
|
||||
('PUT', '/role-management/role/menus', {'menuIds': ['all']}),
|
||||
('POST', '/role-management', {'roleName': '超级管理员'}),
|
||||
('PUT', '/role-management/role', {'roleName': '超级管理员'}),
|
||||
('DELETE', '/role-management/role', None),
|
||||
('PUT', '/role-management/role/data-scope', {'dataScope': '1'}),
|
||||
('POST', '/users', {'userNo': 'attacker', 'nickName': 'Attacker', 'password': 'TestOnly123!', 'roleIds': ['role']}),
|
||||
('PUT', '/users/u', {'roleIds': ['foreign-role']}),
|
||||
('PUT', '/users/u', {'postIds': ['post']}),
|
||||
('POST', '/users/u/reset-password', None),
|
||||
('DELETE', '/users/u', None),
|
||||
('PUT', '/posts/post/roles', {'roleIds': ['role']}),
|
||||
('PUT', '/posts/post/users', {'userIds': ['u']}),
|
||||
('PUT', '/posts/post', {'status': '1'}),
|
||||
('POST', '/menus', {'menuName': 'Grant', 'perms': 'project:project:all'}),
|
||||
('PUT', '/menus/project', {'perms': 'project:project:all'}),
|
||||
('DELETE', '/menus/all', None),
|
||||
])
|
||||
def test_business_admin_cannot_mutate_permission_graph(env, method, path, body):
|
||||
_, _, client = env
|
||||
result = client.request(method, '/web/admin' + path, json=body)
|
||||
assert result.status_code == 403
|
||||
|
||||
|
||||
def test_role_manager_grants_leaf_and_ancestors_only(env):
|
||||
db, _, client = env
|
||||
grant(db, 'system:role:list')
|
||||
result = client.put('/web/admin/role-management/role/menus', json={'menuIds': ['all', 'system:role:list']})
|
||||
assert result.json()['code'] == 200
|
||||
rows = db.query(SysRoleMenu).filter_by(role_id='role', tenant_no='t', del_flag=0).all()
|
||||
assert {r.menu_id for r in rows} == {'all', 'project', 'root', 'system:role:list'}
|
||||
# Same save is idempotent and does not create duplicate active grants.
|
||||
assert client.put('/web/admin/role-management/role/menus', json={'menuIds': ['all', 'system:role:list']}).json()['code'] == 200
|
||||
assert db.query(SysRoleMenu).filter_by(role_id='role', tenant_no='t', del_flag=0).count() == 4
|
||||
|
||||
|
||||
@pytest.mark.parametrize('super_admin', [False, True])
|
||||
@pytest.mark.parametrize('path,body', [
|
||||
('/role-management/foreign-role/menus', {'menuIds': ['all']}),
|
||||
('/role-management/role/menus', {'menuIds': ['all', 'foreign-menu']}),
|
||||
('/role-management/foreign-role', {'roleName': 'Changed'}),
|
||||
('/role-management/foreign-role/data-scope', {'dataScope': '1'}),
|
||||
('/menus/foreign-menu', {'perms': 'project:project:all'}),
|
||||
('/menus/project', {'parentId': 'foreign-menu'}),
|
||||
('/posts/foreign-post', {'status': '1'}),
|
||||
('/users/foreign-user', {'roleIds': ['role']}),
|
||||
('/users/u', {'roleIds': ['foreign-role']}),
|
||||
('/users/u', {'postIds': ['foreign-post']}),
|
||||
])
|
||||
def test_permission_admin_cannot_write_foreign_targets_or_links(env, path, body, super_admin):
|
||||
db, user, client = env
|
||||
for code in ('system:role:list', 'system:menu:list', 'system:post:list', 'system:user:list'):
|
||||
grant(db, code)
|
||||
if super_admin:
|
||||
user['role'] = 'super_admin'
|
||||
before = {r.menu_id for r in db.query(SysRoleMenu).filter_by(role_id='role', del_flag=0)}
|
||||
result = client.put('/web/admin' + path, json=body)
|
||||
assert result.json().get('code', result.status_code) in (400, 403, 404)
|
||||
db.expire_all()
|
||||
assert {r.menu_id for r in db.query(SysRoleMenu).filter_by(role_id='role', del_flag=0)} == before
|
||||
assert db.get(SysRole, 'foreign-role').role_name == 'Foreign'
|
||||
assert db.get(SysUser, 'foreign-user').tenant_no == 'foreign'
|
||||
|
||||
|
||||
@pytest.mark.parametrize('parent', ['missing', 'all'])
|
||||
def test_role_grant_rejects_broken_or_cyclic_ancestry_atomically(env, parent):
|
||||
db, user, client = env
|
||||
user['role'] = 'super_admin'
|
||||
db.get(SysMenu, 'project').parent_id = parent
|
||||
db.commit()
|
||||
result = client.put('/web/admin/role-management/role/menus', json={'menuIds': ['all']})
|
||||
assert result.json().get('code', result.status_code) == 400
|
||||
assert db.query(SysRoleMenu).count() == 0
|
||||
|
||||
@pytest.mark.parametrize('code,path,body', [
|
||||
('system:role:list', '/role-management/role', {'remark': 'authorized'}),
|
||||
('system:role:list', '/role-management/role/data-scope', {'dataScope': '4'}),
|
||||
('system:user:list', '/users/u', {'roleIds': ['role'], 'postIds': ['post']}),
|
||||
('system:post:list', '/posts/post/roles', {'roleIds': ['role']}),
|
||||
('system:post:list', '/posts/post/users', {'userIds': ['u']}),
|
||||
('system:menu:list', '/menus/all', {'menuName': '全部项目'}),
|
||||
])
|
||||
def test_authorized_tenant_administrators_can_manage_their_module(env, code, path, body):
|
||||
db, _, client = env
|
||||
grant(db, code)
|
||||
assert client.put('/web/admin' + path, json=body).json()['code'] == 200
|
||||
|
||||
|
||||
def test_web_and_mobile_ancestor_roots_remain_separate(env):
|
||||
db, user, client = env
|
||||
user['role'] = 'super_admin'
|
||||
db.add_all([
|
||||
SysMenu(id='mobile-root', tenant_no='t', parent_id='0', menu_name='Mobile'),
|
||||
SysMenu(id='mobile-leaf', tenant_no='t', parent_id='mobile-root', menu_name='Mobile business'),
|
||||
])
|
||||
db.commit()
|
||||
assert client.put('/web/admin/role-management/role/menus', json={'menuIds': ['all', 'mobile-leaf']}).json()['code'] == 200
|
||||
assert {r.menu_id for r in db.query(SysRoleMenu).filter_by(role_id='role', del_flag=0)} == {'all', 'project', 'root', 'mobile-leaf', 'mobile-root'}
|
||||
|
||||
@pytest.mark.parametrize('method,path,body', [
|
||||
('POST', '/qc-material-amount-auth/users', {'userId': 'u'}),
|
||||
('DELETE', '/qc-material-amount-auth/users/u', None),
|
||||
('POST', '/qc-material-amount-auth/users/batch-revoke', {'userIds': ['u']}),
|
||||
])
|
||||
def test_business_admin_cannot_self_bind_specialized_authorization_roles(env, method, path, body):
|
||||
from app.api.qc import router as qc_router
|
||||
_, _, client = env
|
||||
client.app.include_router(qc_router)
|
||||
assert client.request(method, '/web/admin' + path, json=body).status_code == 403
|
||||
|
||||
|
||||
def test_specialized_authorization_manager_can_grant_and_revoke(env):
|
||||
from app.api.qc import router as qc_router
|
||||
from app.core.qc_material_amount_auth_constants import (
|
||||
QC_MATERIAL_AMOUNT_AUTH_LIST_PERM, QC_MATERIAL_AMOUNT_VIEW_ROLE_ID,
|
||||
QC_MATERIAL_AMOUNT_VIEW_ROLE_NAME, QC_MATERIAL_AMOUNT_AUTH_DEPT_NAME,
|
||||
)
|
||||
from app.models.models import SysDept
|
||||
db, _, client = env
|
||||
client.app.include_router(qc_router)
|
||||
grant(db, QC_MATERIAL_AMOUNT_AUTH_LIST_PERM)
|
||||
db.add_all([
|
||||
SysDept(id='qc-dept', tenant_no='t', dept_name=QC_MATERIAL_AMOUNT_AUTH_DEPT_NAME),
|
||||
SysRole(id=QC_MATERIAL_AMOUNT_VIEW_ROLE_ID, tenant_no='t', role_name=QC_MATERIAL_AMOUNT_VIEW_ROLE_NAME),
|
||||
])
|
||||
db.get(SysUser, 'u').dept_id = 'qc-dept'
|
||||
db.commit()
|
||||
assert client.post('/web/admin/qc-material-amount-auth/users', json={'userId': 'u'}).json()['code'] == 200
|
||||
assert client.delete('/web/admin/qc-material-amount-auth/users/u').json()['code'] == 200
|
||||
@@ -0,0 +1,176 @@
|
||||
"""Project choices and writes share explicit team/all-project permission boundaries."""
|
||||
import unittest
|
||||
|
||||
from fastapi import HTTPException
|
||||
from sqlalchemy import create_engine
|
||||
from sqlalchemy.orm import Session
|
||||
|
||||
from app.models.base import Base
|
||||
from app.models.models import (BizCollectionProtocol, SysMenu, SysRole, SysRoleMenu,
|
||||
SysUserRole, SysPost, SysPostRole, SysUserPost)
|
||||
from app.models.project_management import Project, ProjectTeam, ProjectTeamMember
|
||||
from app.services import project_management as svc
|
||||
from app.services.project_assignment import create_assignment
|
||||
|
||||
|
||||
class ProjectSelectionPermissionsTest(unittest.TestCase):
|
||||
def setUp(self):
|
||||
self.engine = create_engine('sqlite:///:memory:')
|
||||
Base.metadata.create_all(self.engine)
|
||||
self.db = Session(self.engine)
|
||||
self.user = dict(user_id='member', role='admin', tenant_no='t')
|
||||
kinds = ','.join(svc.COUNT_TYPES)
|
||||
self.db.add_all([
|
||||
ProjectTeam(id='own', tenant_no='t', team_name='Own', leader_user_id='leader'),
|
||||
ProjectTeam(id='other', tenant_no='t', team_name='Other', leader_user_id='other'),
|
||||
ProjectTeam(id='foreign', tenant_no='foreign', team_name='Foreign', leader_user_id='member'),
|
||||
ProjectTeamMember(id='member', tenant_no='t', team_id='own', user_id='member'),
|
||||
Project(id='own-p', tenant_no='t', project_no='OWN', project_name='Own', team_id='own', leader_user_id='other', status='active', biz_types=kinds),
|
||||
Project(id='other-p', tenant_no='t', project_no='OTHER', project_name='Other', team_id='other', leader_user_id='other', status='active', biz_types=kinds),
|
||||
Project(id='foreign-p', tenant_no='foreign', project_no='FOREIGN', project_name='Foreign', team_id='foreign', leader_user_id='member', status='active', biz_types=kinds),
|
||||
])
|
||||
self.db.commit()
|
||||
|
||||
def tearDown(self):
|
||||
self.db.close()
|
||||
self.engine.dispose()
|
||||
|
||||
def grant(self, code, suffix='all', *, inherited=False):
|
||||
self.db.add_all([
|
||||
SysRole(id=suffix, tenant_no='t', role_name=suffix, data_scope='4'),
|
||||
SysMenu(id=suffix, tenant_no='t', menu_name=suffix, menu_type='F', visible='0', perms=code),
|
||||
SysRoleMenu(id=suffix, tenant_no='t', role_id=suffix, menu_id=suffix),
|
||||
])
|
||||
if inherited:
|
||||
self.db.add_all([
|
||||
SysPost(id=suffix, tenant_no='t', post_name=suffix, post_code=suffix),
|
||||
SysPostRole(id=suffix, tenant_no='t', post_id=suffix, role_id=suffix),
|
||||
SysUserPost(id=suffix, tenant_no='t', user_id='member', post_id=suffix),
|
||||
])
|
||||
else:
|
||||
self.db.add(SysUserRole(id=suffix, tenant_no='t', user_id='member', role_id=suffix))
|
||||
self.db.commit()
|
||||
|
||||
def choices(self, user=None, kind='cell_storage'):
|
||||
return {r['id'] for r in svc.options(self.db, 't', user or self.user, kind)['list']}
|
||||
|
||||
def assert_forbidden(self, project_id='other-p', kind='cell_storage'):
|
||||
with self.assertRaises(HTTPException) as caught:
|
||||
create_assignment(self.db, 't', self.user, project_id, kind)
|
||||
self.assertEqual(caught.exception.status_code, 403)
|
||||
|
||||
def test_project_leader_alone_does_not_bypass_team_membership(self):
|
||||
self.db.get(Project, 'other-p').leader_user_id = 'member'
|
||||
self.db.commit()
|
||||
self.assertEqual(self.choices(), {'own-p'})
|
||||
self.assert_forbidden()
|
||||
|
||||
def test_team_leader_and_membership_union(self):
|
||||
self.assertEqual(self.choices(dict(user_id='leader', role='admin')), {'own-p'})
|
||||
self.db.add(ProjectTeamMember(id='second', tenant_no='t', team_id='other', user_id='member'))
|
||||
self.db.commit()
|
||||
self.assertEqual(self.choices(), {'own-p', 'other-p'})
|
||||
self.db.get(ProjectTeamMember, 'second').del_flag = 1
|
||||
self.db.commit()
|
||||
self.assertEqual(self.choices(), {'own-p'})
|
||||
self.assert_forbidden()
|
||||
|
||||
def test_admin_with_no_team_has_no_choices(self):
|
||||
self.assertEqual(self.choices(dict(user_id='outsider', role='admin')), set())
|
||||
|
||||
def test_assignment_edit_does_not_grant_all_projects(self):
|
||||
self.grant('project:assignment:edit', 'assignment')
|
||||
self.assertEqual(self.choices(), {'own-p'})
|
||||
self.assert_forbidden()
|
||||
|
||||
def test_all_permission_in_role_union_allows_all_business_types(self):
|
||||
self.grant('project:assignment:edit', 'assignment')
|
||||
self.grant('project:project:all')
|
||||
for kind in svc.BUSINESS:
|
||||
with self.subTest(kind=kind):
|
||||
self.assertEqual(self.choices(kind=kind), {'own-p', 'other-p'})
|
||||
self.assertEqual(create_assignment(self.db, 't', self.user, 'other-p', kind), 'other-p')
|
||||
with self.assertRaises(HTTPException) as caught:
|
||||
create_assignment(self.db, 't', self.user, 'foreign-p', 'cell_storage')
|
||||
self.assertEqual(caught.exception.status_code, 404)
|
||||
|
||||
def test_all_permission_does_not_bypass_project_state_or_business_type(self):
|
||||
self.grant('project:project:all')
|
||||
self.db.get(Project, 'other-p').status = 'disabled'
|
||||
self.db.commit()
|
||||
self.assertEqual(self.choices(), {'own-p'})
|
||||
with self.assertRaises(HTTPException):
|
||||
create_assignment(self.db, 't', self.user, 'other-p', 'cell_storage')
|
||||
self.db.get(Project, 'other-p').status = 'active'
|
||||
self.db.get(Project, 'other-p').biz_types = 'basic_experiment'
|
||||
self.db.commit()
|
||||
self.assertEqual(self.choices(), {'own-p'})
|
||||
with self.assertRaises(HTTPException):
|
||||
create_assignment(self.db, 't', self.user, 'other-p', 'cell_storage')
|
||||
|
||||
def test_permission_revocation_is_effective_on_next_request(self):
|
||||
self.grant('project:project:all')
|
||||
self.assertEqual(self.choices(), {'own-p', 'other-p'})
|
||||
for model, field, disabled, enabled in [
|
||||
(SysRole, 'status', '0', '1'), (SysRoleMenu, 'del_flag', 1, 0),
|
||||
(SysMenu, 'status', '0', '1'), (SysUserRole, 'del_flag', 1, 0),
|
||||
]:
|
||||
with self.subTest(model=model.__name__):
|
||||
setattr(self.db.get(model, 'all'), field, disabled)
|
||||
self.db.commit()
|
||||
self.assertEqual(self.choices(), {'own-p'})
|
||||
self.assert_forbidden()
|
||||
setattr(self.db.get(model, 'all'), field, enabled)
|
||||
self.db.commit()
|
||||
|
||||
def test_post_inherited_all_permission_and_disabled_post(self):
|
||||
self.grant('project:project:all', inherited=True)
|
||||
self.assertEqual(self.choices(), {'own-p', 'other-p'})
|
||||
self.assertEqual(create_assignment(self.db, 't', self.user, 'other-p', 'presign'), 'other-p')
|
||||
self.db.get(SysPost, 'all').status = '0'
|
||||
self.db.commit()
|
||||
self.assertEqual(self.choices(), {'own-p'})
|
||||
self.assert_forbidden()
|
||||
|
||||
def test_filter_options_require_team_scope_and_preserve_business_scope(self):
|
||||
self.grant('project:assignment:edit', 'assignment')
|
||||
self.db.add_all([
|
||||
BizCollectionProtocol(id='own-record', protocol_no='OWN', tenant_no='t', create_by='member', project_id='own-p'),
|
||||
BizCollectionProtocol(id='other-record', protocol_no='OTHER', tenant_no='t', create_by='member', project_id='other-p'),
|
||||
BizCollectionProtocol(id='hidden-record', protocol_no='HIDDEN', tenant_no='t', create_by='other', project_id='own-p'),
|
||||
])
|
||||
self.db.commit()
|
||||
def filters(**kwargs):
|
||||
return {r['id'] for r in svc.filter_options(self.db, 't', self.user, 'cell_storage', **kwargs)['list']}
|
||||
self.assertEqual(filters(), {'own-p'})
|
||||
self.grant('project:project:all')
|
||||
self.assertEqual(filters(), {'own-p', 'other-p'})
|
||||
self.assertEqual(svc.records(self.db, 't', self.user, 'cell_storage')['total'], 2)
|
||||
# Mobile passes its stricter business query to the same options service.
|
||||
mobile_scope = self.db.query(BizCollectionProtocol).filter(BizCollectionProtocol.id == 'own-record')
|
||||
self.assertEqual(filters(business_scope=mobile_scope), {'own-p'})
|
||||
|
||||
def test_cross_group_reassignment_needs_both_permissions(self):
|
||||
self.grant('project:assignment:edit', 'assignment')
|
||||
self.db.add(BizCollectionProtocol(id='record', protocol_no='R', tenant_no='t', create_by='member', project_id='own-p'))
|
||||
self.db.commit()
|
||||
with self.assertRaises(HTTPException) as caught:
|
||||
svc.preview_assignment(self.db, 't', self.user, 'cell_storage', 'record', 'other-p')
|
||||
self.assertEqual(caught.exception.status_code, 403)
|
||||
self.grant('project:project:all')
|
||||
preview = svc.preview_assignment(self.db, 't', self.user, 'cell_storage', 'record', 'other-p')
|
||||
self.db.get(SysRoleMenu, 'all').del_flag = 1
|
||||
self.db.commit()
|
||||
with self.assertRaises(HTTPException) as caught:
|
||||
svc.assign(self.db, 't', self.user, 'cell_storage', 'record', 'other-p', 'reason', preview['records'])
|
||||
self.assertEqual(caught.exception.status_code, 403)
|
||||
self.assertEqual(self.db.get(BizCollectionProtocol, 'record').project_id, 'own-p')
|
||||
|
||||
def test_all_permission_does_not_grant_maintenance_or_management_actions(self):
|
||||
self.grant('project:project:all')
|
||||
for code in ('project:assignment:edit', 'project:project:edit', 'project:team:edit'):
|
||||
with self.subTest(code=code), self.assertRaises(HTTPException):
|
||||
svc.check_permission(self.db, 't', self.user, code)
|
||||
|
||||
def test_super_admin_retains_tenant_boundary(self):
|
||||
self.assertEqual(self.choices(dict(user_id='root', role='super_admin')), {'own-p', 'other-p'})
|
||||
@@ -0,0 +1,11 @@
|
||||
-- Issue #13:在 project_top_menu 之后执行,可重复执行。
|
||||
-- 独立功能权限,挂在“项目”一级节点下,授权时无需同时授予项目资料管理权限。
|
||||
-- 不从归属维护或管理员角色继承授权,不写 sys_role_menu;由角色管理显式勾选。
|
||||
INSERT INTO sys_menu (id, menu_name, parent_id, menu_type, path, component, perms,
|
||||
is_frame, visible, status, tenant_no, order_num, create_dept_id, create_time, update_time, del_flag)
|
||||
SELECT MD5(CONCAT(parent.tenant_no, ':project:project:all')), '全部项目', parent.id,
|
||||
'F', '', '', 'project:project:all', 1, '0', '1', parent.tenant_no, 20, '', NOW(), NOW(), 0
|
||||
FROM sys_menu parent
|
||||
WHERE parent.perms='menu:project' AND parent.del_flag=0 AND parent.status='1'
|
||||
AND NOT EXISTS (SELECT 1 FROM sys_menu existing
|
||||
WHERE existing.tenant_no=parent.tenant_no AND existing.perms='project:project:all');
|
||||
@@ -8,7 +8,9 @@
|
||||
|
||||
## 权限与兼容
|
||||
|
||||
管理功能使用 project:team:list/edit、project:project:list/edit、project:statistics:list、project:assignment:edit 菜单权限,复用有效角色并集;super_admin 仍限定本次认证租户。代录选任意项目需 project:assignment:edit,不用普通 admin 身份自动放开。订单数据统计、明细及归属调整应用原有 apply_data_scope;项目成员身份不扩权。
|
||||
管理功能使用 project:team:list/edit、project:project:list/edit、project:statistics:list、project:assignment:edit 菜单权限,复用有效角色并集;super_admin 仍限定本次认证租户。跨组选择使用独立功能权限 `project:project:all`(全部项目),在现有角色菜单树的“项目”下分配,不默认授予现有角色。`project:assignment:edit` 仅允许维护操作,不能代替全部项目权限;普通 admin 不自动放开。super_admin 沿用系统全部权限语义,但始终受当前认证租户限制。订单数据统计、明细及归属调整应用原有 apply_data_scope;项目成员身份不扩权。
|
||||
|
||||
授权维护不能由普通业务管理员自授:系统用户、角色、职位、菜单的变更接口分别严格校验 `system:user:list`、`system:role:list`、`system:post:list`、`system:menu:list`,沿用有效角色并集与 super_admin 全权限语义。角色菜单和数据范围读取校验管理权限及目标租户;通用角色/人员下拉保留原有读取用途。目标和关联 ID 均验证认证租户,super_admin 在这些接口也不跨租户。角色菜单保存补齐每个已选节点的同租户祖先(包含各自 Web/移动虚拟根),拒绝断链或循环,不自动勾选兄弟/子权限。质检金额专用角色的授予/撤销/批量撤销同样使用其既有专用管理权限严格校验,阻止另一条用户角色绑定旁路。
|
||||
|
||||
新单据项目必填由 PROJECT_ASSIGNMENT_REQUIRED 配置控制,默认 false 供多端滚动升级;全部入口更新且项目资料准备好后由运维设为 true。数据库历史 NULL 继续合法。编辑未传 projectId 时保留旧值;普通表单不可改变已有非空归属,历史补录与归属调整统一从维护入口审计。未改变归属的历史单据不重验成员或启用状态。
|
||||
|
||||
@@ -19,7 +21,7 @@ validate_assignment(db, tenant_no, user, project_id, biz_type, *, current_projec
|
||||
project_brief(db, tenant_no, project_id) -> dict # projectId/projectName/teamId/teamName
|
||||
```
|
||||
|
||||
用户 user 使用认证上下文字典。调用方在创建时传 required=settings.PROJECT_ASSIGNMENT_REQUIRED;传入项目必须存在、启用、组启用、业务类型匹配且用户为负责人/成员或有代录权限。普通更新若归属不同返回“请通过项目归属维护调整”。转换复制来源项目,不按转换操作人重新分配;转换的目标业务类型仍允许保留来源项目历史归属。
|
||||
用户 user 使用认证上下文字典。调用方在创建时传 required=settings.PROJECT_ASSIGNMENT_REQUIRED;传入项目必须存在、启用、组启用、业务类型匹配且用户为项目组负责人/有效成员或有全部项目权限(仅项目负责人身份不替代项目组成员资格)。普通更新若归属不同返回“请通过项目归属维护调整”。转换复制来源项目,不按转换操作人重新分配;转换的目标业务类型仍允许保留来源项目历史归属。
|
||||
|
||||
## HTTP契约
|
||||
|
||||
@@ -52,6 +54,6 @@ Web P01/P02/P09,参考 system/unit/index.vue、cell-storage/protocol/ProtocolF
|
||||
|
||||
### 查询筛选与新建选择的区别
|
||||
|
||||
公共 options 默认用于新建/归属调整,校验项目启用和成员资格。`usage=filter` 按现有可见业务主单返回历史项目(包含完成/停用),不因成员变更丢失筛选能力。
|
||||
公共 options 默认用于新建/归属调整,校验项目启用、支持业务类型和项目组成员资格或全部项目权限。`usage=filter` 同样限制项目组/全部项目范围,再按现有可见业务主单返回历史项目(包含完成/停用),不扩大订单数据权限。成员退出后不再作为可选项;已有单据的项目名称仍由单据响应回显,普通编辑不清空历史归属。
|
||||
移动列表使用 `/mini/market/{specialist|manager}/project-options?bizType=...`,复用对应菜单权限及现场数据范围;不借公共筛选扩大现场范围。
|
||||
普通创建不能伪造来源预签约/转换来源,普通更新不得改来源关联;只有受控转换入口维护来源图。
|
||||
@@ -29,3 +29,22 @@ Web与移动 SHALL 复用同端现有组件、布局模式和全局样式。
|
||||
#### Scenario: 新增管理界面
|
||||
- **WHEN** 实现已确认Figma布局
|
||||
- **THEN** 列表、弹窗、按钮、标签及选择器遵循现有界面,不引入平行主题。
|
||||
|
||||
### Requirement: 所属项目范围与独立全部项目权限
|
||||
系统 SHALL 在 Web 和移动的所属项目候选及后端归属写入中使用同一范围:本人作为项目组负责人或有效成员所在组的项目,或显式 `project:project:all` 授权下的当前租户项目。多组、有效角色及职位继承角色取并集。普通管理员、项目负责人或归属维护权限本身不得绕过组成员限制;super_admin 沿用系统全部权限语义且不得跨租户。
|
||||
#### Scenario: 无跨组权限
|
||||
- **WHEN** 用户未加入某项目组且不是该组负责人
|
||||
- **THEN** 下拉框不返回其项目,伪造项目 ID 提交被后端拒绝;仅担任该项目负责人不构成组成员资格。
|
||||
#### Scenario: 分配或撤销全部项目
|
||||
- **WHEN** 角色或继承职位授予全部项目权限,或该授权被撤销/停用
|
||||
- **THEN** 后续候选和写入即时按有效授权判定,其他租户、停用项目和不支持业务类型仍不可新选。
|
||||
#### Scenario: 历史归属与数据权限
|
||||
- **WHEN** 用户调组或项目停用后查看、编辑历史订单
|
||||
- **THEN** 原项目仍可回显且无关编辑保留归属;筛选候选取项目范围与原业务数据范围交集,全部项目权限不扩大订单查看和维护权限。
|
||||
|
||||
#### Scenario: 禁止业务用户自行授权
|
||||
- **WHEN** 普通业务管理员直接调用角色菜单、用户角色/职位、职位角色/用户或菜单权限码等变更接口
|
||||
- **THEN** 缺少对应系统管理权限时返回拒绝;有管理权限者也只能变更当前租户目标和关联,用户账号重置/启停同样受管理权限保护。
|
||||
#### Scenario: 角色菜单保存
|
||||
- **WHEN** 管理员只勾选全部项目功能权限
|
||||
- **THEN** 保存其同租户祖先但不授予其他功能,断链/循环/跨租户节点导致整次保存拒绝。
|
||||
@@ -4,6 +4,13 @@
|
||||
**架构:** 新管理模块统一校验与统计;现有业务显式接入;前端复用已有样式。
|
||||
**技术栈:** FastAPI/SQLAlchemy/Pydantic、Vue3/Element Plus、Vue3/Vant。
|
||||
|
||||
用户追加所属项目权限要求:将跨组选择从归属维护权限分离为 `project:project:all`(全部项目),默认仅项目组负责人/有效成员所在组内项目;Web与移动共用后端范围校验。
|
||||
- [x] 在 `backend/tests/test_project_selection_permissions.py` 增加成员/多组/无组、独立全部项目权限、多角色及职位继承、撤权、跨租户和越权提交用例,先执行确认失败。
|
||||
- [x] 在 `backend/app/services/project_management.py` 统一候选范围,赋值校验改用独立权限;筛选仍叠加原订单数据范围,历史回显与转换继承保持原有规则。
|
||||
- [x] 增加幂等 `docs/sql/migrations/2026-11-02-project_all_permission.sql`,在项目一级节点下提供可分配的“全部项目”功能权限,不给现有角色自动授权;同步设计、规格及后端部署入口。
|
||||
- [x] 封堵实际验收发现的权限自授入口:系统用户/角色/职位/菜单变更接口检查对应严格管理权限,目标与引用按租户校验;角色菜单保存补齐祖先、拒绝断链/循环。`backend/tests/test_project_permission_administration.py` 覆盖拒绝、允许、租户和祖先边界。
|
||||
- [x] 运行项目权限与现有归属回归、字段一致性检查;隔离 MySQL 验证迁移和 HTTP 权限边界、浏览器角色配置与下拉框;独立审查后提交推送,继续保留待真机验收状态。
|
||||
|
||||
用户追加菜单调整:新增与市场/生产/质检平级的“项目”,四个项目管理入口迁入其中。沿用现有顶部导航组件及视觉样式,不新增业务页面。实施顺序:前后端菜单树同步 → 增量菜单迁移(保留叶子 ID/角色授权、补齐新祖先)→ 隔离 MySQL 幂等与角色边界验证 → 浏览器导航/旧地址跳转及构建验收。仍保留待真机验收、不合并的交付状态。
|
||||
|
||||
菜单调整验收:两租户 MySQL 迁移各重复执行通过,四子页/原 ID/祖先权限/无额外授权/移动根不变/无重复均核对;后端17项、Web5项测试和定向 ESLint 通过,Web生产构建通过。浏览器确认项目平级、市场移除四菜单及旧地址携带查询参数跳转;独立审查无P1/P2。
|
||||
@@ -30,3 +37,5 @@ Web 类型检查基线和改动版均 413 条诊断,无新增文件/错误码
|
||||
用户于 2026-11-02 明确授权自行切换服务,任务版本已在独立 MySQL 5.7.24 合成库运行,原业务数据库未修改。真实 HTTP/MySQL 62 项断言通过,覆盖五类主单新增、三种转换继承、四业务统计、租户/权限、并发归属冲突及审计;结构与菜单迁移重复执行通过,另验证首次 ADD 与旧迁移修正。
|
||||
浏览器已验收项目组/项目创建、Web/移动预签约关联保存、统计钻取及归属维护;375/414 宽度移动 H5 操作通过。验收中修复前端按钮权限机制、明细弹窗全局样式作用域及 MySQL 关联列排序规则。独立复审无 P1/P2。
|
||||
仍缺微信真机 WebView 联动、集成候选环境验证及合并后运行环境恢复;不能以移动浏览器模拟尺寸代替真机验收,当前不标记完成或合并。
|
||||
|
||||
所属项目权限追加验收:121项相关测试通过;测试先复现跨组及授权配置旁路再修复。隔离MySQL/HTTP共41项检查通过(29项项目/迁移+12项授权闭环),含3租户幂等种子、不自动授予角色、5类主单越权拒绝、移动创建拒绝、同会话授权/撤权、8条自授入口拒绝;浏览器确认角色管理可独立勾选全部项目且保存补祖先。独立复审无P1/P2。字段一致性仍为5项既有警告;本轮无前端源码变更,不重复前端构建,微信真机及集成验收继续待办。
|
||||
Reference in new issue
Block a user