mirror of
https://github.com/paperless-ngx/paperless-ngx.git
synced 2026-10-15 20:47:14 +00:00
Compare commits
2
Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
6c85e8877a | ||
|
|
6b700393a0 |
No files matched your search
@@ -1,5 +1,11 @@
|
||||
from collections.abc import Mapping
|
||||
from collections.abc import Sequence
|
||||
from typing import Any
|
||||
from typing import ClassVar
|
||||
from typing import Literal
|
||||
from typing import TypeGuard
|
||||
from typing import TypeVar
|
||||
from typing import cast
|
||||
|
||||
from django.contrib.auth.models import Group
|
||||
from django.contrib.auth.models import Permission
|
||||
@@ -62,28 +68,23 @@ class PaperlessAdminPermissions(BasePermission):
|
||||
return request.user.is_active and request.user.is_staff
|
||||
|
||||
|
||||
def has_global_statistics_permission(user: User | None) -> bool:
|
||||
if (
|
||||
user is None
|
||||
or not getattr(user, "is_active", False)
|
||||
or not getattr(user, "is_authenticated", False)
|
||||
):
|
||||
return False
|
||||
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)
|
||||
)
|
||||
|
||||
return getattr(user, "is_superuser", False) or user.has_perm(
|
||||
"paperless.view_global_statistics",
|
||||
|
||||
def has_global_statistics_permission(user: User | None) -> bool:
|
||||
return _is_active_and_authenticated(user) and (
|
||||
getattr(user, "is_superuser", False)
|
||||
or user.has_perm("paperless.view_global_statistics")
|
||||
)
|
||||
|
||||
|
||||
def has_system_status_permission(user: User | None) -> bool:
|
||||
if (
|
||||
user is None
|
||||
or not getattr(user, "is_active", False)
|
||||
or not getattr(user, "is_authenticated", False)
|
||||
):
|
||||
return False
|
||||
|
||||
return (
|
||||
return _is_active_and_authenticated(user) and (
|
||||
getattr(user, "is_superuser", False)
|
||||
or getattr(user, "is_staff", False)
|
||||
or user.has_perm("paperless.view_system_monitoring")
|
||||
@@ -705,7 +706,32 @@ def has_perms_owner_aware(user, perms, obj):
|
||||
return obj.owner is None or obj.owner == user or checker.has_perm(perms, obj)
|
||||
|
||||
|
||||
class ViewDocumentsPermissions(BasePermission):
|
||||
# HEAD is deliberately absent: GlobalPermissionsByMethod checks it against GET.
|
||||
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.
|
||||
"""
|
||||
@@ -716,31 +742,18 @@ class ViewDocumentsPermissions(BasePermission):
|
||||
"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
|
||||
|
||||
return request.user.has_perms(self.perms_map.get(request.method, []))
|
||||
|
||||
|
||||
class TrashPermissions(BasePermission):
|
||||
class TrashPermissions(GlobalPermissionsByMethod):
|
||||
"""Check the global document permission for each trash operation."""
|
||||
|
||||
perms_map = {
|
||||
"OPTIONS": ["documents.view_document"],
|
||||
"HEAD": ["documents.view_document"],
|
||||
"GET": ["documents.view_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
|
||||
|
||||
return request.user.has_perms(self.perms_map.get(request.method, []))
|
||||
|
||||
|
||||
class PaperlessNotePermissions(BasePermission):
|
||||
class PaperlessNotePermissions(GlobalPermissionsByMethod):
|
||||
"""
|
||||
Permissions class that checks for model permissions for Notes.
|
||||
"""
|
||||
@@ -760,28 +773,13 @@ class PaperlessNotePermissions(BasePermission):
|
||||
],
|
||||
}
|
||||
|
||||
def has_permission(self, request, view):
|
||||
if not request.user or (not request.user.is_authenticated): # pragma: no cover
|
||||
return False
|
||||
|
||||
perms = self.perms_map[request.method]
|
||||
|
||||
return request.user.has_perms(perms)
|
||||
|
||||
|
||||
class AcknowledgeTasksPermissions(BasePermission):
|
||||
class AcknowledgeTasksPermissions(GlobalPermissionsByMethod):
|
||||
"""
|
||||
Permissions class that checks for model permissions for acknowledging tasks.
|
||||
"""
|
||||
|
||||
perms_map = {
|
||||
"OPTIONS": ["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,6 +3653,34 @@ class TestDocumentApi(DirectoriesMixin, ConsumeTaskMixin, APITestCase):
|
||||
)
|
||||
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:
|
||||
"""
|
||||
GIVEN:
|
||||
|
||||
@@ -1001,6 +1001,21 @@ class TestDocumentSearchApi(DirectoriesMixin, APITestCase):
|
||||
self.assertEqual(response.status_code, status.HTTP_200_OK)
|
||||
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:
|
||||
"""
|
||||
GIVEN:
|
||||
|
||||
Reference in new issue
Block a user