diff --git a/src/documents/serialisers.py b/src/documents/serialisers.py index 2d6ff43a3..ca850e84f 100644 --- a/src/documents/serialisers.py +++ b/src/documents/serialisers.py @@ -864,10 +864,10 @@ def validate_documentlink_targets(user, doc_ids): if user is None: return - target_documents = Document.objects.filter(id__in=doc_ids).select_related("owner") - if not all( - has_perms_owner_aware(user, "change_document", document) - for document in target_documents + if ( + Document.objects.filter(id__in=doc_ids) + .exclude(id__in=permitted_document_ids(user, perm="change_document")) + .exists() ): raise PermissionDenied( _("Insufficient permissions."), diff --git a/src/documents/tests/test_permission_filtering_security.py b/src/documents/tests/test_permission_filtering_security.py index 4ec6cf13c..f7a55f2f2 100644 --- a/src/documents/tests/test_permission_filtering_security.py +++ b/src/documents/tests/test_permission_filtering_security.py @@ -1,5 +1,6 @@ from __future__ import annotations +from http import HTTPStatus from unittest.mock import patch import pytest @@ -193,7 +194,7 @@ class TestAiChatAllDocumentsPermissionBoundary: format="json", ) - assert response.status_code == 200 + assert response.status_code == HTTPStatus.OK mock_stream_chat.assert_called_once() _, kwargs = mock_stream_chat.call_args visible_ids = {doc.pk for doc in kwargs["documents"]} @@ -284,3 +285,149 @@ class TestPermittedDocumentIdsArbitraryPermission: expected_visible=[], expected_hidden=[doc.pk], ) + + +@pytest.mark.django_db +class TestEmailDocumentPermissionBoundary: + def test_email_action_rejects_document_without_view_permission( + self, + rest_api_client, + ): + owner = User.objects.create_user(username="owner") + requester = User.objects.create_user(username="requester") + rest_api_client.force_authenticate(user=requester) + hidden = DocumentFactory(owner=owner) + + response = rest_api_client.post( + "/api/documents/email/", + { + "documents": [hidden.pk], + "addresses": "someone@example.com", + "subject": "test", + "message": "test", + }, + format="json", + ) + assert response.status_code == HTTPStatus.FORBIDDEN + + +@pytest.mark.django_db +class TestBulkEditChangePermissionBoundary: + def test_bulk_edit_rejects_mixed_batch_when_any_document_lacks_change_permission( + self, + rest_api_client, + ): + # A bulk-edit request containing both a document the requester CAN + # change and one they CANNOT should be rejected as a whole: the + # permitted document must not be partially applied just because it + # was bundled with a forbidden one, proving the endpoint checks + # every document in the batch rather than only the first/last. + owner = User.objects.create_user(username="owner") + requester = User.objects.create_user(username="requester") + # grant the global change_document permission so the object-level + # check (not the global has_perm check) is what's under test + requester.user_permissions.add( + Permission.objects.get(codename="change_document"), + ) + rest_api_client.force_authenticate(user=requester) + changeable = DocumentFactory(owner=owner) + assign_perm("view_document", requester, changeable) + assign_perm("change_document", requester, changeable) # fully permitted + target = DocumentFactory(owner=owner) + assign_perm("view_document", requester, target) # view only, NOT change + + response = rest_api_client.post( + "/api/documents/bulk_edit/", + { + "documents": [changeable.pk, target.pk], + "method": "modify_tags", + "parameters": {"add_tags": [], "remove_tags": []}, + }, + format="json", + ) + assert response.status_code == HTTPStatus.FORBIDDEN + + +@pytest.mark.django_db +class TestBulkDownloadPermissionChecksRootDocument: + def test_permission_checked_on_root_not_on_version( + self, + rest_api_client, + paperless_dirs, + _media_settings, + ): + owner = User.objects.create_user(username="owner") + requester = User.objects.create_user(username="requester") + rest_api_client.force_authenticate(user=requester) + root = DocumentFactory(owner=owner) + # a version of root that the requester has NOT been individually granted + version = DocumentFactory(owner=owner, root_document=root, version_index=1) + version.source_path.write_bytes(b"%PDF-1.4 test") + assign_perm("view_document", requester, root) # granted on ROOT only + + response = rest_api_client.post( + "/api/documents/bulk_download/", + {"documents": [version.pk]}, + format="json", + ) + assert ( + response.status_code == HTTPStatus.OK + ) # visible because root is permitted + + # Granted on the VERSION itself, but NOT on the root. If the endpoint + # ever regressed to checking "root OR version" instead of root-only, + # this grant would incorrectly unlock access. This is the case that + # actually discriminates correct (root-only) enforcement from a + # root-or-version bug; a user with no grant at all (the old + # `stranger` case) can't tell the two apart, since they're denied + # either way. + version_only_grantee = User.objects.create_user(username="version_only_grantee") + assign_perm("view_document", version_only_grantee, version) + rest_api_client.force_authenticate(user=version_only_grantee) + response = rest_api_client.post( + "/api/documents/bulk_download/", + {"documents": [version.pk]}, + format="json", + ) + assert ( + response.status_code == HTTPStatus.FORBIDDEN + ) # version-only grant must not substitute for root permission + + +@pytest.mark.django_db +class TestTrashRestorePermissionBoundary: + def test_restore_rejects_document_without_delete_permission( + self, + rest_api_client, + ): + owner = User.objects.create_user(username="owner") + requester = User.objects.create_user(username="requester") + rest_api_client.force_authenticate(user=requester) + doc = DocumentFactory(owner=owner) + assign_perm("view_document", requester, doc) # view only, NOT delete + doc.delete() + + response = rest_api_client.post( + "/api/trash/", + {"documents": [doc.pk], "action": "restore"}, + format="json", + ) + assert response.status_code == HTTPStatus.FORBIDDEN + + def test_restore_allows_document_with_explicit_delete_permission( + self, + rest_api_client, + ): + owner = User.objects.create_user(username="owner") + requester = User.objects.create_user(username="requester") + rest_api_client.force_authenticate(user=requester) + doc = DocumentFactory(owner=owner) + assign_perm("delete_document", requester, doc) + doc.delete() + + response = rest_api_client.post( + "/api/trash/", + {"documents": [doc.pk], "action": "restore"}, + format="json", + ) + assert response.status_code == HTTPStatus.OK diff --git a/src/documents/tests/test_share_link_bundles.py b/src/documents/tests/test_share_link_bundles.py index 0040c5030..9583e7e47 100644 --- a/src/documents/tests/test_share_link_bundles.py +++ b/src/documents/tests/test_share_link_bundles.py @@ -60,7 +60,7 @@ class ShareLinkBundleAPITests(DirectoriesMixin, APITestCase): self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST) self.assertIn("document_ids", response.data) - @mock.patch("documents.views.has_perms_owner_aware", return_value=False) + @mock.patch("documents.views.permitted_document_ids", return_value=set()) def test_create_bundle_rejects_insufficient_permissions(self, perms_mock) -> None: payload = { "document_ids": [self.document.pk], diff --git a/src/documents/views.py b/src/documents/views.py index 65a92c0e9..a6bb6c955 100644 --- a/src/documents/views.py +++ b/src/documents/views.py @@ -1929,14 +1929,14 @@ class DocumentViewSet( message = validated_data.get("message") use_archive_version = validated_data.get("use_archive_version", True) - documents = Document.objects.select_related("owner").filter(pk__in=document_ids) - for document in documents: - if request.user is not None and not has_perms_owner_aware( - request.user, - "view_document", - document, - ): - return HttpResponseForbidden("Insufficient permissions") + documents = Document.objects.filter(pk__in=document_ids) + if ( + request.user is not None + and documents.exclude( + pk__in=permitted_document_ids(request.user), + ).exists() + ): + return HttpResponseForbidden("Insufficient permissions") try: attachments: list[EmailAttachment] = [] @@ -2789,8 +2789,13 @@ class DocumentOperationPermissionMixin(PassUserMixin, DocumentSelectionMixin): ) # check global and object permissions for all documents - has_perms = user.has_perm("documents.change_document") and all( - has_perms_owner_aware(user, "change_document", doc) for doc in document_objs + has_perms = ( + user.has_perm( + "documents.change_document", + ) + and not document_objs.exclude( + pk__in=permitted_document_ids(user, perm="change_document"), + ).exists() ) # check ownership for methods that change original document @@ -3843,9 +3848,10 @@ class BulkDownloadView(DocumentSelectionMixin, GenericAPIView[Any]): content = serializer.validated_data.get("content") follow_filename_format = serializer.validated_data.get("follow_formatting") + permitted_ids = set(permitted_document_ids(request.user)) for document in documents: root_doc = get_root_document(document) - if not has_perms_owner_aware(request.user, "view_document", root_doc): + if root_doc.pk not in permitted_ids: return HttpResponseForbidden("Insufficient permissions") versioned_documents.append( get_latest_version_for_root( @@ -4509,8 +4515,9 @@ class ShareLinkBundleViewSet(PassUserMixin, ModelViewSet[ShareLinkBundle]): ) documents = list(documents_qs) + permitted_ids = set(permitted_document_ids(request.user)) for document in documents: - if not has_perms_owner_aware(request.user, "view_document", document): + if document.pk not in permitted_ids: raise ValidationError( { "document_ids": _( @@ -5314,9 +5321,14 @@ class TrashView(ListModelMixin, PassUserMixin): if doc_ids is not None else self.filter_queryset(self.get_queryset()).all() ) - for doc in docs: - if not has_perms_owner_aware(request.user, "delete_document", doc): - return HttpResponseForbidden("Insufficient permissions") + if docs.exclude( + pk__in=permitted_document_ids( + request.user, + perm="delete_document", + include_deleted=True, + ), + ).exists(): + return HttpResponseForbidden("Insufficient permissions") action = serializer.validated_data.get("action") if action == "restore": for doc in Document.deleted_objects.filter(id__in=doc_ids).all():