mirror of
https://github.com/paperless-ngx/paperless-ngx.git
synced 2026-10-15 12:37:13 +00:00
Compare commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
d891c368e8 |
No files matched your search
@@ -130,12 +130,14 @@ class DocumentMetadataOverrides:
|
|||||||
get_users_with_perms(
|
get_users_with_perms(
|
||||||
doc,
|
doc,
|
||||||
only_with_perms_in=["view_document"],
|
only_with_perms_in=["view_document"],
|
||||||
|
with_group_users=False,
|
||||||
).values_list("id", flat=True),
|
).values_list("id", flat=True),
|
||||||
)
|
)
|
||||||
overrides.change_users = list(
|
overrides.change_users = list(
|
||||||
get_users_with_perms(
|
get_users_with_perms(
|
||||||
doc,
|
doc,
|
||||||
only_with_perms_in=["change_document"],
|
only_with_perms_in=["change_document"],
|
||||||
|
with_group_users=False,
|
||||||
).values_list("id", flat=True),
|
).values_list("id", flat=True),
|
||||||
)
|
)
|
||||||
overrides.custom_fields = {
|
overrides.custom_fields = {
|
||||||
|
|||||||
@@ -1,11 +1,5 @@
|
|||||||
from collections.abc import Mapping
|
|
||||||
from collections.abc import Sequence
|
|
||||||
from typing import Any
|
from typing import Any
|
||||||
from typing import ClassVar
|
|
||||||
from typing import Literal
|
|
||||||
from typing import TypeGuard
|
|
||||||
from typing import TypeVar
|
from typing import TypeVar
|
||||||
from typing import cast
|
|
||||||
|
|
||||||
from django.contrib.auth.models import Group
|
from django.contrib.auth.models import Group
|
||||||
from django.contrib.auth.models import Permission
|
from django.contrib.auth.models import Permission
|
||||||
@@ -68,23 +62,28 @@ class PaperlessAdminPermissions(BasePermission):
|
|||||||
return request.user.is_active and request.user.is_staff
|
return request.user.is_active and request.user.is_staff
|
||||||
|
|
||||||
|
|
||||||
def _is_active_and_authenticated(user: User | None) -> TypeGuard[User]:
|
|
||||||
return (
|
|
||||||
user is not None
|
|
||||||
and getattr(user, "is_active", False)
|
|
||||||
and getattr(user, "is_authenticated", False)
|
|
||||||
)
|
|
||||||
|
|
||||||
|
|
||||||
def has_global_statistics_permission(user: User | None) -> bool:
|
def has_global_statistics_permission(user: User | None) -> bool:
|
||||||
return _is_active_and_authenticated(user) and (
|
if (
|
||||||
getattr(user, "is_superuser", False)
|
user is None
|
||||||
or user.has_perm("paperless.view_global_statistics")
|
or not getattr(user, "is_active", False)
|
||||||
|
or not getattr(user, "is_authenticated", False)
|
||||||
|
):
|
||||||
|
return False
|
||||||
|
|
||||||
|
return getattr(user, "is_superuser", False) or user.has_perm(
|
||||||
|
"paperless.view_global_statistics",
|
||||||
)
|
)
|
||||||
|
|
||||||
|
|
||||||
def has_system_status_permission(user: User | None) -> bool:
|
def has_system_status_permission(user: User | None) -> bool:
|
||||||
return _is_active_and_authenticated(user) and (
|
if (
|
||||||
|
user is None
|
||||||
|
or not getattr(user, "is_active", False)
|
||||||
|
or not getattr(user, "is_authenticated", False)
|
||||||
|
):
|
||||||
|
return False
|
||||||
|
|
||||||
|
return (
|
||||||
getattr(user, "is_superuser", False)
|
getattr(user, "is_superuser", False)
|
||||||
or getattr(user, "is_staff", False)
|
or getattr(user, "is_staff", False)
|
||||||
or user.has_perm("paperless.view_system_monitoring")
|
or user.has_perm("paperless.view_system_monitoring")
|
||||||
@@ -706,32 +705,7 @@ def has_perms_owner_aware(user, perms, obj):
|
|||||||
return obj.owner is None or obj.owner == user or checker.has_perm(perms, obj)
|
return obj.owner is None or obj.owner == user or checker.has_perm(perms, obj)
|
||||||
|
|
||||||
|
|
||||||
# HEAD is deliberately absent: GlobalPermissionsByMethod checks it against GET.
|
class ViewDocumentsPermissions(BasePermission):
|
||||||
MappedMethod = Literal["GET", "POST", "PUT", "PATCH", "DELETE", "OPTIONS"]
|
|
||||||
|
|
||||||
|
|
||||||
class GlobalPermissionsByMethod(BasePermission):
|
|
||||||
"""
|
|
||||||
Requires the global (model-level) permissions listed in ``perms_map`` for
|
|
||||||
the request's HTTP method. HEAD is checked against the ``"GET"`` entry,
|
|
||||||
since DRF runs the GET handler for it. A method missing from
|
|
||||||
``perms_map`` is denied.
|
|
||||||
"""
|
|
||||||
|
|
||||||
perms_map: ClassVar[Mapping[MappedMethod, Sequence[str]]]
|
|
||||||
|
|
||||||
def has_permission(self, request: Any, view: Any) -> bool:
|
|
||||||
if not request.user or not request.user.is_authenticated:
|
|
||||||
return False
|
|
||||||
|
|
||||||
method = "GET" if request.method == "HEAD" else request.method
|
|
||||||
# An unknown method is not a MappedMethod, but .get() returns None for
|
|
||||||
# it, so it is still denied.
|
|
||||||
perms = self.perms_map.get(cast("MappedMethod", method))
|
|
||||||
return perms is not None and request.user.has_perms(perms)
|
|
||||||
|
|
||||||
|
|
||||||
class ViewDocumentsPermissions(GlobalPermissionsByMethod):
|
|
||||||
"""
|
"""
|
||||||
Permissions class that checks for model permissions for only viewing Documents.
|
Permissions class that checks for model permissions for only viewing Documents.
|
||||||
"""
|
"""
|
||||||
@@ -742,18 +716,31 @@ class ViewDocumentsPermissions(GlobalPermissionsByMethod):
|
|||||||
"POST": ["documents.view_document"],
|
"POST": ["documents.view_document"],
|
||||||
}
|
}
|
||||||
|
|
||||||
|
def has_permission(self, request, view):
|
||||||
|
if not request.user or (not request.user.is_authenticated): # pragma: no cover
|
||||||
|
return False
|
||||||
|
|
||||||
class TrashPermissions(GlobalPermissionsByMethod):
|
return request.user.has_perms(self.perms_map.get(request.method, []))
|
||||||
|
|
||||||
|
|
||||||
|
class TrashPermissions(BasePermission):
|
||||||
"""Check the global document permission for each trash operation."""
|
"""Check the global document permission for each trash operation."""
|
||||||
|
|
||||||
perms_map = {
|
perms_map = {
|
||||||
"OPTIONS": ["documents.view_document"],
|
"OPTIONS": ["documents.view_document"],
|
||||||
|
"HEAD": ["documents.view_document"],
|
||||||
"GET": ["documents.view_document"],
|
"GET": ["documents.view_document"],
|
||||||
"POST": ["documents.delete_document"],
|
"POST": ["documents.delete_document"],
|
||||||
}
|
}
|
||||||
|
|
||||||
|
def has_permission(self, request, view):
|
||||||
|
if not request.user or not request.user.is_authenticated: # pragma: no cover
|
||||||
|
return False
|
||||||
|
|
||||||
class PaperlessNotePermissions(GlobalPermissionsByMethod):
|
return request.user.has_perms(self.perms_map.get(request.method, []))
|
||||||
|
|
||||||
|
|
||||||
|
class PaperlessNotePermissions(BasePermission):
|
||||||
"""
|
"""
|
||||||
Permissions class that checks for model permissions for Notes.
|
Permissions class that checks for model permissions for Notes.
|
||||||
"""
|
"""
|
||||||
@@ -773,13 +760,28 @@ class PaperlessNotePermissions(GlobalPermissionsByMethod):
|
|||||||
],
|
],
|
||||||
}
|
}
|
||||||
|
|
||||||
|
def has_permission(self, request, view):
|
||||||
|
if not request.user or (not request.user.is_authenticated): # pragma: no cover
|
||||||
|
return False
|
||||||
|
|
||||||
class AcknowledgeTasksPermissions(GlobalPermissionsByMethod):
|
perms = self.perms_map[request.method]
|
||||||
|
|
||||||
|
return request.user.has_perms(perms)
|
||||||
|
|
||||||
|
|
||||||
|
class AcknowledgeTasksPermissions(BasePermission):
|
||||||
"""
|
"""
|
||||||
Permissions class that checks for model permissions for acknowledging tasks.
|
Permissions class that checks for model permissions for acknowledging tasks.
|
||||||
"""
|
"""
|
||||||
|
|
||||||
perms_map = {
|
perms_map = {
|
||||||
"OPTIONS": ["documents.change_paperlesstask"],
|
|
||||||
"POST": ["documents.change_paperlesstask"],
|
"POST": ["documents.change_paperlesstask"],
|
||||||
}
|
}
|
||||||
|
|
||||||
|
def has_permission(self, request: Any, view: Any) -> bool:
|
||||||
|
if not request.user or not request.user.is_authenticated: # pragma: no cover
|
||||||
|
return False
|
||||||
|
|
||||||
|
perms = self.perms_map.get(request.method, [])
|
||||||
|
|
||||||
|
return request.user.has_perms(perms)
|
||||||
@@ -3653,34 +3653,6 @@ class TestDocumentApi(DirectoriesMixin, ConsumeTaskMixin, APITestCase):
|
|||||||
)
|
)
|
||||||
self.assertEqual(response.status_code, status.HTTP_200_OK)
|
self.assertEqual(response.status_code, status.HTTP_200_OK)
|
||||||
|
|
||||||
def test_notes_head_and_unsupported_methods(self) -> None:
|
|
||||||
"""
|
|
||||||
GIVEN:
|
|
||||||
- A user allowed to view notes on their own document
|
|
||||||
WHEN:
|
|
||||||
- The notes endpoint is requested with HEAD, PUT and PATCH
|
|
||||||
THEN:
|
|
||||||
- HEAD is allowed like GET
|
|
||||||
- PUT and PATCH are rejected with 403 rather than a server error
|
|
||||||
"""
|
|
||||||
user = UserFactory()
|
|
||||||
grant_global(user, "view_note", "view_document")
|
|
||||||
doc = DocumentFactory(owner=user)
|
|
||||||
self.client.force_authenticate(user)
|
|
||||||
|
|
||||||
self.assertEqual(
|
|
||||||
self.client.head(f"/api/documents/{doc.pk}/notes/").status_code,
|
|
||||||
status.HTTP_200_OK,
|
|
||||||
)
|
|
||||||
self.assertEqual(
|
|
||||||
self.client.put(f"/api/documents/{doc.pk}/notes/").status_code,
|
|
||||||
status.HTTP_403_FORBIDDEN,
|
|
||||||
)
|
|
||||||
self.assertEqual(
|
|
||||||
self.client.patch(f"/api/documents/{doc.pk}/notes/").status_code,
|
|
||||||
status.HTTP_403_FORBIDDEN,
|
|
||||||
)
|
|
||||||
|
|
||||||
def test_delete_note(self) -> None:
|
def test_delete_note(self) -> None:
|
||||||
"""
|
"""
|
||||||
GIVEN:
|
GIVEN:
|
||||||
|
|||||||
@@ -1001,21 +1001,6 @@ class TestDocumentSearchApi(DirectoriesMixin, APITestCase):
|
|||||||
self.assertEqual(response.status_code, status.HTTP_200_OK)
|
self.assertEqual(response.status_code, status.HTTP_200_OK)
|
||||||
self.assertEqual(len(response.data), 10)
|
self.assertEqual(len(response.data), 10)
|
||||||
|
|
||||||
def test_search_autocomplete_head_requires_view_document(self) -> None:
|
|
||||||
"""
|
|
||||||
GIVEN:
|
|
||||||
- A user without the global view document permission
|
|
||||||
WHEN:
|
|
||||||
- A HEAD request is made for autocomplete
|
|
||||||
THEN:
|
|
||||||
- It is rejected like the equivalent GET request
|
|
||||||
"""
|
|
||||||
self.client.force_authenticate(user=UserFactory())
|
|
||||||
|
|
||||||
response = self.client.head("/api/search/autocomplete/?term=test")
|
|
||||||
|
|
||||||
self.assertEqual(response.status_code, status.HTTP_403_FORBIDDEN)
|
|
||||||
|
|
||||||
def test_search_autocomplete_respect_permissions(self) -> None:
|
def test_search_autocomplete_respect_permissions(self) -> None:
|
||||||
"""
|
"""
|
||||||
GIVEN:
|
GIVEN:
|
||||||
|
|||||||
@@ -34,8 +34,10 @@ from documents.tests.utils import GetConsumerMixin
|
|||||||
from paperless_mail.models import MailRule
|
from paperless_mail.models import MailRule
|
||||||
from paperless_testing.assertions import FileSystemAssertsMixin
|
from paperless_testing.assertions import FileSystemAssertsMixin
|
||||||
from paperless_testing.dirs import DirectoriesMixin
|
from paperless_testing.dirs import DirectoriesMixin
|
||||||
|
from paperless_testing.factories import DocumentFactory
|
||||||
from paperless_testing.factories import UserFactory
|
from paperless_testing.factories import UserFactory
|
||||||
from paperless_testing.fakes.progress import FakeProgressManager
|
from paperless_testing.fakes.progress import FakeProgressManager
|
||||||
|
from paperless_testing.permissions import grant_object
|
||||||
|
|
||||||
|
|
||||||
class _BaseNewStyleParser:
|
class _BaseNewStyleParser:
|
||||||
@@ -1683,6 +1685,32 @@ class TestMetadataOverrides(TestCase):
|
|||||||
self.assertEqual(base.actor_id, 2)
|
self.assertEqual(base.actor_id, 2)
|
||||||
self.assertEqual(base.version_label, "v2")
|
self.assertEqual(base.version_label, "v2")
|
||||||
|
|
||||||
|
def test_from_document_copies_user_and_group_grants_separately(self) -> None:
|
||||||
|
"""
|
||||||
|
GIVEN:
|
||||||
|
- A document with view and change granted directly to one user and
|
||||||
|
to a group with another user as a member
|
||||||
|
WHEN:
|
||||||
|
- Overrides are built from the document
|
||||||
|
THEN:
|
||||||
|
- The user lists hold only the directly granted user
|
||||||
|
- The group lists hold the group
|
||||||
|
"""
|
||||||
|
direct_user = UserFactory()
|
||||||
|
group_member = UserFactory()
|
||||||
|
group = Group.objects.create(name="group")
|
||||||
|
group_member.groups.add(group)
|
||||||
|
document = DocumentFactory()
|
||||||
|
grant_object(direct_user, document, "view_document", "change_document")
|
||||||
|
grant_object(group, document, "view_document", "change_document")
|
||||||
|
|
||||||
|
overrides = DocumentMetadataOverrides.from_document(document)
|
||||||
|
|
||||||
|
self.assertEqual(overrides.view_users, [direct_user.pk])
|
||||||
|
self.assertEqual(overrides.change_users, [direct_user.pk])
|
||||||
|
self.assertEqual(overrides.view_groups, [group.pk])
|
||||||
|
self.assertEqual(overrides.change_groups, [group.pk])
|
||||||
|
|
||||||
|
|
||||||
class TestBarcodeApplyDetectedASN(TestCase):
|
class TestBarcodeApplyDetectedASN(TestCase):
|
||||||
"""
|
"""
|
||||||
|
|||||||
Reference in new issue
Block a user