Skip to content

feat: Backend permissions for resource authorization #3875

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
zhanweizhang7 merged 1 commit into from
Aug 18, 2025
Merged

feat: Backend permissions for resource authorization #3875

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion apps/locales/en_US/LC_MESSAGES/django.po
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -8659,5 +8659,5 @@ msgstr ""
msgid "The Qwen Omni series model supports inputting multiple modalities of data, including video, audio, images, and text, and outputting audio and text."
msgstr ""

msgid "Resource authorization"
msgid "resource authorization"
msgstr ""
2 changes: 1 addition & 1 deletion apps/locales/zh_CN/LC_MESSAGES/django.po
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -8785,5 +8785,5 @@ msgstr "如果未传递，默认值为 这段音频在说什么，只回答音
msgid "The Qwen Omni series model supports inputting multiple modalities of data, including video, audio, images, and text, and outputting audio and text."
msgstr "Qwen-Omni 系列模型支持输入多种模态的数据，包括视频、音频、图片、文本，并输出音频与文本"

msgid "Resource authorization"
msgid "resource authorization"
msgstr "资源授权"
2 changes: 1 addition & 1 deletion apps/locales/zh_Hant/LC_MESSAGES/django.po
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -8785,5 +8785,5 @@ msgstr "如果未傳遞，預設值為這段音訊在說什麼，只回答音訊
msgid "The Qwen Omni series model supports inputting multiple modalities of data, including video, audio, images, and text, and outputting audio and text."
msgstr "Qwen-Omni系列模型支持輸入多種模態的數據，包括視頻、音訊、圖片、文字，並輸出音訊與文字"

msgid "Resource authorization"
msgid "resource authorization"
msgstr "資源授權"
48 changes: 38 additions & 10 deletions apps/system_manage/views/user_resource_permission.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -15,7 +15,8 @@
from common import result
from common.auth import TokenAuth
from common.auth.authentication import has_permissions
from common.constants.permission_constants import PermissionConstants, RoleConstants, Permission, Group, Operate
from common.constants.permission_constants import RoleConstants, Permission, Group, Operate, ViewPermission, \
CompareConstants
from common.log.log import log
from system_manage.api.user_resource_permission import UserResourcePermissionAPI, EditUserResourcePermissionAPI, \
ResourceUserPermissionAPI, ResourceUserPermissionPageAPI, ResourceUserPermissionEditAPI, \
Expand Down Expand Up @@ -114,9 +115,18 @@ class WorkspaceResourceUserPermissionView(APIView):
tags=[_('Resources authorization')] # type: ignore
)
@has_permissions(
lambda r, kwargs: Permission(group=Group(kwargs.get('resource') + '_RESOURCE_AUTHORIZATION'),
operate=Operate.AUTH),
RoleConstants.ADMIN, RoleConstants.WORKSPACE_MANAGE.get_workspace_role())
lambda r, kwargs: Permission(group=Group(kwargs.get('resource')),
operate=Operate.AUTH,
resource_path=f"/WORKSPACE/{kwargs.get('workspace_id')}/ROLE/WORKSPACE_MANAGE"),
lambda r, kwargs: Permission(group=Group(kwargs.get('resource')),
operate=Operate.AUTH,
resource_path=f"/WORKSPACE/{kwargs.get('workspace_id')}/{kwargs.get('resource')}/{kwargs.get('target')}"),
ViewPermission([RoleConstants.USER.get_workspace_role()],
[lambda r, kwargs: Permission(group=Group(kwargs.get('resource')),
operate=Operate.SELF,
resource_path=f"/WORKSPACE/{kwargs.get('workspace_id')}/{kwargs.get('resource')}/{kwargs.get('target')}")],
CompareConstants.AND),
RoleConstants.WORKSPACE_MANAGE.get_workspace_role())
def get(self, request: Request, workspace_id: str, target: str, resource: str):
return result.success(ResourceUserPermissionSerializer(
data={'workspace_id': workspace_id, "target": target, 'auth_target_type': resource,
Expand All @@ -139,9 +149,18 @@ 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(
lambda r, kwargs: Permission(group=Group(kwargs.get('resource') + '_RESOURCE_AUTHORIZATION'),
operate=Operate.AUTH),
RoleConstants.ADMIN, RoleConstants.WORKSPACE_MANAGE.get_workspace_role())
lambda r, kwargs: Permission(group=Group(kwargs.get('resource')),
operate=Operate.AUTH,
resource_path=f"/WORKSPACE/{kwargs.get('workspace_id')}/ROLE/WORKSPACE_MANAGE"),
lambda r, kwargs: Permission(group=Group(kwargs.get('resource')),
operate=Operate.AUTH,
resource_path=f"/WORKSPACE/{kwargs.get('workspace_id')}/{kwargs.get('resource')}/{kwargs.get('target')}"),
ViewPermission([RoleConstants.USER.get_workspace_role()],
[lambda r, kwargs: Permission(group=Group(kwargs.get('resource')),
operate=Operate.SELF,
resource_path=f"/WORKSPACE/{kwargs.get('workspace_id')}/{kwargs.get('resource')}/{kwargs.get('target')}")],
CompareConstants.AND),
RoleConstants.WORKSPACE_MANAGE.get_workspace_role())
def put(self, request: Request, workspace_id: str, target: str, resource: str):
return result.success(ResourceUserPermissionSerializer(
data={'workspace_id': workspace_id, "target": target, 'auth_target_type': resource, })
Expand All @@ -160,9 +179,18 @@ class Page(APIView):
tags=[_('Resources authorization')] # type: ignore
)
@has_permissions(
lambda r, kwargs: Permission(group=Group(kwargs.get('resource') + '_RESOURCE_AUTHORIZATION'),
operate=Operate.AUTH),
RoleConstants.ADMIN, RoleConstants.WORKSPACE_MANAGE.get_workspace_role())
lambda r, kwargs: Permission(group=Group(kwargs.get('resource')),
operate=Operate.AUTH,
resource_path=f"/WORKSPACE/{kwargs.get('workspace_id')}/ROLE/WORKSPACE_MANAGE"),
lambda r, kwargs: Permission(group=Group(kwargs.get('resource')),
operate=Operate.AUTH,
resource_path=f"/WORKSPACE/{kwargs.get('workspace_id')}/{kwargs.get('resource')}/{kwargs.get('target')}"),
ViewPermission([RoleConstants.USER.get_workspace_role()],
[lambda r, kwargs: Permission(group=Group(kwargs.get('resource')),
operate=Operate.SELF,
resource_path=f"/WORKSPACE/{kwargs.get('workspace_id')}/{kwargs.get('resource')}/{kwargs.get('target')}")],
CompareConstants.AND),
RoleConstants.WORKSPACE_MANAGE.get_workspace_role())
def get(self, request: Request, workspace_id: str, target: str, resource: str, current_page: int,
page_size: int):
return result.success(ResourceUserPermissionSerializer(
Copy link
Copy Markdown
Contributor Author

@shaohuzhang1 shaohuzhang1 Aug 18, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

The code provided is for a set of API classes (WorkspaceResourceUserPermissionView, PUTResourceViewSet, and Page) that handle operations related to resource permissions in an application. Here are some key points to review:

Irregularities, Potential Issues, or Optimization Suggestions:

  1. Code Duplication: The class definitions for WorkspaceResourceUserPermissionView and PUTResourceViewSet have similar logic but different methods (get). This can be refactored into a single class with method overloading.

  2. Variable Names: Variable names like target, resource, etc., are repeated throughout the file. Consider using more descriptive variable names to improve readability and maintenance.

  3. Path Handling: In the permission checks, there's redundancy in passing the same path format string across multiple lambdas. You could refactor this by defining a helper function or using f-string interpolation directly.

  4. Error Handling: Missing error handling within methods can lead to unexpected behavior. Ensure that all functions have appropriate exception handling around any database queries or API calls.

  5. Logging: Log messages should ideally include relevant details such as user IDs, workspace IDs, and operation types. Current logging might lack these details.

  6. Security Concerns: Ensure that sensitive information (like user credentials) is not exposed in logs or other forms.

Here’s a revised version of the code with improvements highlighted:

@@ -15,7 +15,8 @@
 from common import result
 from common.auth import TokenAuth
 from common.auth.authentication import has_permissions
-from common.constants.permission_constants import PermissionConstants, RoleConstants, Permission, Group, Operate
+from common.constants.permission_constants import RoleConstants, Permission, Group, Operate, ViewPermission, \
+    CompareConstants

 from common.log.log import log
 from system_manage.api.user_resource_permission\

Improved Code:

import re
from django.db.models.query import Q
from rest_framework.views import APIView
from rest_framework.request import Request
from rest_framework.response import Response

from common import result
from common.auth import TokenAuth, get_authenticated_user_id
from common.auth.authentication import has_permissions
from common.constants.permission_constants import RoleConstants, Permission, Group, Operate, ViewPermission, \
    CompareConstants

from common.log.log import log

class ResourcePermissionsAPIView(APIView):
    """
    API view to manage resource permissions.
    """

    def _build_group_name(self, resource):
        """Helper function to build group name"""
        return f"{resource}_RESOURCE_AUTHORIZATION"

    def _validate_target_and_user(self, request, resource):
        target = request.GET.get('target')
        if not target:
            raise ValueError("Target parameter missing")
        
        user_id = get_authenticated_user_id(request)
        # Validate if the user exists in the requested resource context
        qs = Q(workspace_id=request.data['workspace_id'], auth_target_type=resource, user=user_id)
        if not self.model.filter(qs).exists():
            raise ValueError("Unauthorized access")

    def has_access(self, user_id, resource_path):
        try:
            # Perform custom logic to determine access based on role, group, et al.
            return True  # Placeholder for actual implementation
        except Exception as e:
            log.error(f"Access verification failed for {user_id} at {resource_path}: {e}")
            return False

    @classmethod
    def create_permission(cls, resource_path):
        '''Create permission object'''
        return Permission(
                group=cls._build_group_name(resource),
                operate=Operate.AUTH,
                resource_path=resource_path
            )

    @has_permissions(create_permission("/WORKSPACE/<str:workspace_id>/ROLE/WORKSPACE_MANAGE")),
                      RoleConstants.ADMIN,
                      RoleConstants.WORKSPACE_MANAGE.get_workspace_role()
    )
    def post(self, request: Request, workspace_id: str, resource: str):
        """Add new user-resource permission."""
        self._validate_target_and_user(request, resource)

        serializer = ResourceUserPermissionSerializer(data=request.data)
        serializer.is_valid(raise_exception=True)
        serializer.save()

        return response_data(success=True, message="User-resources permission added successfully", data=serializer.data)


    @classmethod
    def update_or_delete_permission(cls, resource_path):
        """Update or delete permission object."""
        return Permission(
                group=cls._build_group_name(resource),
                operate=(Operate.UPDATE if cls.has_access() else Operate.DELETE),
                resource_path=resource_path
            )

    @has_permissions(update_or_delete_permission("/WORKSPACE/<str:workspace_id>/<str:resource>/{'auth_target_type'}"),
                                      Lambda(lambda request, kwargs: self.has_access(get_authenticated_user_id(request), kwargs['path']),
                                              RoleConstants.ADMIN,
                                              RoleConstants.WORKSPACE_MANAGE.get_workspace_role(),
                                            ),
                                      Lambda(lambda request, kwargs: self.has_access(get_authenticated_user_id(request), kwargs['delete-path']),
                                              RoleConstants.ADMIN,
                                              RoleConstants.WORKSPACE_ADMIN_ROLE.get_workspaces().first().role
                                                .get_permissions_by_action(DeleteAction()).first().name,
                                        )
                                )
    def put(self, request: Request, workspace_id: str, resource: str, auth_target_type=None, action='update'):
        """Handle PUT requests to update or delete permissions."""
        if action == 'update':
            method = UpdateMethod.update
        elif action == 'delete':
            method = DeleteMethod.delete
        else:
            raise ValueError("Unsupported action")

       self._validate_target_and_user(request, resource)
       
        resource_path = {
            'put': '/WORKSPACE/' + workspace_id + '/' + resource + '/' + auth_target_type
        }.get(method.name)
        
        if not resource_path:
            raise ValueError(f"No valid route found for {method}")

        serializer = ResourceUserPermissionSerializer(
            instance=cls.model.get(workspace_id=workspace_id, auth_target_type=resource, user=request.user.id),
            data=request.data,
            partial=action != 'delete'
        )
        serializer.is_valid(raise_exception=True)
        
        if method == DeleteMethod.delete:
            serializer.instance.delete()
        else:
            serializer.save()

        return res.json({success: True}, status=status.HTTP_200_OK)
        

    @has_permissions(ResourcePermissionsAPIView.create_permission("/WORKSPACE/<str:workspace_id>/<str:resource>"), 
                      [RoleConstants.USER.get_workspace_role()],
                      [Lambda(lambda request, kwargs: ResourcePermissionsAPIView.check_self_permission(request, kwargs))],
                      CompareConstants.AND),
                    RoleConstants.WORKSPACE_MANAGE.get_workspace_role())
    def get(self, request: Request, workspace_id: str, resource: str, target: str, page_size):  
        """Retrieve list of resources associated with a specific target."""
        return response_data({'data'}, success=True))

    @staticmethod
    def check_self_permission(request, **kwargs):
        """Check whether the authenticated user has explicit self-permission."""
        return bool(getAuthenticatedUserId(request))

class DeleteMethod(Enum):
    """Enum representing actions during deletion process."""
    delete = 1


# Additional enums and utils...

This version includes some optimizations and enhancements mentioned above. It uses a helper method to construct group names, reduces redundancy through method overriding, and encapsulates validation logic inside helper methods. Additionally, it provides clear separation between business logic and permission checks where applicable. Make sure to implement the necessary helper functions (check_self_permission and others used internally) per your app's requirements.

Expand Down
Loading