mirror of
https://github.com/paperless-ngx/paperless-ngx.git
synced 2026-09-14 21:58:00 +00:00
Fix: update some api global perms inconsistencies (#14086)
This commit is contained in:
@@ -4,6 +4,7 @@ import json
|
||||
import shutil
|
||||
import zipfile
|
||||
|
||||
from django.contrib.auth.models import Permission
|
||||
from django.contrib.auth.models import User
|
||||
from django.test import override_settings
|
||||
from django.utils import timezone
|
||||
@@ -326,6 +327,9 @@ class TestBulkDownload(DirectoriesMixin, SampleDirMixin, APITestCase):
|
||||
|
||||
def test_download_insufficient_permissions(self) -> None:
|
||||
user = User.objects.create_user(username="temp_user")
|
||||
user.user_permissions.add(
|
||||
Permission.objects.get(codename="view_document"),
|
||||
)
|
||||
self.client.force_authenticate(user=user)
|
||||
|
||||
self.doc2.owner = self.user
|
||||
|
||||
@@ -1084,6 +1084,8 @@ class TestBulkEditAPI(DirectoriesMixin, APITestCase):
|
||||
user1 = User.objects.create(username="user1")
|
||||
self.client.force_authenticate(user=user1)
|
||||
|
||||
assign_perm("view_document", user1, self.doc2)
|
||||
|
||||
response = self.client.post(
|
||||
"/api/documents/selection_data/",
|
||||
json.dumps({"documents": [self.doc2.id]}),
|
||||
@@ -1091,7 +1093,18 @@ class TestBulkEditAPI(DirectoriesMixin, APITestCase):
|
||||
)
|
||||
|
||||
self.assertEqual(response.status_code, status.HTTP_403_FORBIDDEN)
|
||||
self.assertEqual(response.content, b"Insufficient permissions")
|
||||
|
||||
user1.user_permissions.add(
|
||||
Permission.objects.get(codename="view_document"),
|
||||
)
|
||||
user1 = User.objects.get(pk=user1.pk)
|
||||
self.client.force_authenticate(user=user1)
|
||||
response = self.client.post(
|
||||
"/api/documents/selection_data/",
|
||||
json.dumps({"documents": [self.doc2.id]}),
|
||||
content_type="application/json",
|
||||
)
|
||||
self.assertEqual(response.status_code, status.HTTP_200_OK)
|
||||
|
||||
@mock.patch("documents.serialisers.bulk_edit.set_permissions")
|
||||
def test_set_permissions(self, m) -> None:
|
||||
|
||||
@@ -3615,6 +3615,55 @@ class TestDocumentApi(DirectoriesMixin, ConsumeTaskMixin, APITestCase):
|
||||
self.assertEqual(response.content, b"Insufficient permissions to delete notes")
|
||||
self.assertEqual(response.status_code, status.HTTP_403_FORBIDDEN)
|
||||
|
||||
def test_notes_require_global_document_permissions(self) -> None:
|
||||
user = User.objects.create_user(username="note_editor")
|
||||
user.user_permissions.add(
|
||||
*Permission.objects.filter(
|
||||
codename__in=["view_note", "add_note", "delete_note"],
|
||||
),
|
||||
)
|
||||
doc = Document.objects.create(
|
||||
title="test",
|
||||
mime_type="application/pdf",
|
||||
content="notes",
|
||||
owner=user,
|
||||
)
|
||||
note = Note.objects.create(note="Existing", document=doc, user=user)
|
||||
self.client.force_authenticate(user)
|
||||
|
||||
response = self.client.get(f"/api/documents/{doc.pk}/notes/")
|
||||
self.assertEqual(response.status_code, status.HTTP_403_FORBIDDEN)
|
||||
|
||||
user.user_permissions.add(
|
||||
Permission.objects.get(codename="view_document"),
|
||||
)
|
||||
user = User.objects.get(pk=user.pk)
|
||||
self.client.force_authenticate(user)
|
||||
response = self.client.get(f"/api/documents/{doc.pk}/notes/")
|
||||
self.assertEqual(response.status_code, status.HTTP_200_OK)
|
||||
|
||||
response = self.client.post(
|
||||
f"/api/documents/{doc.pk}/notes/",
|
||||
data={"note": "New"},
|
||||
)
|
||||
self.assertEqual(response.status_code, status.HTTP_403_FORBIDDEN)
|
||||
|
||||
user.user_permissions.add(
|
||||
Permission.objects.get(codename="change_document"),
|
||||
)
|
||||
user = User.objects.get(pk=user.pk)
|
||||
self.client.force_authenticate(user)
|
||||
response = self.client.post(
|
||||
f"/api/documents/{doc.pk}/notes/",
|
||||
data={"note": "New"},
|
||||
)
|
||||
self.assertEqual(response.status_code, status.HTTP_200_OK)
|
||||
|
||||
response = self.client.delete(
|
||||
f"/api/documents/{doc.pk}/notes/?id={note.pk}",
|
||||
)
|
||||
self.assertEqual(response.status_code, status.HTTP_200_OK)
|
||||
|
||||
def test_delete_note(self) -> None:
|
||||
"""
|
||||
GIVEN:
|
||||
@@ -3981,6 +4030,21 @@ class TestDocumentApi(DirectoriesMixin, ConsumeTaskMixin, APITestCase):
|
||||
|
||||
assign_perm("view_document", user1, doc)
|
||||
|
||||
create_resp = self.client.post(
|
||||
"/api/share_links/",
|
||||
data={
|
||||
"document": doc.pk,
|
||||
"file_version": "original",
|
||||
},
|
||||
format="json",
|
||||
)
|
||||
self.assertEqual(create_resp.status_code, status.HTTP_403_FORBIDDEN)
|
||||
|
||||
user1.user_permissions.add(
|
||||
Permission.objects.get(codename="view_document"),
|
||||
)
|
||||
user1 = User.objects.get(pk=user1.pk)
|
||||
self.client.force_authenticate(user1)
|
||||
create_resp = self.client.post(
|
||||
"/api/share_links/",
|
||||
data={
|
||||
|
||||
@@ -457,6 +457,9 @@ class TestApiStoragePaths(DirectoriesMixin, APITestCase):
|
||||
def test_test_storage_path_requires_document_view_permission(self) -> None:
|
||||
owner = User.objects.create_user(username="owner")
|
||||
unprivileged = User.objects.create_user(username="unprivileged")
|
||||
unprivileged.user_permissions.add(
|
||||
Permission.objects.get(codename="view_document"),
|
||||
)
|
||||
document = Document.objects.create(
|
||||
mime_type="application/pdf",
|
||||
owner=owner,
|
||||
@@ -488,6 +491,23 @@ class TestApiStoragePaths(DirectoriesMixin, APITestCase):
|
||||
)
|
||||
assign_perm("view_document", viewer, document)
|
||||
|
||||
self.client.force_authenticate(user=viewer)
|
||||
response = self.client.post(
|
||||
f"{self.ENDPOINT}test/",
|
||||
json.dumps(
|
||||
{
|
||||
"document": document.id,
|
||||
"path": "path/{{ title }}",
|
||||
},
|
||||
),
|
||||
content_type="application/json",
|
||||
)
|
||||
self.assertEqual(response.status_code, status.HTTP_403_FORBIDDEN)
|
||||
|
||||
viewer.user_permissions.add(
|
||||
Permission.objects.get(codename="view_document"),
|
||||
)
|
||||
viewer = User.objects.get(pk=viewer.pk)
|
||||
self.client.force_authenticate(user=viewer)
|
||||
response = self.client.post(
|
||||
f"{self.ENDPOINT}test/",
|
||||
@@ -530,6 +550,9 @@ class TestApiStoragePaths(DirectoriesMixin, APITestCase):
|
||||
password="password",
|
||||
email="owner@example.com",
|
||||
)
|
||||
owner.user_permissions.add(
|
||||
Permission.objects.get(codename="view_document"),
|
||||
)
|
||||
document = Document.objects.create(
|
||||
mime_type="application/pdf",
|
||||
owner=owner,
|
||||
@@ -605,6 +628,9 @@ class TestApiStoragePaths(DirectoriesMixin, APITestCase):
|
||||
checksum="123",
|
||||
)
|
||||
assign_perm("view_document", viewer, document)
|
||||
viewer.user_permissions.add(
|
||||
Permission.objects.get(codename="view_document"),
|
||||
)
|
||||
|
||||
self.client.force_authenticate(user=viewer)
|
||||
response = self.client.post(
|
||||
@@ -692,6 +718,9 @@ class TestApiStoragePaths(DirectoriesMixin, APITestCase):
|
||||
)
|
||||
document.tags.add(private_tag)
|
||||
assign_perm("view_document", viewer, document)
|
||||
viewer.user_permissions.add(
|
||||
Permission.objects.get(codename="view_document"),
|
||||
)
|
||||
|
||||
self.client.force_authenticate(user=viewer)
|
||||
response = self.client.post(
|
||||
@@ -745,6 +774,9 @@ class TestApiStoragePaths(DirectoriesMixin, APITestCase):
|
||||
value_int=42,
|
||||
)
|
||||
assign_perm("view_document", viewer, document)
|
||||
viewer.user_permissions.add(
|
||||
Permission.objects.get(codename="view_document"),
|
||||
)
|
||||
|
||||
self.client.force_authenticate(user=viewer)
|
||||
response = self.client.post(
|
||||
|
||||
@@ -69,6 +69,16 @@ class TestTrashAPI(DirectoriesMixin, APITestCase):
|
||||
self.assertEqual(resp.status_code, status.HTTP_200_OK)
|
||||
self.assertEqual(Document.global_objects.count(), 0)
|
||||
|
||||
def test_trash_list_requires_global_document_view_permission(self) -> None:
|
||||
user = User.objects.create_user(username="trash_owner")
|
||||
document = Document.objects.create(title="Owned", owner=user)
|
||||
document.delete()
|
||||
self.client.force_authenticate(user)
|
||||
|
||||
response = self.client.get("/api/trash/")
|
||||
|
||||
self.assertEqual(response.status_code, status.HTTP_403_FORBIDDEN)
|
||||
|
||||
def test_trash_api_empty_all(self) -> None:
|
||||
"""
|
||||
GIVEN:
|
||||
|
||||
@@ -309,6 +309,9 @@ class TestEmailDocumentPermissionBoundary:
|
||||
):
|
||||
owner = User.objects.create_user(username="owner")
|
||||
requester = User.objects.create_user(username="requester")
|
||||
requester.user_permissions.add(
|
||||
Permission.objects.get(codename="view_document"),
|
||||
)
|
||||
rest_api_client.force_authenticate(user=requester)
|
||||
hidden = DocumentFactory(owner=owner)
|
||||
|
||||
@@ -364,6 +367,27 @@ class TestBulkEditChangePermissionBoundary:
|
||||
|
||||
@pytest.mark.django_db
|
||||
class TestBulkDownloadPermissionChecksRootDocument:
|
||||
def test_download_requires_global_view_permission(
|
||||
self,
|
||||
rest_api_client,
|
||||
paperless_dirs,
|
||||
_media_settings,
|
||||
):
|
||||
owner = User.objects.create_user(username="owner")
|
||||
requester = User.objects.create_user(username="requester")
|
||||
root = DocumentFactory(owner=owner)
|
||||
root.source_path.write_bytes(b"%PDF-1.4 test")
|
||||
assign_perm("view_document", requester, root)
|
||||
rest_api_client.force_authenticate(user=requester)
|
||||
|
||||
response = rest_api_client.post(
|
||||
"/api/documents/bulk_download/",
|
||||
{"documents": [root.pk]},
|
||||
format="json",
|
||||
)
|
||||
|
||||
assert response.status_code == HTTPStatus.FORBIDDEN
|
||||
|
||||
def test_permission_checked_on_root_not_on_version(
|
||||
self,
|
||||
rest_api_client,
|
||||
@@ -372,6 +396,9 @@ class TestBulkDownloadPermissionChecksRootDocument:
|
||||
):
|
||||
owner = User.objects.create_user(username="owner")
|
||||
requester = User.objects.create_user(username="requester")
|
||||
requester.user_permissions.add(
|
||||
Permission.objects.get(codename="view_document"),
|
||||
)
|
||||
rest_api_client.force_authenticate(user=requester)
|
||||
root = DocumentFactory(owner=owner)
|
||||
# a version of root that the requester has NOT been individually granted
|
||||
@@ -396,6 +423,9 @@ class TestBulkDownloadPermissionChecksRootDocument:
|
||||
# `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")
|
||||
version_only_grantee.user_permissions.add(
|
||||
Permission.objects.get(codename="view_document"),
|
||||
)
|
||||
assign_perm("view_document", version_only_grantee, version)
|
||||
rest_api_client.force_authenticate(user=version_only_grantee)
|
||||
response = rest_api_client.post(
|
||||
@@ -417,6 +447,9 @@ class TestTrashRestorePermissionBoundary:
|
||||
):
|
||||
owner = User.objects.create_user(username="owner")
|
||||
requester = User.objects.create_user(username="requester")
|
||||
requester.user_permissions.add(
|
||||
Permission.objects.get(codename="delete_document"),
|
||||
)
|
||||
rest_api_client.force_authenticate(user=requester)
|
||||
doc = DocumentFactory(owner=owner)
|
||||
assign_perm("view_document", requester, doc) # view only, NOT delete
|
||||
@@ -435,6 +468,9 @@ class TestTrashRestorePermissionBoundary:
|
||||
):
|
||||
owner = User.objects.create_user(username="owner")
|
||||
requester = User.objects.create_user(username="requester")
|
||||
requester.user_permissions.add(
|
||||
Permission.objects.get(codename="delete_document"),
|
||||
)
|
||||
rest_api_client.force_authenticate(user=requester)
|
||||
doc = DocumentFactory(owner=owner)
|
||||
assign_perm("delete_document", requester, doc)
|
||||
@@ -447,6 +483,22 @@ class TestTrashRestorePermissionBoundary:
|
||||
)
|
||||
assert response.status_code == HTTPStatus.OK
|
||||
|
||||
def test_restore_requires_global_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.FORBIDDEN
|
||||
|
||||
|
||||
@pytest.mark.django_db
|
||||
class TestTrashViewExcludesExplicitlyGrantedDocuments:
|
||||
@@ -463,6 +515,9 @@ class TestTrashViewExcludesExplicitlyGrantedDocuments:
|
||||
def test_explicit_grant_does_not_leak_trashed_document(self, rest_api_client):
|
||||
owner = User.objects.create_user(username="trash_owner")
|
||||
grantee = User.objects.create_user(username="trash_grantee")
|
||||
grantee.user_permissions.add(
|
||||
Permission.objects.get(codename="view_document"),
|
||||
)
|
||||
doc = DocumentFactory(owner=owner)
|
||||
doc.delete() # soft delete
|
||||
assign_perm("view_document", grantee, doc)
|
||||
|
||||
@@ -6,8 +6,10 @@ from pathlib import Path
|
||||
from unittest import mock
|
||||
|
||||
from django.conf import settings
|
||||
from django.contrib.auth.models import Permission
|
||||
from django.contrib.auth.models import User
|
||||
from django.utils import timezone
|
||||
from guardian.shortcuts import assign_perm
|
||||
from rest_framework import serializers
|
||||
from rest_framework import status
|
||||
from rest_framework.test import APITestCase
|
||||
@@ -48,6 +50,37 @@ class ShareLinkBundleAPITests(DirectoriesMixin, APITestCase):
|
||||
delay_mock.assert_called_once()
|
||||
self.assertEqual(delay_mock.call_args.kwargs["kwargs"]["bundle_id"], bundle.pk)
|
||||
|
||||
@mock.patch("documents.views.build_share_link_bundle.apply_async")
|
||||
def test_create_bundle_requires_global_document_view_permission(
|
||||
self,
|
||||
delay_mock,
|
||||
) -> None:
|
||||
owner = User.objects.create_user(username="document_owner")
|
||||
requester = User.objects.create_user(username="bundle_creator")
|
||||
requester.user_permissions.add(
|
||||
Permission.objects.get(codename="add_sharelinkbundle"),
|
||||
)
|
||||
document = DocumentFactory.create(owner=owner)
|
||||
assign_perm("view_document", requester, document)
|
||||
self.client.force_authenticate(requester)
|
||||
payload = {
|
||||
"document_ids": [document.pk],
|
||||
"file_version": ShareLink.FileVersion.ARCHIVE,
|
||||
"expiration_days": 7,
|
||||
}
|
||||
|
||||
response = self.client.post(self.ENDPOINT, payload, format="json")
|
||||
self.assertEqual(response.status_code, status.HTTP_403_FORBIDDEN)
|
||||
|
||||
requester.user_permissions.add(
|
||||
Permission.objects.get(codename="view_document"),
|
||||
)
|
||||
requester = User.objects.get(pk=requester.pk)
|
||||
self.client.force_authenticate(requester)
|
||||
response = self.client.post(self.ENDPOINT, payload, format="json")
|
||||
self.assertEqual(response.status_code, status.HTTP_201_CREATED)
|
||||
delay_mock.assert_called_once()
|
||||
|
||||
def test_create_bundle_rejects_missing_documents(self) -> None:
|
||||
payload = {
|
||||
"document_ids": [9999],
|
||||
|
||||
@@ -141,6 +141,9 @@ class TestViews(DirectoriesMixin, TestCase):
|
||||
codename__contains="sharelink",
|
||||
)
|
||||
self.user.user_permissions.add(*sharelink_permissions)
|
||||
self.user.user_permissions.add(
|
||||
Permission.objects.get(codename="view_document"),
|
||||
)
|
||||
self.user.save()
|
||||
|
||||
self.client.force_login(self.user)
|
||||
@@ -202,6 +205,9 @@ class TestViews(DirectoriesMixin, TestCase):
|
||||
codename__contains="sharelink",
|
||||
)
|
||||
self.user.user_permissions.add(*sharelink_permissions)
|
||||
self.user.user_permissions.add(
|
||||
Permission.objects.get(codename="view_document"),
|
||||
)
|
||||
self.client.force_login(self.user)
|
||||
|
||||
create_response = self.client.post(
|
||||
|
||||
Reference in New Issue
Block a user