Skip to content

fix: enhance permission checks by adding user group resource permissions #7857

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
wxg0103 merged 2 commits into from
Nov 12, 2026
Merged

fix: enhance permission checks by adding user group resource permissions #7857

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
e97e34a
fix: enhance validation logic to include workspace_id in model checks
wxg0103 Nov 12, 2026
1b39ef8
fix: enhance permission checks by adding user group resource permissions
wxg0103 Nov 12, 2026
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
12 changes: 5 additions & 7 deletions apps/models_provider/serializers/model_serializer.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -316,9 +316,11 @@ class Create(serializers.Serializer):
credential = serializers.DictField(required=True, label=_("certification information"))
workspace_id = serializers.CharField(required=False, label=_("workspace id"), max_length=128)

def is_valid(self, *, raise_exception=False):
def is_valid(self, *, raise_exception=False, workspace_id=None):
if not super().is_valid(raise_exception=raise_exception):
return False
if workspace_id is None:
workspace_id = self.data.get("workspace_id", "None")
provider = self.data.get("provider")
try:
provider_handler = ModelProvideConstants[provider]
Expand All @@ -342,11 +344,7 @@ def is_valid(self, *, raise_exception=False):
400,
_("The field and default_value fields are required for the {index}th item").format(index=index),
)
if (
QuerySet(Model)
.filter(name=self.data.get("name"), workspace_id=self.data.get("workspace_id", "None"))
.exists()
):
if QuerySet(Model).filter(name=self.data.get("name"), workspace_id=workspace_id).exists():
raise AppApiException(
500, _("base model【{model_name}】already exists").format(model_name=self.data.get("name"))
)
Expand All @@ -364,7 +362,7 @@ def insert(self, workspace_id, with_valid=True):
status = Status.SUCCESS
if with_valid:
try:
self.is_valid(raise_exception=True)
self.is_valid(raise_exception=True, workspace_id=workspace_id)
except AppApiException as e:
if e.code == ValidCode.model_not_fount:
status = Status.DOWNLOAD

@shaohuzhang1 shaohuzhang1 Nov 12, 2026

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

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

Here is an analysis of the code changes provided, focusing on irregularities, potential issues, and optimization suggestions.

1. Irregularities & Potential Issues

A. Logic Inconsistency with Default workspace_id

  • Issue: There is a dangerous inconsistency in how workspace_id handles defaults.
    • In is_valid: workspace_id = self.data.get("workspace_id", "None"). It retrieves the value from the request data. If missing, it sets it to the string "None".
    • In insert: The user passes a workspace_id argument. This is passed to is_valid.
    • In QuerySet.filter: It uses this workspace_id.
  • Risk: If self.data contains workspace_id but the caller passes a different workspace_id (or vice versa), or if self.data lacks it while the caller provides it, the logic might behave unexpectedly depending on which path is taken. Specifically, overriding the method signature to accept workspace_id but then ignoring it if self.data has one (via .get(...)) creates ambiguous precedence.
  • Best Practice: Ideally, is_valid should validate strictly against input data. Business logic dependencies like workspace_id scoping should ideally be handled either entirely via serializer context or explicitly passed down without mixing sources (self.data vs. function arguments) for the same concept unless carefully controlled.

B. String "None" vs. Python None

  • Issue: The default value for missing workspace_id is set to the string "None". Later, workspace_id=workspace_id is used in the DB filter. If a valid workspace ID ever legitimately equals the string "None", this will cause collision bugs.
  • Suggestion: Use None (Python NULL) for unprovided values if the database column allows NULLs, or ensure strictly that "None" is reserved and never used as a real ID. If using strings, consider explicit sentinel values (e.g., "default") rather than capitalized "None".

C. Error Handling Scope

  • Issue: The try...except block only catches AppApiException with specific code ValidCode.model_not_fount. If provider_handler setup raises another exception or if other validation errors raise unexpected exceptions, they may propagate incorrectly or result in status not being updated appropriately.

2. Optimization Suggestions

A. Avoid QuerySet.filter().exists() for Simple Checks (Minor Optimization)

While .exists() is generally efficient, you are filtering by name + workspace_id. Ensure there is a proper composite index on (name, workspace_id) in the Model to prevent full table scans, especially if this check happens during frequent inserts.

B. Extract Validation Logic

The is_valid method is growing large. Consider extracting individual checks into separate methods for readability and testability:

  • validate_provider()
  • validate_fields_structure()
  • check_unique_constraint()

C. Context Instead of Parameter Overriding

In Django REST Framework, passing extra parameters to is_valid like workspace_id=None is uncommon and breaks standard serializer interfaces. Consider one of these cleaner approaches:

  1. Use Serializer Context: Set context['workspace_id'] when creating the serializer instance, then access it in is_valid via self.context.get('workspace_id').
  2. Separate Service Layer Logic: Keep is_valid purely for input format validation. Move the database uniqueness check and business rule validation to the insert method or a dedicated service class, where workspace_id is naturally available.

Example using Context:

# When instantiating serializer
serializer = Create(data=request_data, context={'workspace_id': ws_id})

# Inside Create class
def is_valid(self, *, raise_exception=False):
    if not super().is_valid(raise_exception=raise_exception):
        return False
    
    # Get from context instead of adding param to is_valid
    workspace_id = self.context.get('workspace_id', 'None')
    
    # ... rest of validation using workspace_id

D. Naming Convention

Avoid modifying external library method signatures (is_valid). It violates the Liskov Substitution Principle and makes debugging harder. If you must override, stick to the original signature.

Summary of Recommended Actions:

  1. Refactor to use self.context for passing workspace_id into the serializer instead of altering the is_valid signature.
  2. Ensure DB Indexes exist for (workspace_id, name) to optimize the duplicate check.
  3. Clarify Default Value Policy: Decide firmly between using None (Python null) or a reserved string like 'global'/'default' for unset workspaces, and document it.

Expand Down
19 changes: 16 additions & 3 deletions apps/oss/serializers/file.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -40,7 +40,7 @@
from knowledge.services import validate_knowledge_file_size
from maxkb.const import CONFIG
from rest_framework import serializers
from system_manage.models import ChatUserApiKey, WorkspaceUserResourcePermission
from system_manage.models import ChatUserApiKey, WorkspaceUserGroupResourcePermission, WorkspaceUserResourcePermission
from system_manage.models.resource_mapping import ResourceMapping, ResourceType
from tools.serializers.tool import UploadedFileField
from users.models import User
Expand Down Expand Up @@ -417,7 +417,7 @@ def _check_workspace_resource_permission(
else ["VIEW", "MANAGE"]
)
if (
not QuerySet(WorkspaceUserResourcePermission)
QuerySet(WorkspaceUserResourcePermission)
.filter(
target=target_id,
workspace_id=workspace_id,
Expand All @@ -427,7 +427,20 @@ def _check_workspace_resource_permission(
)
.exists()
):
_deny()
return
if (
QuerySet(WorkspaceUserGroupResourcePermission)
.filter(
target=target_id,
workspace_id=workspace_id,
user_group__user_relations__user_id=user_id,
auth_target_type=auth_target_type,
permission_list__overlap=permission_list,
)
.exists()
):
return
_deny()


class FileSerializer(serializers.Serializer):

@shaohuzhang1 shaohuzhang1 Nov 12, 2026

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

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

The provided code change introduces significant logic errors and performance issues. Below is a detailed analysis:

1. Critical Logic Error: Inverted Permission Check

Location: _check_workspace_resource_permission function, specifically the first if condition.

  • Before:

    if not QuerySet(...).filter(...).exists():
    _deny()

    Logic: If the permission does not exist, deny access. This implies existence = Allow.

  • After (Patch):

    if (
    QuerySet(WorkspaceUserResourcePermission) # Note missing "not"
    .filter(...)
    .exists()
):
    return # Early return if permission EXISTS

    Logic: If the permission exists, return (allow access). This part seems logically fine if it was intended to allow access on direct permission. However, look at the subsequent code path:

  • The Real Problem: The original code likely had a structure like:

    if not has_direct_permission:
     if not has_group_permission:
         _deny()
     else:
         pass # Allow via group
     # Or simply:
     # if not (has_direct OR has_group): _deny()

    The new patch changes the logic to:

    1. If Direct Permission Exists -> Return (Allow).
    2. If Group Permission Exists -> Return (Allow).
    3. Else -> Deny.

    While this seems logically correct for an "OR" permission model (Direct OR Group), there is a subtle bug in the original diff context which isn't fully visible, but more importantly, the first if statement lost the not keyword.

    Wait, let's re-read carefully.

    Original:

    if (not QuerySet(...).filter(...).exists()):
    _deny()

    This means: If User does NOT have direct permission -> Deny immediately. It ignores group permissions entirely in the original snippet? Or perhaps the original code continued after the if block with a fallback? Given the indentation, the _deny() is inside the if.

    New Code:

    if (QuerySet(...).exists()):
    return
if (QuerySet(...Group...).exists()):
    return
_deny()

    This new logic is actually correct for implementing "Allow if Direct OR Group". The removal of not was necessary because the flow changed from "Fail fast if missing" to "Pass fast if present".

    HOWEVER, there is a major risk: Did the original code have other logic after the first if?
    If the original code was:

    if not has_direct:
    _deny() # This denied everyone without direct perms
# ... unreachable code for group check?

    Then the patch fixes a bug where group permissions were ignored.

    But if the original code was:

    if not has_direct:
    if not has_group:
        _deny()
    else:
        pass # allowed
else:
    pass # allowed

    Then the patch is a refactor.

    Verdict: The logic appears correct now (Allow if Direct or Group, else Deny). Ensure that _deny() raises an exception or otherwise halts execution properly.

2. Performance Issue: N+1 Query / Double Database Hit

Location: The two separate QuerySet.exists() calls.

  • Problem: The code executes two separate database queries:
    1. Check WorkspaceUserResourcePermission.
    2. If that fails, check WorkspaceUserGroupResourcePermission.
  • Optimization: Combine them into a single query if possible, or ensure indexing is perfect. However, since they are different models, combining is hard.
  • Better Approach: Since exists() is relatively cheap, this might be acceptable. BUT, if the first query returns True, we skip the second. This is good short-circuiting.
  • Recommendation: This is generally acceptable. However, ensure user_group__user_relations__user_id filtering is indexed correctly, as joining tables can be expensive.

3. Potential Bug: permission_list__overlap Field Name

Location: WorkspaceUserGroupResourcePermission.filter(... permission_list__overlap=permission_list ...)

  • Issue: permission_list__overlap is specific to PostgreSQL arrays.
  • Risk: If you are using MySQL, SQLite, or any DB without array support, this will crash or behave unexpectedly.
  • Suggestion: Verify that your database backend supports overlap lookup. If not, you need to handle intersection/coverage differently (e.g., fetch groups and check in Python, though that's less efficient for large datasets).

4. Code Style & Readability

  • Repeated Filter Arguments: The filter arguments are duplicated between the two checks. 
    target=target_id,
workspace_id=workspace_id,
auth_target_type=auth_target_type,
  • Refactoring Suggestion: Extract these common filters into a dictionary or a helper function to reduce duplication.

5. Security Consideration

  • User Context: Ensure that user_id passed here is trusted and corresponds to the currently authenticated user in the request context. Do not allow client-supplied user IDs to bypass permission checks.

✅ Recommended Refactored Code

def _check_workspace_resource_permission(
    self, target_id: str, workspace_id: str, user_id: str, auth_target_type: str, permission_list: List[str]
) -> None:
    """
    Check if the user has permission either directly or via their user groups.
    """
    # Base filters common to both direct and group permissions
    common_filters = {
        "target": target_id,
        "workspace_id": workspace_id,
        "auth_target_type": auth_target_type,
    }

    # 1. Check Direct User Permission
    # Note: Assumes 'permission' field is used here. Adjust field name if different.
    if WorkspaceUserResourcePermission.objects.filter(
        user_id=user_id,
        permission__in=permission_list, # Adjust 'permission' field name as per your model
        **common_filters
    ).exists():
        return

    # 2. Check Group-based Permission
    if WorkspaceUserGroupResourcePermission.objects.filter(
        user_group__user_relations__user_id=user_id,
        permission_list__overlap=permission_list,  # Ensure DB supports this
        **common_filters
    ).exists():
        return

    # 3. Deny Access
    raise PermissionDenied("Insufficient permissions") # Or however _deny() is implemented

Summary of Issues to Address:

  1. Verify _deny() implementation: Ensure it stops execution (raises Http404 or PermissionDenied).
  2. Database Compatibility: Confirm __overlap works with your DB backend.
  3. Indexing: Ensure indexes exist on (target, workspace_id, user_id) and similar composite keys to keep exists() fast.
  4. Field Names: Double-check that permission vs permission_list field names match your models exactly.

Expand Down
17 changes: 16 additions & 1 deletion apps/oss/tests/test_file_permissions.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -9,7 +9,7 @@
from common.exception.app_exception import AppUnauthorizedFailed
from knowledge.models import Document, File, FileSourceType, Knowledge, Paragraph
from oss.serializers.file import FileSerializer, _auth_system, _check_workspace_resource_permission
from system_manage.models import WorkspaceUserResourcePermission
from system_manage.models import WorkspaceUserGroupResourcePermission, WorkspaceUserResourcePermission
from users.models import User


Expand All @@ -19,6 +19,9 @@ def setUp(self):
self.grants = MagicMock()
self.grants.filter.return_value = self.grants
self.grants.exists.return_value = False
self.group_grants = MagicMock()
self.group_grants.filter.return_value = self.group_grants
self.group_grants.exists.return_value = False
self.user_query = MagicMock()
self.user_query.filter.return_value.first.return_value = SimpleNamespace(id="user")
self.document_query = MagicMock()
Expand All @@ -32,6 +35,7 @@ def setUp(self):
Paragraph: self.document_query,
Knowledge: self.knowledge_query,
WorkspaceUserResourcePermission: self.grants,
WorkspaceUserGroupResourcePermission: self.group_grants,
}[model]
self.enterContext(patch("oss.serializers.file.get_auth", return_value=self.auth))
self.file = File(source_type=FileSourceType.DOCUMENT, source_id="document")
Expand Down Expand Up @@ -73,6 +77,17 @@ def test_explicit_view_and_manage_grants_allow_source_file_reads(self):
permission_list__overlap=["VIEW", "MANAGE"],
)

def test_user_group_view_and_manage_grants_allow_source_file_reads(self):
self.group_grants.exists.return_value = True
_auth_system(self.file, "user")
self.group_grants.filter.assert_called_once_with(
target="knowledge",
workspace_id="ws",
user_group__user_relations__user_id="user",
auth_target_type="KNOWLEDGE",
permission_list__overlap=["VIEW", "MANAGE"],
)

def test_custom_workspace_manager_uses_the_permission_object(self):
self.auth.roles = {"EXTENDS_WORKSPACE_MANAGE:w:ws"}
self.auth.permissions = {"KNOWLEDGE_WORKSPACE_MANAGE:w:ws": PermissionConstants.KNOWLEDGE_READ.value.bit()}

@shaohuzhang1 shaohuzhang1 Nov 12, 2026

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

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

Here is a review of the provided code diff, focusing on test logic, mocking strategy, and potential issues.

1. Mocking Strategy & Isolation (Potential Issue)

The most significant concern is how WorkspaceUserGroupResourcePermission is mocked.

  • Current Approach: The patch replaces the model class with a MagicMock (self.group_grants). This means no database query is actually executed.
  • Risk: The code in _auth_system likely constructs a queryset like WorkspaceUserGroupResourcePermission.objects.filter(...).
    • If the implementation uses direct field lookups (e.g., .filter(user_group__user_relations__user_id=user)), this might work if the mock intercepts .objects or if the model is replaced entirely at the module level where it's imported.
    • However, if the code relies on custom managers, related descriptor access, or database-level features (like JSONB overlaps, which are hinted at by permission_list__overlap), mocking the class rather than the objects manager can be fragile.
    • Critique: By patching the model itself (WorkspaceUserGroupResourcePermission: self.group_grants), you are assuming that _auth_system interacts with this model only via a path that hits this specific reference. If _auth_system imports the model directly and uses Model.objects, but your patch replaces Model, does Model.objects return self.group_grants?
      • In Django, if you patch module.Model, then module.Model.objects returns a manager tied to the mock. This can work, but it requires careful verification that the mock's .objects attribute (or default mock behavior) aligns with .filter() calls.
      • Better Practice: Usually, we patch the manager or the specific method used, or ensure the mock is structured to mimic the ORM behavior more closely. Patching the model class itself is often considered "low-level" mocking that can break if the internal structure of the model changes.

2. Test Assertion Specificity

In test_user_group_view_and_manage_grants_allow_source_file_reads:

self.group_grants.filter.assert_called_once_with(
    target="knowledge",
    workspace_id="ws",
    user_group__user_relations__user_id="user",
    auth_target_type="KNOWLEDGE",
    permission_list__overlap=["VIEW", "MANAGE"],
)
  • Pros: Very precise assertion of the query parameters. This ensures the complex join path (user_group__user_relations__user_id) is exactly as expected.
  • Cons: High coupling to implementation details. If the underlying query optimization changes (e.g., switching from a JOIN-based lookup to a subquery, or pre-fetching user group IDs for performance), this test will fail even if the functional behavior is identical.
  • Suggestion: Consider testing the outcome (access granted/denied) first, and only assert the specific filter arguments if the security-critical nature of the query requires exact enforcement against SQL injection or logical flaws. Alternatively, extract the filter dictionary into a variable to make it reusable and less brittle.

3. Missing Negative Test Case

You added a positive test:

  • test_user_group_view_and_manage_grants_allow_source_file_reads (Access Allowed)

Recommendation: Add a negative test case:

  • test_user_group_view_and_manage_grants_denies_file_reads_when_no_permission_exists
    • Set self.group_grants.exists.return_value = False.
    • Verify that an AppUnauthorizedFailed exception (or similar) is raised, assuming other permissions (individual user grants) also fail.
    • Currently, setUp sets self.grants.exists.return_value = False, so if group_grants also returns False, the total permission check should fail. Ensure this failure propagates correctly.

4. Code Clarity & Structure

  • Variable Naming: self.grants vs self.group_grants. Clear enough, but consider self.user_resource_perms_mock and self.user_group_perms_mock for slightly higher readability, especially since these mocks represent different permission scopes.
  • Patch Target: Ensure that WorkspaceUserGroupResourcePermission is imported in oss/serializers/file.py (where _auth_system likely resides) in a way that allows the patch to take effect.
    • If _auth_system does from system_manage.models import WorkspaceUserGroupResourcePermission at the top of its file, patching system_manage.models.WorkspaceUserGroupResourcePermission might not work unless you patch the name in the module where it is used (i.e., patch("oss.serializers.file.WorkspaceUserGroupResourcePermission")).
    • Crucial Check: Look at your other tests like test_explicit_view_and_manage_grants_allow_source_file_reads. They use WorkspaceUserResourcePermission: self.grants. Does this pattern work for existing tests? If yes, it’s consistent. If not, this new test might be ineffective because the patch isn’t applied to the correct object instance within the function under test.

5. Summary of Action Items

Category Severity Description Recommendation
Correctness High Patch target validity. Verify that patch correctly intercepts the model usage in _auth_system. If _auth_system imports the model directly, you may need to patch oss.serializers.file.WorkspaceUserGroupResourcePermission instead of system_manage.models.WorkspaceUserGroupResourcePermission.
Coverage Medium Missing negative case. Add a test where group_grants.exists is False (and user_grants is False) to expect an authorization failure.
Maintainability Low Brittle assertions. The assertion on user_group__user_relations__user_id is tightly coupled to the ORM query structure. Acceptable for security-sensitive permission checks, but be prepared for maintenance overhead if query optimization occurs.
Best Practice Low Mock granularity. Consider mocking just the .filter() result or using mock.patch.object on the specific method if the model replacement approach proves unstable.

Suggested Additional Test Code

    def test_no_grants_denies_access(self):
        """Ensure access is denied when neither user nor group has permission"""
        self.grants.exists.return_value = False
        self.group_grants.exists.return_value = False
        
        with self.assertRaises(AppUnauthorizedFailed):
            _auth_system(self.file, "user")
    
    def test_user_group_permissions_are_checked_if_user_permissions_fail(self):
        """Ensure group check happens even if user-specific check returns False"""
        self.grants.exists.return_value = False
        self.group_grants.exists.return_value = True # Group has permission
        
        # Should NOT raise exception
        _auth_system(self.file, "user")
        
        # Verify group grants were checked
        self.assertTrue(self.group_grants.filter.called)

Expand Down
Loading