mirror of
https://github.com/paperless-ngx/paperless-ngx.git
synced 2026-07-22 20:04:58 +00:00
Compare commits
2
Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
f68093fcb8 | ||
|
|
f613082249 |
@@ -392,15 +392,34 @@ class OwnedObjectSerializer(
|
||||
}
|
||||
|
||||
def get_user_can_change(self, obj) -> bool:
|
||||
checker = ObjectPermissionChecker(self.user) if self.user is not None else None
|
||||
return (
|
||||
obj.owner is None
|
||||
or obj.owner == self.user
|
||||
or (
|
||||
self.user is not None
|
||||
and checker.has_perm(f"change_{obj.__class__.__name__.lower()}", obj)
|
||||
if obj.owner is None or obj.owner == self.user:
|
||||
return True
|
||||
if self.user is None:
|
||||
return False
|
||||
if self.user.is_active and self.user.is_superuser:
|
||||
# Mirrors guardian's own ObjectPermissionChecker.has_perm() shortcut --
|
||||
# superusers aren't necessarily granted explicit object permissions,
|
||||
# so the batched context below would otherwise incorrectly say no.
|
||||
return True
|
||||
|
||||
# Prefer the page-level batch computed by BulkPermissionMixin
|
||||
# (get_serializer_context) over a fresh per-object guardian check,
|
||||
# which would otherwise query the permission tables once per row.
|
||||
users_change_perms = self.context.get("users_change_perms")
|
||||
groups_change_perms = self.context.get("groups_change_perms")
|
||||
if users_change_perms is not None and groups_change_perms is not None:
|
||||
if self.user.pk in users_change_perms.get(obj.pk, []):
|
||||
return True
|
||||
user_group_ids = getattr(self, "_user_group_ids", None)
|
||||
if user_group_ids is None:
|
||||
user_group_ids = set(self.user.groups.values_list("id", flat=True))
|
||||
self._user_group_ids = user_group_ids
|
||||
return bool(
|
||||
user_group_ids.intersection(groups_change_perms.get(obj.pk, [])),
|
||||
)
|
||||
)
|
||||
|
||||
checker = ObjectPermissionChecker(self.user)
|
||||
return checker.has_perm(f"change_{obj.__class__.__name__.lower()}", obj)
|
||||
|
||||
@staticmethod
|
||||
def get_shared_object_pks(objects: Iterable):
|
||||
|
||||
@@ -568,6 +568,30 @@ class TestApiAuth(DirectoriesMixin, APITestCase):
|
||||
self.assertNotIn("user_can_change", results[0])
|
||||
self.assertNotIn("is_shared_by_requester", results[0])
|
||||
|
||||
def test_superuser_user_can_change_without_explicit_grant(self) -> None:
|
||||
"""
|
||||
A superuser has implicit change access to every document, even one
|
||||
owned by someone else with no explicit guardian grant -- mirrors
|
||||
guardian's own ObjectPermissionChecker.has_perm() superuser shortcut.
|
||||
"""
|
||||
superuser = User.objects.create_superuser(username="admin")
|
||||
other_user = User.objects.create_user(username="user2")
|
||||
Document.objects.create(
|
||||
title="Test",
|
||||
content="content",
|
||||
checksum="1",
|
||||
owner=other_user,
|
||||
)
|
||||
|
||||
self.client.force_authenticate(superuser)
|
||||
|
||||
response = self.client.get("/api/documents/", format="json")
|
||||
|
||||
self.assertEqual(response.status_code, status.HTTP_200_OK)
|
||||
results = response.json()["results"]
|
||||
self.assertEqual(len(results), 1)
|
||||
self.assertTrue(results[0]["user_can_change"])
|
||||
|
||||
@mock.patch("allauth.mfa.adapter.DefaultMFAAdapter.is_mfa_enabled")
|
||||
def test_basic_auth_mfa_enabled(self, mock_is_mfa_enabled) -> None:
|
||||
"""
|
||||
|
||||
+15
-7
@@ -431,14 +431,12 @@ class BulkPermissionMixin:
|
||||
This avoid fetching permissions object by object in database.
|
||||
"""
|
||||
context = super().get_serializer_context()
|
||||
try:
|
||||
full_perms = get_boolean(
|
||||
str(self.request.query_params.get("full_perms", "false")),
|
||||
)
|
||||
except ValueError:
|
||||
full_perms = False
|
||||
|
||||
if not full_perms:
|
||||
if getattr(self, "action", None) != "list":
|
||||
# Batching only pays off across a page of objects; for single-object
|
||||
# actions (retrieve, update, ...) the per-object fallback in
|
||||
# get_user_can_change()/_get_perms() is cheap and avoids scanning
|
||||
# the whole queryset here.
|
||||
return context
|
||||
|
||||
# Check which objects are being paginated
|
||||
@@ -943,6 +941,7 @@ class EmailDocumentDetailSchema(EmailSerializer):
|
||||
),
|
||||
)
|
||||
class DocumentViewSet(
|
||||
BulkPermissionMixin,
|
||||
PassUserMixin,
|
||||
RetrieveModelMixin,
|
||||
UpdateModelMixin,
|
||||
@@ -2291,6 +2290,15 @@ class UnifiedSearchViewSet(DocumentViewSet):
|
||||
return SearchResultSerializer
|
||||
return DocumentSerializer
|
||||
|
||||
def get_serializer_context(self):
|
||||
if self._is_search_request():
|
||||
# BulkPermissionMixin.get_serializer_context() (inherited via
|
||||
# DocumentViewSet) assumes it's batching permissions for a page of
|
||||
# real Document instances. Tantivy search results are SearchHit/
|
||||
# dict-like objects instead, so skip straight past it here.
|
||||
return super(BulkPermissionMixin, self).get_serializer_context()
|
||||
return super().get_serializer_context()
|
||||
|
||||
def _get_active_search_params(self, request: Request | None = None) -> list[str]:
|
||||
request = request or self.request
|
||||
return [
|
||||
|
||||
Reference in New Issue
Block a user