From a54289250c87e2d72011c69485b3c1dd195e3f97 Mon Sep 17 00:00:00 2001 From: wxg0103 <727495428@qq.com> Date: Mon, 9 Nov 2026 10:31:39 +0800 Subject: [PATCH] fix: enforce detailed permissions for resource authorization --- .../auth/constants/permission_constants.py | 2 +- apps/system_manage/tests.py | 549 +++++++++++++++++- .../views/user_group_resource_permission.py | 69 ++- .../views/user_resource_permission.py | 72 ++- 4 files changed, 670 insertions(+), 22 deletions(-) diff --git a/apps/common/auth/constants/permission_constants.py b/apps/common/auth/constants/permission_constants.py index 0a166811181..97dbbb11cf5 100644 --- a/apps/common/auth/constants/permission_constants.py +++ b/apps/common/auth/constants/permission_constants.py @@ -269,7 +269,7 @@ class PermissionConstants(Enum): PermissionMeta( role_list=[RoleConstants.WORKSPACE_MANAGE, RoleConstants.USER], category=Category.WORKSPACE, - scope=[PermissionScopeConstants.WORKSPACE_RESOURCE], + scope=[PermissionScopeConstants.WORKSPACE, PermissionScopeConstants.WORKSPACE_RESOURCE], resource_permission_group_list=[ResourcePermissionConst.MODEL_MANAGE], ), ) diff --git a/apps/system_manage/tests.py b/apps/system_manage/tests.py index 3866389f841..2cdef66732d 100644 --- a/apps/system_manage/tests.py +++ b/apps/system_manage/tests.py @@ -1,16 +1,31 @@ from types import SimpleNamespace from unittest.mock import patch +from django.core.cache import cache +from django.http import QueryDict from django.test import SimpleTestCase, TestCase from common.auth.constants.permission_constants import PermissionConstants +from common.auth.constants.role_constants import RoleConstants +from common.auth.handle.impl.user_token import get_auth from common.auth.struct.aggregate_permission import AggregatePermission +from common.auth.struct.auth import Principal from common.auth.struct.permission import Permission +from common.constants.authentication_type import UserType +from common.constants.cache_version import Cache_Version +from common.database_model_manage.database_model_manage import DatabaseModelManage from common.exception.app_exception import AppUnauthorizedFailed +from role_setting.models import RolePermission, RoleSetting, UserRoleRelation from system_manage.models import ChatUser, UserGroup, UserGroupRelation +from system_manage.models.log_management import Log from system_manage.serializers.chat_user import ChatUserSerializer -from system_manage.views import resource_chat_user_group +from system_manage.views import ( + resource_chat_user_group, + user_group_resource_permission, + user_resource_permission, +) from system_manage.views.resource_chat_user import ResourceChatUser, get_chat_user_permission_constant +from users.models import User class ChatUserPermissionTests(SimpleTestCase): @@ -141,6 +156,538 @@ def test_extended_user_can_edit_chat_users_and_groups(self): group_edit.return_value.edit.assert_called_once_with([]) +class ResourcePermissionViewTests(SimpleTestCase): + workspace_id = "workspace-1" + other_workspace_id = "workspace-2" + resources = ("APPLICATION", "KNOWLEDGE", "TOOL", "MODEL") + resource_permission_bits = { + "APPLICATION": (1 << 0, 1 << 1), + "KNOWLEDGE": (1 << 2, 1 << 3), + "TOOL": (1 << 4, 1 << 5), + "MODEL": (1 << 6, 1 << 7), + } + + @staticmethod + def build_request(roles, permissions=None): + return SimpleNamespace( + query_params=QueryDict(), + data={}, + user=None, + META={}, + path="/", + auth=SimpleNamespace(roles=set(roles), permissions=permissions or {}), + ) + + def detailed_permissions(self, resource, operations): + read_bit, edit_bit = self.resource_permission_bits[resource] + bits = 0 + if "READ" in operations: + bits |= read_bit + if "EDIT" in operations: + bits |= edit_bit + return {"RESOURCE_PERMISSION": bits} + + def endpoint_calls(self, request, workspace_id, resource): + user_view = user_resource_permission.WorkSpaceUserResourcePermissionView() + user_group_view = user_group_resource_permission.WorkSpaceUserGroupResourcePermissionView() + user_kwargs = { + "workspace_id": workspace_id, + "user_id": "user-1", + "resource": resource, + } + user_group_kwargs = { + "workspace_id": workspace_id, + "user_group_id": "group-1", + "resource": resource, + } + return { + ("READ", "user list"): lambda: user_view.get(request, **user_kwargs), + ("READ", "user page"): lambda: user_view.Page.get( + user_view.Page(), request, **user_kwargs, current_page=1, page_size=20 + ), + ("EDIT", "user edit"): lambda: user_view.put(request, **user_kwargs), + ("READ", "user group list"): lambda: user_group_view.get(request, **user_group_kwargs), + ("READ", "user group page"): lambda: user_group_view.Page.get( + user_group_view.Page(), request, **user_group_kwargs, current_page=1, page_size=20 + ), + ("EDIT", "user group edit"): lambda: user_group_view.put(request, **user_group_kwargs), + } + + def assert_endpoint_operations(self, request, workspace_id, resource, allowed_operations): + with ( + patch.object(Log, "save"), + patch("system_manage.views.user_resource_permission.result.success", return_value=True), + patch("system_manage.views.user_resource_permission.UserResourcePermissionSerializer") as user_serializer, + patch( + "system_manage.views.user_group_resource_permission.UserGroupResourcePermissionSerializer" + ) as user_group_serializer, + ): + user_serializer.return_value.list.return_value = None + user_serializer.return_value.page.return_value = None + user_serializer.return_value.edit.return_value = None + user_group_serializer.return_value.list.return_value = None + user_group_serializer.return_value.page.return_value = None + user_group_serializer.return_value.edit.return_value = None + + for (operation, endpoint), call in self.endpoint_calls(request, workspace_id, resource).items(): + with self.subTest(resource=resource, endpoint=endpoint): + if operation in allowed_operations: + self.assertIs(True, call()) + else: + with self.assertRaises(AppUnauthorizedFailed): + call() + + def test_builtin_admin_can_access_all_resource_authorization_endpoints(self): + request = self.build_request({"ADMIN"}) + + for resource in self.resources: + with self.subTest(resource=resource): + self.assert_endpoint_operations( + request, + self.workspace_id, + resource, + {"READ", "EDIT"}, + ) + + def test_builtin_workspace_manager_can_access_own_workspace(self): + request = self.build_request( + { + "WORKSPACE_MANAGE", + f"WORKSPACE_MANAGE:w:{self.workspace_id}", + } + ) + + for resource in self.resources: + with self.subTest(resource=resource): + self.assert_endpoint_operations( + request, + self.workspace_id, + resource, + {"READ", "EDIT"}, + ) + + def test_builtin_workspace_manager_cannot_access_other_workspace(self): + request = self.build_request( + { + "WORKSPACE_MANAGE", + f"WORKSPACE_MANAGE:w:{self.workspace_id}", + } + ) + + for resource in self.resources: + with self.subTest(resource=resource): + self.assert_endpoint_operations(request, self.other_workspace_id, resource, set()) + + def test_inherited_roles_need_matching_read_and_edit_permissions(self): + roles_list = ( + {"EXTENDS_ADMIN"}, + {f"EXTENDS_WORKSPACE_MANAGE:w:{self.workspace_id}"}, + ) + + for roles in roles_list: + for resource in self.resources: + with self.subTest(roles=roles, resource=resource, permissions="none"): + self.assert_endpoint_operations( + self.build_request(roles), + self.workspace_id, + resource, + set(), + ) + with self.subTest(roles=roles, resource=resource, permissions="READ"): + self.assert_endpoint_operations( + self.build_request(roles, self.detailed_permissions(resource, {"READ"})), + self.workspace_id, + resource, + {"READ"}, + ) + with self.subTest(roles=roles, resource=resource, permissions="EDIT"): + self.assert_endpoint_operations( + self.build_request(roles, self.detailed_permissions(resource, {"EDIT"})), + self.workspace_id, + resource, + {"EDIT"}, + ) + + +class ResourceTargetPermissionViewTests(SimpleTestCase): + workspace_id = "workspace-1" + other_workspace_id = "workspace-2" + target = "target-1" + resources = ("APPLICATION", "KNOWLEDGE", "TOOL", "MODEL") + resource_authorization_bits = { + "APPLICATION": 1 << 10, + "KNOWLEDGE": 1 << 10, + "TOOL": 1 << 9, + "MODEL": 1 << 5, + } + system_authorization_permissions = { + "APPLICATION": ("SYSTEM_RESOURCE_APPLICATION", 1 << 5), + "KNOWLEDGE": ("SYSTEM_RESOURCE_KNOWLEDGE", 1 << 8), + "TOOL": ("SYSTEM_RESOURCE_TOOL", 1 << 5), + "MODEL": ("SYSTEM_RESOURCE_MODEL", 1 << 3), + } + + @staticmethod + def build_request(roles, permissions=None): + return SimpleNamespace( + query_params=QueryDict(), + data={}, + user=SimpleNamespace( + id="user-1", + type="SYSTEM_USER", + email="user@example.com", + phone="", + nick_name="User", + username="user-1", + ), + META={}, + path="/", + auth=SimpleNamespace(roles=set(roles), permissions=permissions or {}), + ) + + def detailed_permissions(self, resource, workspace_id, workspace_manage=False): + permission = {f"{resource}:w:{workspace_id}:r:{self.target}": self.resource_authorization_bits[resource]} + if workspace_manage: + permission[f"{resource}_WORKSPACE_MANAGE:w:{workspace_id}"] = self.resource_authorization_bits[resource] + return permission + + def system_authorization(self, resource): + key, bit = self.system_authorization_permissions[resource] + return {key: bit} + + def endpoint_calls(self, request, workspace_id, resource): + user_view = user_resource_permission.WorkspaceResourceUserPermissionView() + user_group_view = user_group_resource_permission.WorkspaceResourceUserGroupPermissionView() + kwargs = { + "workspace_id": workspace_id, + "target": self.target, + "resource": resource, + } + return { + "user list": lambda: user_view.get(request, **kwargs), + "user page": lambda: user_view.Page.get(user_view.Page(), request, **kwargs, current_page=1, page_size=20), + "user edit": lambda: user_view.put(request, **kwargs), + "user group list": lambda: user_group_view.get(request, **kwargs), + "user group page": lambda: user_group_view.Page.get( + user_group_view.Page(), request, **kwargs, current_page=1, page_size=20 + ), + "user group edit": lambda: user_group_view.put(request, **kwargs), + } + + def assert_endpoint_access(self, request, workspace_id, resource, allowed): + with ( + patch.object(Log, "save"), + patch("system_manage.views.user_resource_permission.result.success", return_value=True), + patch("system_manage.views.user_resource_permission.ResourceUserPermissionSerializer") as user_serializer, + patch( + "system_manage.views.user_group_resource_permission.ResourceUserGroupPermissionSerializer" + ) as user_group_serializer, + ): + user_serializer.return_value.list.return_value = None + user_serializer.return_value.page.return_value = None + user_serializer.return_value.edit.return_value = None + user_group_serializer.return_value.list.return_value = None + user_group_serializer.return_value.page.return_value = None + user_group_serializer.return_value.edit.return_value = None + + for endpoint, call in self.endpoint_calls(request, workspace_id, resource).items(): + with self.subTest(resource=resource, endpoint=endpoint): + if allowed: + self.assertIs(True, call()) + else: + with self.assertRaises(AppUnauthorizedFailed): + call() + + def test_builtin_admin_can_access_target_authorization_endpoints(self): + request = self.build_request({"ADMIN"}) + + for resource in self.resources: + with self.subTest(resource=resource): + self.assert_endpoint_access(request, self.workspace_id, resource, True) + + def test_workspace_manager_is_limited_to_own_workspace(self): + request = self.build_request( + { + "WORKSPACE_MANAGE", + f"WORKSPACE_MANAGE:w:{self.workspace_id}", + } + ) + + for resource in self.resources: + with self.subTest(resource=resource, workspace="own"): + self.assert_endpoint_access(request, self.workspace_id, resource, True) + with self.subTest(resource=resource, workspace="other"): + self.assert_endpoint_access(request, self.other_workspace_id, resource, False) + + def test_inherited_admin_requires_system_resource_authorization_permission(self): + for resource in self.resources: + roles = {"EXTENDS_ADMIN"} + with self.subTest(resource=resource, permissions="none"): + self.assert_endpoint_access(self.build_request(roles), self.workspace_id, resource, False) + with self.subTest(resource=resource, permissions="system authorization"): + self.assert_endpoint_access( + self.build_request(roles, self.system_authorization(resource)), + self.workspace_id, + resource, + True, + ) + self.assert_endpoint_access( + self.build_request(roles, self.system_authorization(resource)), + self.other_workspace_id, + resource, + True, + ) + + def test_inherited_workspace_manager_is_limited_to_workspace(self): + roles = {f"EXTENDS_WORKSPACE_MANAGE:w:{self.workspace_id}"} + + for resource in self.resources: + with self.subTest(resource=resource, permissions="none"): + self.assert_endpoint_access(self.build_request(roles), self.workspace_id, resource, False) + with self.subTest(resource=resource, permissions="resource authorization"): + permissions = self.detailed_permissions(resource, self.workspace_id) + self.assert_endpoint_access( + self.build_request(roles, permissions), + self.workspace_id, + resource, + True, + ) + self.assert_endpoint_access( + self.build_request(roles, permissions), + self.other_workspace_id, + resource, + False, + ) + + +class ResourcePermissionRealAuthTests(TestCase): + workspace_id = "workspace-1" + other_workspace_id = "workspace-2" + target = "target-1" + resources = ("APPLICATION", "KNOWLEDGE", "TOOL", "MODEL") + system_auth_permissions = { + "APPLICATION": "RESOURCE_APPLICATION_AUTH", + "KNOWLEDGE": "RESOURCE_KNOWLEDGE_AUTH", + "TOOL": "RESOURCE_TOOL_AUTH", + "MODEL": "RESOURCE_MODEL_AUTH", + } + + def setUp(self): + DatabaseModelManage.init() + self.user = User.objects.create( + username="resource-permission-user", + nick_name="Resource Permission User", + password="encrypted-password", + role=RoleConstants.USER.name, + ) + + def tearDown(self): + self.clear_auth_cache() + super().tearDown() + + def clear_auth_cache(self): + for cache_version in (Cache_Version.PERMISSION_LIST, Cache_Version.ROLE_LIST): + cache.delete( + cache_version.get_key(user_id=self.user.id), + version=cache_version.get_version(), + ) + + def assign_role(self, role_type, workspace_id, permission_names): + role = RoleSetting.objects.create( + role_name=f"custom-{role_type}-{workspace_id}", + internal=False, + type=role_type, + ) + UserRoleRelation.objects.create(user=self.user, role=role, workspace_id=workspace_id) + RolePermission.objects.bulk_create( + [ + RolePermission( + role=role, + permission_id=str(PermissionConstants[permission_name].value), + ) + for permission_name in permission_names + ] + ) + self.clear_auth_cache() + return role + + def assign_builtin_role(self, role_id, workspace_id): + UserRoleRelation.objects.create( + user=self.user, + role=RoleSetting.objects.get(id=role_id), + workspace_id=workspace_id, + ) + self.clear_auth_cache() + + def build_request(self): + return SimpleNamespace( + query_params=QueryDict(), + data={}, + user=Principal(self.user.id, UserType.SYSTEM_USER, self.user), + META={}, + path="/", + auth=get_auth(self.user), + ) + + def assert_list_permissions(self, request, workspace_id, resource, allowed_operations): + user_view = user_resource_permission.WorkSpaceUserResourcePermissionView() + user_group_view = user_group_resource_permission.WorkSpaceUserGroupResourcePermissionView() + user_kwargs = { + "workspace_id": workspace_id, + "user_id": str(self.user.id), + "resource": resource, + } + user_group_kwargs = { + "workspace_id": workspace_id, + "user_group_id": "group-1", + "resource": resource, + } + calls = { + ("READ", "user list"): lambda: user_view.get(request, **user_kwargs), + ("READ", "user page"): lambda: user_view.Page.get( + user_view.Page(), request, **user_kwargs, current_page=1, page_size=20 + ), + ("EDIT", "user edit"): lambda: user_view.put(request, **user_kwargs), + ("READ", "user group list"): lambda: user_group_view.get(request, **user_group_kwargs), + ("READ", "user group page"): lambda: user_group_view.Page.get( + user_group_view.Page(), request, **user_group_kwargs, current_page=1, page_size=20 + ), + ("EDIT", "user group edit"): lambda: user_group_view.put(request, **user_group_kwargs), + } + with ( + patch.object(Log, "save"), + patch("system_manage.views.user_resource_permission.result.success", return_value=True), + patch("system_manage.views.user_resource_permission.UserResourcePermissionSerializer") as user_serializer, + patch( + "system_manage.views.user_group_resource_permission.UserGroupResourcePermissionSerializer" + ) as user_group_serializer, + ): + user_serializer.return_value.list.return_value = None + user_serializer.return_value.page.return_value = None + user_serializer.return_value.edit.return_value = None + user_group_serializer.return_value.list.return_value = None + user_group_serializer.return_value.page.return_value = None + user_group_serializer.return_value.edit.return_value = None + + for (operation, endpoint), call in calls.items(): + with self.subTest(resource=resource, endpoint=endpoint, workspace_id=workspace_id): + if operation in allowed_operations: + self.assertIs(True, call()) + else: + with self.assertRaises(AppUnauthorizedFailed): + call() + + def assert_target_permissions(self, request, workspace_id, resource, allowed): + user_view = user_resource_permission.WorkspaceResourceUserPermissionView() + user_group_view = user_group_resource_permission.WorkspaceResourceUserGroupPermissionView() + kwargs = { + "workspace_id": workspace_id, + "target": self.target, + "resource": resource, + } + calls = { + "user list": lambda: user_view.get(request, **kwargs), + "user page": lambda: user_view.Page.get(user_view.Page(), request, **kwargs, current_page=1, page_size=20), + "user edit": lambda: user_view.put(request, **kwargs), + "user group list": lambda: user_group_view.get(request, **kwargs), + "user group page": lambda: user_group_view.Page.get( + user_group_view.Page(), request, **kwargs, current_page=1, page_size=20 + ), + "user group edit": lambda: user_group_view.put(request, **kwargs), + } + with ( + patch.object(Log, "save"), + patch("system_manage.views.user_resource_permission.result.success", return_value=True), + patch("system_manage.views.user_resource_permission.ResourceUserPermissionSerializer") as user_serializer, + patch( + "system_manage.views.user_group_resource_permission.ResourceUserGroupPermissionSerializer" + ) as user_group_serializer, + ): + user_serializer.return_value.list.return_value = None + user_serializer.return_value.page.return_value = None + user_serializer.return_value.edit.return_value = None + user_group_serializer.return_value.list.return_value = None + user_group_serializer.return_value.page.return_value = None + user_group_serializer.return_value.edit.return_value = None + + for endpoint, call in calls.items(): + with self.subTest(resource=resource, endpoint=endpoint, workspace_id=workspace_id): + if allowed: + self.assertIs(True, call()) + else: + with self.assertRaises(AppUnauthorizedFailed): + call() + + def test_builtin_admin_uses_role(self): + self.assign_builtin_role(RoleConstants.ADMIN.name, "None") + request = self.build_request() + for resource in self.resources: + with self.subTest(role="ADMIN", resource=resource): + self.assert_list_permissions(request, self.workspace_id, resource, {"READ", "EDIT"}) + self.assert_target_permissions(request, self.workspace_id, resource, True) + + def test_builtin_workspace_manage_is_limited_to_workspace(self): + self.assign_builtin_role(RoleConstants.WORKSPACE_MANAGE.name, self.workspace_id) + request = self.build_request() + for resource in self.resources: + with self.subTest(role="WORKSPACE_MANAGE", resource=resource, workspace="own"): + self.assert_list_permissions(request, self.workspace_id, resource, {"READ", "EDIT"}) + self.assert_target_permissions(request, self.workspace_id, resource, True) + with self.subTest(role="WORKSPACE_MANAGE", resource=resource, workspace="other"): + self.assert_list_permissions(request, self.other_workspace_id, resource, set()) + self.assert_target_permissions(request, self.other_workspace_id, resource, False) + + def test_inherited_admin_requires_system_resource_permissions(self): + self.assign_role(RoleConstants.ADMIN.name, "None", []) + request = self.build_request() + for resource in self.resources: + with self.subTest(resource=resource, permissions="none"): + self.assert_list_permissions(request, self.workspace_id, resource, set()) + self.assert_target_permissions(request, self.workspace_id, resource, False) + + RolePermission.objects.all().delete() + permissions = ( + [f"{resource}_RESOURCE_PERMISSION_READ" for resource in self.resources] + + [f"{resource}_RESOURCE_PERMISSION_EDIT" for resource in self.resources] + + list(self.system_auth_permissions.values()) + ) + role = RoleSetting.objects.get(role_name=f"custom-{RoleConstants.ADMIN.name}-None") + RolePermission.objects.bulk_create( + [ + RolePermission( + role=role, + permission_id=str(PermissionConstants[permission_name].value), + ) + for permission_name in permissions + ] + ) + self.clear_auth_cache() + request = self.build_request() + for resource in self.resources: + with self.subTest(resource=resource, permissions="detailed"): + self.assert_list_permissions(request, self.workspace_id, resource, {"READ", "EDIT"}) + self.assert_target_permissions(request, self.workspace_id, resource, True) + self.assert_target_permissions(request, self.other_workspace_id, resource, True) + + def test_inherited_workspace_manage_requires_permissions_within_workspace(self): + resources = self.resources + self.assign_role( + RoleConstants.WORKSPACE_MANAGE.name, + self.workspace_id, + [f"{resource}_RESOURCE_PERMISSION_READ" for resource in resources] + + [f"{resource}_RESOURCE_PERMISSION_EDIT" for resource in resources] + + [f"{resource}_RESOURCE_AUTHORIZATION" for resource in resources], + ) + request = self.build_request() + for resource in resources: + with self.subTest(resource=resource, workspace="own"): + self.assert_list_permissions(request, self.workspace_id, resource, {"READ", "EDIT"}) + self.assert_target_permissions(request, self.workspace_id, resource, True) + with self.subTest(resource=resource, workspace="other"): + self.assert_list_permissions(request, self.other_workspace_id, resource, set()) + self.assert_target_permissions(request, self.other_workspace_id, resource, False) + + class ChatUserEditTests(TestCase): def test_status_and_password_changes_preserve_user_groups(self): user = ChatUser.objects.create( diff --git a/apps/system_manage/views/user_group_resource_permission.py b/apps/system_manage/views/user_group_resource_permission.py index 50d2632ab11..d127186301b 100644 --- a/apps/system_manage/views/user_group_resource_permission.py +++ b/apps/system_manage/views/user_group_resource_permission.py @@ -57,9 +57,18 @@ class WorkSpaceUserGroupResourcePermissionView(APIView): tags=[_("Resources authorization")], # type: ignore ) @has_permissions( - lambda r, kwargs: PermissionConstants[kwargs.get("resource") + "_RESOURCE_PERMISSION_READ"], RoleConstants.ADMIN, - RoleConstants.WORKSPACE_MANAGE, + RoleConstants.WORKSPACE_MANAGE.get_workspace_role(), + ViewPermission( + [RoleConstants.EXTENDS_ADMIN], + [lambda r, kwargs: PermissionConstants[kwargs.get("resource") + "_RESOURCE_PERMISSION_READ"]], + compare=CompareConstants.AND, + ), + ViewPermission( + [RoleConstants.EXTENDS_WORKSPACE_MANAGE.get_workspace_role()], + [lambda r, kwargs: PermissionConstants[kwargs.get("resource") + "_RESOURCE_PERMISSION_READ"]], + compare=CompareConstants.AND, + ), ) def get(self, request: Request, workspace_id: str, user_group_id: str, resource: str): return result.success( @@ -86,11 +95,18 @@ def get(self, request: Request, workspace_id: str, user_group_id: str, resource: get_operation_object=lambda r, k: get_user_operation_object(k.get("user_group_id")), ) @has_permissions( - lambda r, kwargs: PermissionConstants[ - kwargs.get("resource") + "_RESOURCE_PERMISSION_EDIT" - ]._build_workspace_permission(), RoleConstants.ADMIN, RoleConstants.WORKSPACE_MANAGE.get_workspace_role(), + ViewPermission( + [RoleConstants.EXTENDS_ADMIN], + [lambda r, kwargs: PermissionConstants[kwargs.get("resource") + "_RESOURCE_PERMISSION_EDIT"]], + compare=CompareConstants.AND, + ), + ViewPermission( + [RoleConstants.EXTENDS_WORKSPACE_MANAGE.get_workspace_role()], + [lambda r, kwargs: PermissionConstants[kwargs.get("resource") + "_RESOURCE_PERMISSION_EDIT"]], + compare=CompareConstants.AND, + ), ) def put(self, request: Request, workspace_id: str, user_group_id: str, resource: str): return result.success( @@ -113,11 +129,18 @@ class Page(APIView): tags=[_("Resources authorization")], # type: ignore ) @has_permissions( - lambda r, kwargs: PermissionConstants[ - kwargs.get("resource") + "_RESOURCE_PERMISSION_READ" - ]._build_workspace_permission(), RoleConstants.ADMIN, RoleConstants.WORKSPACE_MANAGE.get_workspace_role(), + ViewPermission( + [RoleConstants.EXTENDS_ADMIN], + [lambda r, kwargs: PermissionConstants[kwargs.get("resource") + "_RESOURCE_PERMISSION_READ"]], + compare=CompareConstants.AND, + ), + ViewPermission( + [RoleConstants.EXTENDS_WORKSPACE_MANAGE.get_workspace_role()], + [lambda r, kwargs: PermissionConstants[kwargs.get("resource") + "_RESOURCE_PERMISSION_READ"]], + compare=CompareConstants.AND, + ), ) def get( self, @@ -156,6 +179,16 @@ class WorkspaceResourceUserGroupPermissionView(APIView): tags=[_("Resources authorization")], # type: ignore ) @has_permissions( + RoleConstants.ADMIN, + ViewPermission( + [RoleConstants.EXTENDS_ADMIN], + [ + lambda r, kwargs: PermissionConstants[ + "RESOURCE_" + kwargs.get("resource").replace("_FOLDER", "") + "_AUTH" + ] + ], + compare=CompareConstants.AND, + ), lambda r, kwargs: PermissionConstants[ kwargs.get("resource").replace("_FOLDER", "") + "_RESOURCE_AUTHORIZATION" ].get_workspace_permission_workspace_manage_role(), @@ -202,6 +235,16 @@ def get(self, request: Request, workspace_id: str, target: str, resource: str): get_operation_object=lambda r, k: get_user_operation_object(k.get("user_id")), ) @has_permissions( + RoleConstants.ADMIN, + ViewPermission( + [RoleConstants.EXTENDS_ADMIN], + [ + lambda r, kwargs: PermissionConstants[ + "RESOURCE_" + kwargs.get("resource").replace("_FOLDER", "") + "_AUTH" + ] + ], + compare=CompareConstants.AND, + ), lambda r, kwargs: PermissionConstants[ kwargs.get("resource").replace("_FOLDER", "") + "_RESOURCE_AUTHORIZATION" ].get_workspace_permission_workspace_manage_role(), @@ -243,6 +286,16 @@ class Page(APIView): tags=[_("Resources authorization")], # type: ignore ) @has_permissions( + RoleConstants.ADMIN, + ViewPermission( + [RoleConstants.EXTENDS_ADMIN], + [ + lambda r, kwargs: PermissionConstants[ + "RESOURCE_" + kwargs.get("resource").replace("_FOLDER", "") + "_AUTH" + ] + ], + compare=CompareConstants.AND, + ), lambda r, kwargs: PermissionConstants[ kwargs.get("resource").replace("_FOLDER", "") + "_RESOURCE_AUTHORIZATION" ].get_workspace_permission_workspace_manage_role(), diff --git a/apps/system_manage/views/user_resource_permission.py b/apps/system_manage/views/user_resource_permission.py index 68d6cbab04d..a90f2acc82a 100644 --- a/apps/system_manage/views/user_resource_permission.py +++ b/apps/system_manage/views/user_resource_permission.py @@ -17,12 +17,9 @@ from common.auth import TokenAuth from common.auth.authentication import has_permissions from common.auth.constants.compare_constants import CompareConstants -from common.auth.constants.group_constants import Group -from common.auth.constants.operate_constants import Operate from common.auth.constants.permission_constants import PermissionConstants from common.auth.constants.role_constants import RoleConstants from common.auth.struct.aggregate_permission import ViewPermission -from common.auth.struct.permission import Permission from common.log.log import log from system_manage.api.user_resource_permission import ( UserResourcePermissionAPI, @@ -58,11 +55,18 @@ class WorkSpaceUserResourcePermissionView(APIView): tags=[_("Resources authorization")], # type: ignore ) @has_permissions( - lambda r, kwargs: PermissionConstants[ - kwargs.get("resource") + "_RESOURCE_PERMISSION_READ" - ]._build_workspace_permission(), RoleConstants.ADMIN, RoleConstants.WORKSPACE_MANAGE.get_workspace_role(), + ViewPermission( + [RoleConstants.EXTENDS_ADMIN], + [lambda r, kwargs: PermissionConstants[kwargs.get("resource") + "_RESOURCE_PERMISSION_READ"]], + compare=CompareConstants.AND, + ), + ViewPermission( + [RoleConstants.EXTENDS_WORKSPACE_MANAGE.get_workspace_role()], + [lambda r, kwargs: PermissionConstants[kwargs.get("resource") + "_RESOURCE_PERMISSION_READ"]], + compare=CompareConstants.AND, + ), ) def get(self, request: Request, workspace_id: str, user_id: str, resource: str): return result.success( @@ -89,11 +93,18 @@ def get(self, request: Request, workspace_id: str, user_id: str, resource: str): get_operation_object=lambda r, k: get_user_operation_object(k.get("user_id")), ) @has_permissions( - lambda r, kwargs: PermissionConstants[ - kwargs.get("resource") + "_RESOURCE_PERMISSION_EDIT" - ]._build_workspace_permission(), RoleConstants.ADMIN, RoleConstants.WORKSPACE_MANAGE.get_workspace_role(), + ViewPermission( + [RoleConstants.EXTENDS_ADMIN], + [lambda r, kwargs: PermissionConstants[kwargs.get("resource") + "_RESOURCE_PERMISSION_EDIT"]], + compare=CompareConstants.AND, + ), + ViewPermission( + [RoleConstants.EXTENDS_WORKSPACE_MANAGE.get_workspace_role()], + [lambda r, kwargs: PermissionConstants[kwargs.get("resource") + "_RESOURCE_PERMISSION_EDIT"]], + compare=CompareConstants.AND, + ), ) def put(self, request: Request, workspace_id: str, user_id: str, resource: str): return result.success( @@ -116,11 +127,18 @@ class Page(APIView): tags=[_("Resources authorization")], # type: ignore ) @has_permissions( - lambda r, kwargs: PermissionConstants[ - kwargs.get("resource") + "_RESOURCE_PERMISSION_READ" - ]._build_workspace_permission(), RoleConstants.ADMIN, RoleConstants.WORKSPACE_MANAGE.get_workspace_role(), + ViewPermission( + [RoleConstants.EXTENDS_ADMIN], + [lambda r, kwargs: PermissionConstants[kwargs.get("resource") + "_RESOURCE_PERMISSION_READ"]], + compare=CompareConstants.AND, + ), + ViewPermission( + [RoleConstants.EXTENDS_WORKSPACE_MANAGE.get_workspace_role()], + [lambda r, kwargs: PermissionConstants[kwargs.get("resource") + "_RESOURCE_PERMISSION_READ"]], + compare=CompareConstants.AND, + ), ) def get( self, request: Request, workspace_id: str, user_id: str, resource: str, current_page: str, page_size: str @@ -153,6 +171,16 @@ class WorkspaceResourceUserPermissionView(APIView): tags=[_("Resources authorization")], # type: ignore ) @has_permissions( + RoleConstants.ADMIN, + ViewPermission( + [RoleConstants.EXTENDS_ADMIN], + [ + lambda r, kwargs: PermissionConstants[ + "RESOURCE_" + kwargs.get("resource").replace("_FOLDER", "") + "_AUTH" + ] + ], + compare=CompareConstants.AND, + ), lambda r, kwargs: PermissionConstants[ kwargs.get("resource").replace("_FOLDER", "") + "_RESOURCE_AUTHORIZATION" ].get_workspace_permission_workspace_manage_role(), @@ -204,6 +232,16 @@ def get(self, request: Request, workspace_id: str, target: str, resource: str): get_operation_object=lambda r, k: get_user_operation_object(k.get("user_id")), ) @has_permissions( + RoleConstants.ADMIN, + ViewPermission( + [RoleConstants.EXTENDS_ADMIN], + [ + lambda r, kwargs: PermissionConstants[ + "RESOURCE_" + kwargs.get("resource").replace("_FOLDER", "") + "_AUTH" + ] + ], + compare=CompareConstants.AND, + ), lambda r, kwargs: PermissionConstants[ kwargs.get("resource").replace("_FOLDER", "") + "_RESOURCE_AUTHORIZATION" ].get_workspace_permission_workspace_manage_role(), @@ -245,6 +283,16 @@ class Page(APIView): tags=[_("Resources authorization")], # type: ignore ) @has_permissions( + RoleConstants.ADMIN, + ViewPermission( + [RoleConstants.EXTENDS_ADMIN], + [ + lambda r, kwargs: PermissionConstants[ + "RESOURCE_" + kwargs.get("resource").replace("_FOLDER", "") + "_AUTH" + ] + ], + compare=CompareConstants.AND, + ), lambda r, kwargs: PermissionConstants[ kwargs.get("resource").replace("_FOLDER", "") + "_RESOURCE_AUTHORIZATION" ].get_workspace_permission_workspace_manage_role(),