diff --git a/src/documents/serialisers.py b/src/documents/serialisers.py index c2c875029..ec0cf2812 100644 --- a/src/documents/serialisers.py +++ b/src/documents/serialisers.py @@ -90,6 +90,7 @@ from documents.templating.utils import convert_format_str_to_template_format from documents.templating.workflows import validate_workflow_template from documents.validators import uri_validator from documents.validators import url_validator +from documents.versioning import get_root_document from documents.versioning import has_prefetched_effective_content from documents.versioning import sort_versions_newest_first @@ -2894,7 +2895,7 @@ class ShareLinkSerializer(OwnedObjectSerializer): and has_perms_owner_aware( self.user, "view_document", - document, + get_root_document(document), ) ): return document diff --git a/src/documents/tests/test_api_document_versions.py b/src/documents/tests/test_api_document_versions.py index 6fce8fa49..890853f7a 100644 --- a/src/documents/tests/test_api_document_versions.py +++ b/src/documents/tests/test_api_document_versions.py @@ -8,6 +8,7 @@ from auditlog.models import LogEntry # type: ignore[import-untyped] from django.contrib.contenttypes.models import ContentType from django.core.files.uploadedfile import SimpleUploadedFile from django.test import TestCase as DjangoTestCase +from django.test import override_settings from django.utils import timezone from rest_framework import status from rest_framework.test import APITestCase @@ -16,13 +17,17 @@ from documents.data_models import DocumentSource from documents.filters import EffectiveContentFilter from documents.filters import TitleContentFilter from documents.models import Document +from documents.models import Note +from documents.models import ShareLink from documents.versioning import annotate_effective_content from documents.views import DocumentSelectionMixin from paperless_testing.dirs import DirectoriesMixin from paperless_testing.factories import DocumentFactory from paperless_testing.factories import UserFactory from paperless_testing.http import read_streaming_response +from paperless_testing.permissions import grant_all_global from paperless_testing.permissions import grant_global +from paperless_testing.permissions import grant_object if TYPE_CHECKING: from pathlib import Path @@ -1043,3 +1048,110 @@ class TestBulkSelectionExcludesVersions(DjangoTestCase): ) self.assertEqual(selected, [root.id]) + + +class TestVersionActionPermissions(DirectoriesMixin, APITestCase): + def setUp(self): + super().setUp() + self.user = UserFactory() + grant_all_global(self.user) + self.client.force_authenticate(self.user) + self.root = DocumentFactory(owner=UserFactory()) + self.version = DocumentFactory(root_document=self.root, owner=None) + + @override_settings(AUDIT_LOG_ENABLED=True) + def test_actions_reject_stale_version_ownership(self): + note = Note.objects.create(document=self.version, note="Version note") + for owner in (None, self.user): + self.version.owner = owner + self.version.save(update_fields=["owner"]) + for action in ("notes", "suggestions", "history", "share_links"): + with self.subTest(owner=owner, action=action): + response = self.client.get( + f"/api/documents/{self.version.pk}/{action}/", + ) + self.assertEqual(response.status_code, 403) + response = self.client.post( + f"/api/documents/{self.version.pk}/notes/", + {"note": "New note"}, + ) + self.assertEqual(response.status_code, 403) + response = self.client.delete( + f"/api/documents/{self.version.pk}/notes/?id={note.pk}", + ) + self.assertEqual(response.status_code, 403) + response = self.client.post( + "/api/share_links/", + {"document": self.version.pk, "file_version": "original"}, + ) + self.assertEqual(response.status_code, 403) + response = self.client.post( + "/api/share_link_bundles/", + {"document_ids": [self.version.pk], "file_version": "original"}, + format="json", + ) + self.assertEqual(response.status_code, 400) + response = self.client.post( + "/api/documents/email/", + { + "documents": [self.version.pk], + "addresses": "recipient@example.com", + "subject": "Version", + "message": "Version", + }, + format="json", + ) + self.assertEqual(response.status_code, 403) + self.assertTrue(Note.objects.filter(pk=note.pk).exists()) + self.assertFalse(ShareLink.objects.exists()) + + @mock.patch("documents.views.build_share_link_bundle.apply_async") + def test_root_permissions_allow_sharing_a_private_version(self, build_mock): + self.version.owner = UserFactory() + self.version.save(update_fields=["owner"]) + grant_object(self.user, self.root, "view_document", "change_document") + note = Note.objects.create(document=self.version, note="Version note") + response = self.client.get(f"/api/documents/{self.version.pk}/notes/") + self.assertEqual(response.status_code, 200) + self.assertEqual(response.data[0]["id"], note.pk) + response = self.client.post( + "/api/share_links/", + {"document": self.version.pk, "file_version": "original"}, + ) + self.assertEqual(response.status_code, 201) + self.assertEqual(ShareLink.objects.get().document_id, self.version.pk) + response = self.client.get(f"/api/documents/{self.version.pk}/share_links/") + self.assertEqual(response.status_code, 200) + self.assertEqual(len(response.data), 1) + response = self.client.post( + "/api/share_link_bundles/", + {"document_ids": [self.version.pk], "file_version": "original"}, + format="json", + ) + self.assertEqual(response.status_code, 201) + build_mock.assert_called_once() + + def test_root_view_permission_does_not_allow_note_changes(self): + grant_object(self.user, self.root, "view_document") + note = Note.objects.create(document=self.version, note="Version note") + response = self.client.get(f"/api/documents/{self.version.pk}/notes/") + self.assertEqual(response.status_code, 200) + response = self.client.post( + f"/api/documents/{self.version.pk}/notes/", + {"note": "New note"}, + ) + self.assertEqual(response.status_code, 403) + response = self.client.delete( + f"/api/documents/{self.version.pk}/notes/?id={note.pk}", + ) + self.assertEqual(response.status_code, 403) + self.assertTrue(Note.objects.filter(pk=note.pk).exists()) + + @override_settings(AUDIT_LOG_ENABLED=True) + def test_history_uses_root_ownership(self): + self.root.owner = self.user + self.root.save(update_fields=["owner"]) + self.version.owner = UserFactory() + self.version.save(update_fields=["owner"]) + response = self.client.get(f"/api/documents/{self.version.pk}/history/") + self.assertEqual(response.status_code, 200) diff --git a/src/documents/views.py b/src/documents/views.py index 515fd33d9..9c15f29b0 100644 --- a/src/documents/views.py +++ b/src/documents/views.py @@ -1550,13 +1550,16 @@ class DocumentViewSet( ) def suggestions(self, request, pk=None): doc = get_object_or_404( - Document.objects.select_related("owner").prefetch_related("versions"), + Document.objects.select_related( + "owner", + "root_document__owner", + ).prefetch_related("versions"), pk=pk, ) if request.user is not None and not has_perms_owner_aware( request.user, "change_document", - doc, + get_root_document(doc), ): return HttpResponseForbidden("Insufficient permissions") @@ -1856,15 +1859,15 @@ class DocumentViewSet( currentUser = request.user try: doc = ( - Document.objects.select_related("owner") + Document.objects.select_related("owner", "root_document__owner") .prefetch_related("notes") - .only("pk", "owner__id") + .only("pk", "owner__id", "root_document__owner__id") .get(pk=pk) ) if currentUser is not None and not has_perms_owner_aware( currentUser, "view_document", - doc, + get_root_document(doc), ): return HttpResponseForbidden("Insufficient permissions to view notes") except Document.DoesNotExist: @@ -1886,7 +1889,7 @@ class DocumentViewSet( if currentUser is not None and not has_perms_owner_aware( currentUser, "change_document", - doc, + get_root_document(doc), ): return HttpResponseForbidden( "Insufficient permissions to create notes", @@ -1929,7 +1932,7 @@ class DocumentViewSet( if currentUser is not None and not has_perms_owner_aware( currentUser, "change_document", - doc, + get_root_document(doc), ): return HttpResponseForbidden("Insufficient permissions to delete notes") @@ -1973,11 +1976,13 @@ class DocumentViewSet( def share_links(self, request, pk=None): currentUser = request.user try: - doc = Document.objects.select_related("owner").get(pk=pk) + doc = Document.objects.select_related("owner", "root_document__owner").get( + pk=pk, + ) if currentUser is not None and not has_perms_owner_aware( currentUser, "change_document", - doc, + get_root_document(doc), ): return HttpResponseForbidden( "Insufficient permissions to add share link", @@ -2008,10 +2013,11 @@ class DocumentViewSet( if not settings.AUDIT_LOG_ENABLED: return HttpResponseBadRequest("Audit log is disabled") try: - doc = Document.objects.get(pk=pk) + doc = Document.objects.select_related("root_document__owner").get(pk=pk) + root_doc = get_root_document(doc) if not request.user.has_perm("auditlog.view_logentry") or ( - doc.owner is not None - and doc.owner != request.user + root_doc.owner is not None + and root_doc.owner != request.user and not request.user.is_superuser ): return HttpResponseForbidden( @@ -2099,14 +2105,13 @@ class DocumentViewSet( message = validated_data.get("message") use_archive_version = validated_data.get("use_archive_version", True) - 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") + documents = Document.objects.filter(pk__in=document_ids).select_related( + "root_document__owner", + ) + if request.user is not None: + permitted_ids = set(permitted_document_ids(request.user)) + if any(get_root_document(doc).pk not in permitted_ids for doc in documents): + return HttpResponseForbidden("Insufficient permissions") try: attachments: list[EmailAttachment] = [] @@ -4777,6 +4782,7 @@ class ShareLinkBundleViewSet(PassUserMixin, ModelViewSet[ShareLinkBundle]): document_ids = serializer.validated_data["document_ids"] documents_qs = Document.objects.filter(pk__in=document_ids).select_related( "owner", + "root_document__owner", ) found_ids = set(documents_qs.values_list("pk", flat=True)) missing = sorted(set(document_ids) - found_ids) @@ -4793,7 +4799,7 @@ class ShareLinkBundleViewSet(PassUserMixin, ModelViewSet[ShareLinkBundle]): documents = list(documents_qs) permitted_ids = set(permitted_document_ids(request.user)) for document in documents: - if document.pk not in permitted_ids: + if get_root_document(document).pk not in permitted_ids: raise ValidationError( { "document_ids": _(