import base64 import json from unittest import mock from allauth.mfa.models import Authenticator from allauth.mfa.totp.internal import auth as totp_auth from django.contrib.auth.models import Group from django.contrib.auth.models import User from guardian.shortcuts import get_perms from guardian.shortcuts import get_users_with_perms from rest_framework import status from rest_framework.test import APITestCase from documents.models import Correspondent from documents.models import Document from documents.models import DocumentType from documents.models import MatchingModel from documents.models import StoragePath from documents.models import Tag from paperless_testing.dirs import DirectoriesMixin from paperless_testing.factories import UserFactory from paperless_testing.permissions import grant_all_global from paperless_testing.permissions import grant_global from paperless_testing.permissions import grant_object class TestApiAuth(DirectoriesMixin, APITestCase): def test_auth_required(self) -> None: d = Document.objects.create(title="Test") self.assertEqual( self.client.get("/api/documents/").status_code, status.HTTP_401_UNAUTHORIZED, ) self.assertEqual( self.client.get(f"/api/documents/{d.id}/").status_code, status.HTTP_401_UNAUTHORIZED, ) self.assertEqual( self.client.get(f"/api/documents/{d.id}/download/").status_code, status.HTTP_401_UNAUTHORIZED, ) self.assertEqual( self.client.get(f"/api/documents/{d.id}/preview/").status_code, status.HTTP_401_UNAUTHORIZED, ) self.assertEqual( self.client.get(f"/api/documents/{d.id}/thumb/").status_code, status.HTTP_401_UNAUTHORIZED, ) self.assertEqual( self.client.get("/api/tags/").status_code, status.HTTP_401_UNAUTHORIZED, ) self.assertEqual( self.client.get("/api/correspondents/").status_code, status.HTTP_401_UNAUTHORIZED, ) self.assertEqual( self.client.get("/api/document_types/").status_code, status.HTTP_401_UNAUTHORIZED, ) self.assertEqual( self.client.get("/api/logs/").status_code, status.HTTP_401_UNAUTHORIZED, ) self.assertEqual( self.client.get("/api/saved_views/").status_code, status.HTTP_401_UNAUTHORIZED, ) self.assertEqual( self.client.get("/api/search/autocomplete/").status_code, status.HTTP_401_UNAUTHORIZED, ) self.assertEqual( self.client.get("/api/documents/bulk_edit/").status_code, status.HTTP_401_UNAUTHORIZED, ) self.assertEqual( self.client.get("/api/documents/bulk_download/").status_code, status.HTTP_401_UNAUTHORIZED, ) self.assertEqual( self.client.get("/api/documents/selection_data/").status_code, status.HTTP_401_UNAUTHORIZED, ) def test_api_version_no_auth(self) -> None: response = self.client.get("/api/documents/") self.assertNotIn("X-Api-Version", response) self.assertNotIn("X-Version", response) def test_api_version_with_auth(self) -> None: user = UserFactory(username="test", superuser=True) self.client.force_authenticate(user) response = self.client.get("/api/documents/") self.assertIn("X-Api-Version", response) self.assertIn("X-Version", response) def test_api_insufficient_permissions(self) -> None: user = UserFactory(username="test") self.client.force_authenticate(user) Document.objects.create(title="Test") self.assertEqual( self.client.get("/api/documents/").status_code, status.HTTP_403_FORBIDDEN, ) self.assertEqual( self.client.get("/api/tags/").status_code, status.HTTP_403_FORBIDDEN, ) self.assertEqual( self.client.get("/api/correspondents/").status_code, status.HTTP_403_FORBIDDEN, ) self.assertEqual( self.client.get("/api/document_types/").status_code, status.HTTP_403_FORBIDDEN, ) self.assertEqual( self.client.get("/api/logs/").status_code, status.HTTP_403_FORBIDDEN, ) self.assertEqual( self.client.get("/api/saved_views/").status_code, status.HTTP_403_FORBIDDEN, ) self.assertEqual( self.client.get("/api/search/autocomplete/?term=test").status_code, status.HTTP_403_FORBIDDEN, ) def test_api_sufficient_permissions(self) -> None: user = UserFactory(username="test") grant_all_global(user) user.is_staff = True self.client.force_authenticate(user) Document.objects.create(title="Test") self.assertEqual( self.client.get("/api/documents/").status_code, status.HTTP_200_OK, ) self.assertEqual(self.client.get("/api/tags/").status_code, status.HTTP_200_OK) self.assertEqual( self.client.get("/api/correspondents/").status_code, status.HTTP_200_OK, ) self.assertEqual( self.client.get("/api/document_types/").status_code, status.HTTP_200_OK, ) self.assertEqual(self.client.get("/api/logs/").status_code, status.HTTP_200_OK) self.assertEqual( self.client.get("/api/saved_views/").status_code, status.HTTP_200_OK, ) def test_api_get_object_permissions(self) -> None: user1 = UserFactory(username="test1") user2 = UserFactory(username="test2") grant_global(user1, "view_document") self.client.force_authenticate(user1) self.assertEqual( self.client.get("/api/documents/").status_code, status.HTTP_200_OK, ) d = Document.objects.create(title="Test", content="the content 1", checksum="1") # no owner self.assertEqual( self.client.get(f"/api/documents/{d.id}/").status_code, status.HTTP_200_OK, ) d2 = Document.objects.create( title="Test 2", content="the content 2", checksum="2", owner=user2, ) self.assertEqual( self.client.get(f"/api/documents/{d2.id}/").status_code, status.HTTP_404_NOT_FOUND, ) def test_api_default_owner(self) -> None: """ GIVEN: - API request to create an object (Tag) WHEN: - owner is not set at all THEN: - Object created with current user as owner """ user1 = UserFactory(username="user1", superuser=True) self.client.force_authenticate(user1) response = self.client.post( "/api/tags/", json.dumps( { "name": "test1", "matching_algorithm": MatchingModel.MATCH_AUTO, }, ), content_type="application/json", ) self.assertEqual(response.status_code, status.HTTP_201_CREATED) tag1 = Tag.objects.filter(name="test1").first() self.assertEqual(tag1.owner, user1) def test_api_set_no_owner(self) -> None: """ GIVEN: - API request to create an object (Tag) WHEN: - owner is passed as None THEN: - Object created with no owner """ user1 = UserFactory(username="user1", superuser=True) self.client.force_authenticate(user1) response = self.client.post( "/api/tags/", json.dumps( { "name": "test1", "matching_algorithm": MatchingModel.MATCH_AUTO, "owner": None, }, ), content_type="application/json", ) self.assertEqual(response.status_code, status.HTTP_201_CREATED) tag1 = Tag.objects.filter(name="test1").first() self.assertEqual(tag1.owner, None) def test_api_set_owner_w_permissions(self) -> None: """ GIVEN: - API request to create an object (Tag) that supplies set_permissions object WHEN: - owner is passed as user id - view > users is set & view > groups is set THEN: - Object permissions are set appropriately """ user1 = UserFactory(username="user1", superuser=True) user2 = User.objects.create(username="user2") group1 = Group.objects.create(name="group1") self.client.force_authenticate(user1) response = self.client.post( "/api/tags/", json.dumps( { "name": "test1", "matching_algorithm": MatchingModel.MATCH_AUTO, "owner": user1.id, "set_permissions": { "view": { "users": [user2.id], "groups": [group1.id], }, "change": { "users": None, "groups": None, }, }, }, ), content_type="application/json", ) self.assertEqual(response.status_code, status.HTTP_201_CREATED) tag1 = Tag.objects.filter(name="test1").first() from guardian.core import ObjectPermissionChecker checker = ObjectPermissionChecker(user2) self.assertEqual(checker.has_perm("view_tag", tag1), True) self.assertIn("view_tag", get_perms(group1, tag1)) def test_api_set_other_owner_w_permissions(self) -> None: """ GIVEN: - API request to create an object (Tag) WHEN: - a different owner than is logged in is set - view > groups is set THEN: - Object permissions are set appropriately """ user1 = UserFactory(username="user1", superuser=True) user2 = User.objects.create(username="user2") group1 = Group.objects.create(name="group1") self.client.force_authenticate(user1) response = self.client.post( "/api/tags/", json.dumps( { "name": "test1", "matching_algorithm": MatchingModel.MATCH_AUTO, "owner": user2.id, "set_permissions": { "view": { "users": None, "groups": [group1.id], }, "change": { "users": None, "groups": None, }, }, }, ), content_type="application/json", ) self.assertEqual(response.status_code, status.HTTP_201_CREATED) tag1 = Tag.objects.filter(name="test1").first() self.assertEqual(tag1.owner, user2) self.assertIn("view_tag", get_perms(group1, tag1)) def test_api_set_doc_permissions(self) -> None: """ GIVEN: - API request to update doc permissions and owner WHEN: - owner is set - view > users is set & view > groups is set THEN: - Object permissions are set appropriately """ doc = Document.objects.create( title="test", mime_type="application/pdf", content="this is a document", ) user1 = UserFactory(username="user1", superuser=True) user2 = User.objects.create(username="user2") group1 = Group.objects.create(name="group1") self.client.force_authenticate(user1) response = self.client.patch( f"/api/documents/{doc.id}/", json.dumps( { "owner": user1.id, "set_permissions": { "view": { "users": [user2.id], "groups": [group1.id], }, "change": { "users": None, "groups": None, }, }, }, ), content_type="application/json", ) self.assertEqual(response.status_code, status.HTTP_200_OK) doc = Document.objects.get(pk=doc.id) self.assertEqual(doc.owner, user1) from guardian.core import ObjectPermissionChecker checker = ObjectPermissionChecker(user2) self.assertTrue(checker.has_perm("view_document", doc)) self.assertIn("view_document", get_perms(group1, doc)) def test_patch_doesnt_remove_permissions(self) -> None: """ GIVEN: - existing document with permissions set WHEN: - PATCH API request to update doc that is not json THEN: - Object permissions are not removed """ doc = Document.objects.create( title="test", mime_type="application/pdf", content="this is a document", ) user1 = UserFactory(username="user1", superuser=True) user2 = User.objects.create(username="user2") group1 = Group.objects.create(name="group1") doc.owner = user1 doc.save() grant_object(user2, doc, "view_document") grant_object(user2, doc, "change_document") grant_object(group1, doc, "view_document") grant_object(group1, doc, "change_document") self.client.force_authenticate(user1) response = self.client.patch( f"/api/documents/{doc.id}/", { "archive_serial_number": "123", }, ) self.assertEqual(response.status_code, status.HTTP_200_OK) doc = Document.objects.get(pk=doc.id) self.assertEqual(doc.owner, user1) from guardian.core import ObjectPermissionChecker checker = ObjectPermissionChecker(user2) self.assertTrue(checker.has_perm("view_document", doc)) self.assertIn("view_document", get_perms(group1, doc)) self.assertTrue(checker.has_perm("change_document", doc)) self.assertIn("change_document", get_perms(group1, doc)) def test_document_permissions_change_requires_owner(self) -> None: owner = UserFactory(username="owner") editor = UserFactory(username="editor") grant_all_global(editor) doc = Document.objects.create( title="Ownered doc", content="sensitive", checksum="abc123", mime_type="application/pdf", owner=owner, ) grant_object(editor, doc, "view_document") grant_object(editor, doc, "change_document") self.client.force_authenticate(editor) response = self.client.patch( f"/api/documents/{doc.pk}/", json.dumps( { "set_permissions": { "view": { "users": [editor.id], "groups": [], }, "change": { "users": None, "groups": None, }, }, }, ), content_type="application/json", ) self.assertEqual(response.status_code, status.HTTP_403_FORBIDDEN) self.client.force_authenticate(editor) response = self.client.patch( f"/api/documents/{doc.pk}/", json.dumps( { "owner": editor.id, }, ), content_type="application/json", ) self.assertEqual(response.status_code, status.HTTP_403_FORBIDDEN) def test_dynamic_permissions_fields(self) -> None: user1 = UserFactory(username="user1") grant_global(user1, "view_document") user2 = UserFactory(username="user2") Document.objects.create(title="Test", content="content 1", checksum="1") doc2 = Document.objects.create( title="Test2", content="content 2", checksum="2", owner=user2, ) doc3 = Document.objects.create( title="Test3", content="content 3", checksum="3", owner=user2, ) doc4 = Document.objects.create( title="Test4", content="content 4", checksum="4", owner=user1, ) grant_object(user1, doc2, "view_document") grant_object(user1, doc3, "view_document") grant_object(user1, doc3, "change_document") grant_object(user2, doc4, "view_document") self.client.force_authenticate(user1) response = self.client.get( "/api/documents/?ordering=-id", format="json", ) self.assertEqual(response.status_code, status.HTTP_200_OK) resp_data = response.json() # The response will contain the documents in reversed order of creation # due to #6982, but previously this code relied on implicit ordering # so let's ensure the order is as expected: results = resp_data["results"][::-1] self.assertNotIn("permissions", results[0]) self.assertIn("user_can_change", results[0]) self.assertTrue(results[0]["user_can_change"]) # doc1 self.assertFalse(results[0]["is_shared_by_requester"]) # doc1 self.assertFalse(results[1]["user_can_change"]) # doc2 self.assertTrue(results[2]["user_can_change"]) # doc3 self.assertTrue(results[3]["is_shared_by_requester"]) # doc4 response = self.client.get( "/api/documents/?full_perms=true", format="json", ) self.assertEqual(response.status_code, status.HTTP_200_OK) resp_data = response.json() # See above about response ordering results = resp_data["results"][::-1] self.assertIn("permissions", results[0]) self.assertNotIn("user_can_change", results[0]) self.assertNotIn("is_shared_by_requester", results[0]) def test_superuser_user_can_change_without_explicit_grant(self) -> None: """ A superuser has implicit change access to every document, even one owned by someone else with no explicit guardian grant -- mirrors guardian's own ObjectPermissionChecker.has_perm() superuser shortcut. """ superuser = UserFactory(username="admin", superuser=True) other_user = UserFactory(username="user2") Document.objects.create( title="Test", content="content", checksum="1", owner=other_user, ) self.client.force_authenticate(superuser) response = self.client.get("/api/documents/", format="json") self.assertEqual(response.status_code, status.HTTP_200_OK) results = response.json()["results"] self.assertEqual(len(results), 1) self.assertTrue(results[0]["user_can_change"]) @mock.patch("allauth.mfa.adapter.DefaultMFAAdapter.is_mfa_enabled") def test_basic_auth_mfa_enabled(self, mock_is_mfa_enabled) -> None: """ GIVEN: - User with MFA enabled WHEN: - API request is made with basic auth THEN: - MFA required error is returned """ user1 = UserFactory(username="user1") user1.set_password("password") user1.save() mock_is_mfa_enabled.return_value = True response = self.client.get( "/api/documents/", HTTP_AUTHORIZATION="Basic " + base64.b64encode(b"user1:password").decode(), ) self.assertEqual(response.status_code, status.HTTP_401_UNAUTHORIZED) self.assertEqual(response.data["detail"], "MFA required") @mock.patch("allauth.mfa.totp.internal.auth.TOTP.validate_code") def test_get_token_mfa_enabled(self, mock_validate_code) -> None: """ GIVEN: - User with MFA enabled WHEN: - API request is made to obtain an auth token THEN: - MFA code is required """ user1 = UserFactory(username="user1") user1.set_password("password") user1.save() response = self.client.post( "/api/token/", data={ "username": "user1", "password": "password", }, ) self.assertEqual(response.status_code, status.HTTP_200_OK) secret = totp_auth.generate_totp_secret() totp_auth.TOTP.activate( user1, secret, ) # no code response = self.client.post( "/api/token/", data={ "username": "user1", "password": "password", }, ) self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST) self.assertEqual(response.data["non_field_errors"][0], "MFA code is required") # invalid code mock_validate_code.return_value = False response = self.client.post( "/api/token/", data={ "username": "user1", "password": "password", "code": "123456", }, ) self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST) self.assertEqual(response.data["non_field_errors"][0], "Invalid MFA code") # valid code mock_validate_code.return_value = True response = self.client.post( "/api/token/", data={ "username": "user1", "password": "password", "code": "123456", }, ) self.assertEqual(response.status_code, status.HTTP_200_OK) class TestApiUser(DirectoriesMixin, APITestCase): ENDPOINT = "/api/users/" def setUp(self) -> None: super().setUp() self.user = UserFactory(username="temp_admin", superuser=True) self.client.force_authenticate(user=self.user) def test_get_users(self) -> None: """ GIVEN: - Configured users WHEN: - API call is made to get users THEN: - Configured users are provided """ user1 = User.objects.create( username="testuser", password="test", first_name="Test", last_name="User", ) response = self.client.get(self.ENDPOINT) self.assertEqual(response.status_code, status.HTTP_200_OK) self.assertEqual(response.data["count"], 2) returned_user2 = response.data["results"][1] self.assertEqual(returned_user2["username"], user1.username) self.assertEqual(returned_user2["password"], "**********") self.assertEqual(returned_user2["first_name"], user1.first_name) self.assertEqual(returned_user2["last_name"], user1.last_name) def test_create_user(self) -> None: """ WHEN: - API request is made to add a user account THEN: - A new user account is created """ user1 = { "username": "testuser", "password": "areallysupersecretpassword235", "first_name": "Test", "last_name": "User", } response = self.client.post( self.ENDPOINT, data=user1, ) self.assertEqual(response.status_code, status.HTTP_201_CREATED) returned_user1 = User.objects.get(username="testuser") self.assertEqual(returned_user1.username, user1["username"]) self.assertEqual(returned_user1.first_name, user1["first_name"]) self.assertEqual(returned_user1.last_name, user1["last_name"]) def test_delete_user(self) -> None: """ GIVEN: - Existing user account WHEN: - API request is made to delete a user account THEN: - Account is deleted """ user1 = User.objects.create( username="testuser", password="test", first_name="Test", last_name="User", ) nUsers = User.objects.count() response = self.client.delete( f"{self.ENDPOINT}{user1.pk}/", ) self.assertEqual(response.status_code, status.HTTP_204_NO_CONTENT) self.assertEqual(User.objects.count(), nUsers - 1) def test_update_user(self) -> None: """ GIVEN: - Existing user accounts WHEN: - API request is made to update user account THEN: - The user account is updated, password only updated if not '****' """ user1 = User.objects.create( username="testuser", password="test", first_name="Test", last_name="User", ) initial_password = user1.password response = self.client.patch( f"{self.ENDPOINT}{user1.pk}/", data={ "first_name": "Updated Name 1", "password": "******", }, ) self.assertEqual(response.status_code, status.HTTP_200_OK) returned_user1 = User.objects.get(pk=user1.pk) self.assertEqual(returned_user1.first_name, "Updated Name 1") self.assertEqual(returned_user1.password, initial_password) response = self.client.patch( f"{self.ENDPOINT}{user1.pk}/", data={ "first_name": "Updated Name 2", "password": "newreallystrongpassword456", }, ) self.assertEqual(response.status_code, status.HTTP_200_OK) returned_user2 = User.objects.get(pk=user1.pk) self.assertEqual(returned_user2.first_name, "Updated Name 2") self.assertNotEqual(returned_user2.password, initial_password) def test_deactivate_totp(self) -> None: """ GIVEN: - Existing user account with TOTP enabled WHEN: - API request by a superuser is made to deactivate TOTP - API request by a regular user is made to deactivate TOTP THEN: - TOTP is deactivated, if exists - Regular user is forbidden from deactivating TOTP """ user1 = User.objects.create( username="testuser", password="test", first_name="Test", last_name="User", ) Authenticator.objects.create( user=user1, type=Authenticator.Type.TOTP, data={}, ) response = self.client.post( f"{self.ENDPOINT}{user1.pk}/deactivate_totp/", ) self.assertEqual(response.status_code, status.HTTP_200_OK) self.assertEqual(Authenticator.objects.filter(user=user1).count(), 0) # fail if already deactivated response = self.client.post( f"{self.ENDPOINT}{user1.pk}/deactivate_totp/", ) self.assertEqual(response.status_code, status.HTTP_404_NOT_FOUND) regular_user = UserFactory(username="regular_user") grant_all_global(regular_user) self.client.force_authenticate(regular_user) Authenticator.objects.create( user=user1, type=Authenticator.Type.TOTP, data={}, ) response = self.client.post( f"{self.ENDPOINT}{user1.pk}/deactivate_totp/", ) self.assertEqual(response.status_code, status.HTTP_403_FORBIDDEN) def test_only_superusers_can_create_or_alter_superuser_status(self) -> None: """ GIVEN: - Existing user account WHEN: - API request is made to add a user account with superuser status - API request is made to change superuser status THEN: - Only superusers can change superuser status """ user1 = UserFactory(username="user1") grant_all_global(user1) user2 = UserFactory(username="user2", superuser=True) self.client.force_authenticate(user1) response = self.client.patch( f"{self.ENDPOINT}{user1.pk}/", json.dumps( { "is_superuser": True, }, ), content_type="application/json", ) self.assertEqual(response.status_code, status.HTTP_403_FORBIDDEN) response = self.client.post( f"{self.ENDPOINT}", json.dumps( { "username": "user3", "is_superuser": True, }, ), content_type="application/json", ) self.assertEqual(response.status_code, status.HTTP_403_FORBIDDEN) response = self.client.post( f"{self.ENDPOINT}", json.dumps( { "username": "user4", "is_superuser": "true", }, ), content_type="application/json", ) self.assertEqual(response.status_code, status.HTTP_403_FORBIDDEN) self.client.force_authenticate(user2) response = self.client.patch( f"{self.ENDPOINT}{user1.pk}/", json.dumps( { "is_superuser": True, }, ), content_type="application/json", ) self.assertEqual(response.status_code, status.HTTP_200_OK) returned_user1 = User.objects.get(pk=user1.pk) self.assertEqual(returned_user1.is_superuser, True) response = self.client.patch( f"{self.ENDPOINT}{user1.pk}/", json.dumps( { "is_superuser": False, }, ), content_type="application/json", ) self.assertEqual(response.status_code, status.HTTP_200_OK) returned_user1 = User.objects.get(pk=user1.pk) self.assertEqual(returned_user1.is_superuser, False) def test_only_superusers_can_create_or_alter_staff_status(self) -> None: """ GIVEN: - Existing user account WHEN: - API request is made to add a user account with staff status - API request is made to change staff status THEN: - Only superusers can change staff status """ user1 = UserFactory(username="user1") grant_all_global(user1) user2 = UserFactory(username="user2", superuser=True) self.client.force_authenticate(user1) response = self.client.patch( f"{self.ENDPOINT}{user1.pk}/", json.dumps( { "is_staff": "true", }, ), content_type="application/json", ) self.assertEqual(response.status_code, status.HTTP_403_FORBIDDEN) response = self.client.post( f"{self.ENDPOINT}", json.dumps( { "username": "user3", "is_staff": 1, }, ), content_type="application/json", ) self.assertEqual(response.status_code, status.HTTP_403_FORBIDDEN) self.client.force_authenticate(user2) response = self.client.patch( f"{self.ENDPOINT}{user1.pk}/", json.dumps( { "is_staff": True, }, ), content_type="application/json", ) self.assertEqual(response.status_code, status.HTTP_200_OK) returned_user1 = User.objects.get(pk=user1.pk) self.assertEqual(returned_user1.is_staff, True) class TestApiGroup(DirectoriesMixin, APITestCase): ENDPOINT = "/api/groups/" def setUp(self) -> None: super().setUp() self.user = UserFactory(username="temp_admin", superuser=True) self.client.force_authenticate(user=self.user) def test_get_groups(self) -> None: """ GIVEN: - Configured groups WHEN: - API call is made to get groups THEN: - Configured groups are provided """ group1 = Group.objects.create( name="Test Group", ) response = self.client.get(self.ENDPOINT) self.assertEqual(response.status_code, status.HTTP_200_OK) self.assertEqual(response.data["count"], 1) returned_group1 = response.data["results"][0] self.assertEqual(returned_group1["name"], group1.name) def test_create_group(self) -> None: """ WHEN: - API request is made to add a group THEN: - A new group is created """ group1 = { "name": "Test Group", } response = self.client.post( self.ENDPOINT, data=group1, ) self.assertEqual(response.status_code, status.HTTP_201_CREATED) returned_group1 = Group.objects.get(name="Test Group") self.assertEqual(returned_group1.name, group1["name"]) def test_delete_group(self) -> None: """ GIVEN: - Existing group WHEN: - API request is made to delete a group THEN: - Group is deleted """ group1 = Group.objects.create( name="Test Group", ) response = self.client.delete( f"{self.ENDPOINT}{group1.pk}/", ) self.assertEqual(response.status_code, status.HTTP_204_NO_CONTENT) self.assertEqual(len(Group.objects.all()), 0) def test_update_group(self) -> None: """ GIVEN: - Existing groups WHEN: - API request is made to update group THEN: - The group is updated """ group1 = Group.objects.create( name="Test Group", ) response = self.client.patch( f"{self.ENDPOINT}{group1.pk}/", data={ "name": "Updated Name 1", }, ) self.assertEqual(response.status_code, status.HTTP_200_OK) returned_group1 = Group.objects.get(pk=group1.pk) self.assertEqual(returned_group1.name, "Updated Name 1") class TestBulkEditObjectPermissions(APITestCase): def setUp(self) -> None: super().setUp() self.temp_admin = UserFactory(username="temp_admin", superuser=True) self.client.force_authenticate(user=self.temp_admin) self.t1 = Tag.objects.create(name="t1") self.t2 = Tag.objects.create(name="t2") self.c1 = Correspondent.objects.create(name="c1") self.dt1 = DocumentType.objects.create(name="dt1") self.sp1 = StoragePath.objects.create(name="sp1") self.user1 = User.objects.create(username="user1") self.user2 = User.objects.create(username="user2") self.user3 = User.objects.create(username="user3") def test_bulk_object_set_permissions(self) -> None: """ GIVEN: - Existing objects WHEN: - bulk_edit_objects API endpoint is called with set_permissions operation THEN: - Permissions and / or owner are changed """ permissions = { "view": { "users": [self.user1.id, self.user2.id], "groups": [], }, "change": { "users": [self.user1.id], "groups": [], }, } response = self.client.post( "/api/bulk_edit_objects/", json.dumps( { "objects": [self.t1.id, self.t2.id], "object_type": "tags", "operation": "set_permissions", "permissions": permissions, }, ), content_type="application/json", ) self.assertEqual(response.status_code, status.HTTP_200_OK) self.assertIn(self.user1, get_users_with_perms(self.t1)) response = self.client.post( "/api/bulk_edit_objects/", json.dumps( { "objects": [self.c1.id], "object_type": "correspondents", "operation": "set_permissions", "permissions": permissions, }, ), content_type="application/json", ) self.assertEqual(response.status_code, status.HTTP_200_OK) self.assertIn(self.user1, get_users_with_perms(self.c1)) response = self.client.post( "/api/bulk_edit_objects/", json.dumps( { "objects": [self.dt1.id], "object_type": "document_types", "operation": "set_permissions", "permissions": permissions, }, ), content_type="application/json", ) self.assertEqual(response.status_code, status.HTTP_200_OK) self.assertIn(self.user1, get_users_with_perms(self.dt1)) response = self.client.post( "/api/bulk_edit_objects/", json.dumps( { "objects": [self.sp1.id], "object_type": "storage_paths", "operation": "set_permissions", "permissions": permissions, }, ), content_type="application/json", ) self.assertEqual(response.status_code, status.HTTP_200_OK) self.assertIn(self.user1, get_users_with_perms(self.sp1)) response = self.client.post( "/api/bulk_edit_objects/", json.dumps( { "objects": [self.t1.id, self.t2.id], "object_type": "tags", "operation": "set_permissions", "owner": self.user3.id, }, ), content_type="application/json", ) self.assertEqual(response.status_code, status.HTTP_200_OK) self.assertEqual(Tag.objects.get(pk=self.t2.id).owner, self.user3) response = self.client.post( "/api/bulk_edit_objects/", json.dumps( { "objects": [self.sp1.id], "object_type": "storage_paths", "operation": "set_permissions", "owner": self.user3.id, }, ), content_type="application/json", ) self.assertEqual(response.status_code, status.HTTP_200_OK) self.assertEqual(StoragePath.objects.get(pk=self.sp1.id).owner, self.user3) def test_bulk_object_set_permissions_merge(self) -> None: """ GIVEN: - Existing objects WHEN: - bulk_edit_objects API endpoint is called with set_permissions operation with merge=True or merge=False (default) THEN: - Permissions and / or owner are replaced or merged, depending on the merge flag """ permissions = { "view": { "users": [self.user1.id, self.user2.id], "groups": [], }, "change": { "users": [self.user1.id], "groups": [], }, } grant_object(self.user3, self.t1, "view_tag") self.t1.owner = self.user3 self.t1.save() # merge=True response = self.client.post( "/api/bulk_edit_objects/", json.dumps( { "objects": [self.t1.id, self.t2.id], "object_type": "tags", "owner": self.user1.id, "permissions": permissions, "operation": "set_permissions", "merge": True, }, ), content_type="application/json", ) self.t1.refresh_from_db() self.t2.refresh_from_db() self.assertEqual(response.status_code, status.HTTP_200_OK) # user3 should still be owner of t1 since was set prior self.assertEqual(self.t1.owner, self.user3) # user1 should now be owner of t2 since it didn't have an owner self.assertEqual(self.t2.owner, self.user1) # user1 should be added self.assertIn(self.user1, get_users_with_perms(self.t1)) # user3 should be preserved self.assertIn(self.user3, get_users_with_perms(self.t1)) # merge=False (default) response = self.client.post( "/api/bulk_edit_objects/", json.dumps( { "objects": [self.t1.id, self.t2.id], "object_type": "tags", "permissions": permissions, "operation": "set_permissions", "merge": False, }, ), content_type="application/json", ) self.assertEqual(response.status_code, status.HTTP_200_OK) # user1 should be added self.assertIn(self.user1, get_users_with_perms(self.t1)) # user3 should be removed self.assertNotIn(self.user3, get_users_with_perms(self.t1)) def test_bulk_edit_object_permissions_insufficient_perms(self) -> None: """ GIVEN: - Objects owned by user other than logged in user WHEN: - bulk_edit_objects API endpoint is called with set_permissions operation THEN: - User is not able to change permissions """ self.t1.owner = User.objects.get(username="temp_admin") self.t1.save() self.client.force_authenticate(user=self.user1) response = self.client.post( "/api/bulk_edit_objects/", json.dumps( { "objects": [self.t1.id, self.t2.id], "object_type": "tags", "operation": "set_permissions", "owner": self.user1.id, }, ), content_type="application/json", ) self.assertEqual(response.status_code, status.HTTP_403_FORBIDDEN) self.assertEqual(response.content, b"Insufficient permissions") def test_bulk_edit_object_permissions_shared_object_not_owner(self) -> None: """ GIVEN: - Object owned by another user, shared with the logged in user with change permissions WHEN: - bulk_edit_objects API endpoint is called with set_permissions operation THEN: - User is not able to take ownership or change permissions, consistent with the single object API """ self.t1.owner = self.user2 self.t1.save() grant_object(self.user1, self.t1, "view_tag") grant_object(self.user1, self.t1, "change_tag") grant_global(self.user1, "view_tag", "change_tag") user1 = User.objects.get(pk=self.user1.pk) self.client.force_authenticate(user=user1) response = self.client.post( "/api/bulk_edit_objects/", json.dumps( { "objects": [self.t1.id], "object_type": "tags", "operation": "set_permissions", "owner": user1.id, "permissions": { "view": {"users": [user1.id], "groups": []}, "change": {"users": [user1.id], "groups": []}, }, "merge": False, }, ), content_type="application/json", ) self.assertEqual(response.status_code, status.HTTP_403_FORBIDDEN) self.assertEqual(Tag.objects.get(pk=self.t1.id).owner, self.user2) # the single object endpoint refuses the same request response = self.client.patch( f"/api/tags/{self.t1.id}/", {"owner": user1.id}, format="json", ) self.assertEqual(response.status_code, status.HTTP_403_FORBIDDEN) self.assertEqual(Tag.objects.get(pk=self.t1.id).owner, self.user2) def test_bulk_edit_object_permissions_all_with_shared_objects(self) -> None: """ GIVEN: - Objects owned by the logged in user, unowned objects and objects owned by another user but shared with the logged in user WHEN: - bulk_edit_objects API endpoint is called with set_permissions operation and all = True THEN: - The request is refused and no objects are changed """ owned = Tag.objects.create(name="owned", owner=self.user1) shared = Tag.objects.create(name="shared", owner=self.user2) grant_object(self.user1, shared, "view_tag") grant_object(self.user1, shared, "change_tag") grant_global(self.user1, "view_tag", "change_tag") user1 = User.objects.get(pk=self.user1.pk) self.client.force_authenticate(user=user1) response = self.client.post( "/api/bulk_edit_objects/", json.dumps( { "objects": [], "all": True, "object_type": "tags", "operation": "set_permissions", "permissions": { "view": {"users": [self.user3.id], "groups": []}, "change": {"users": [self.user3.id], "groups": []}, }, "merge": False, }, ), content_type="application/json", ) self.assertEqual(response.status_code, status.HTTP_403_FORBIDDEN) # nothing was changed, including the objects the user does own self.assertNotIn(self.user3, get_users_with_perms(owned)) self.assertNotIn(self.user3, get_users_with_perms(self.t1)) self.assertNotIn(self.user3, get_users_with_perms(shared)) self.assertEqual(Tag.objects.get(pk=shared.pk).owner, self.user2) def test_bulk_edit_object_delete_shared_object_not_owner(self) -> None: """ GIVEN: - Object owned by another user, shared with the logged in user with change and delete permissions WHEN: - bulk_edit_objects API endpoint is called with delete operation THEN: - User is not able to delete the object, consistent with documents """ self.t1.owner = self.user2 self.t1.save() grant_object(self.user1, self.t1, "view_tag") grant_object(self.user1, self.t1, "change_tag") grant_object(self.user1, self.t1, "delete_tag") grant_global(self.user1, "view_tag", "change_tag", "delete_tag") user1 = User.objects.get(pk=self.user1.pk) self.client.force_authenticate(user=user1) response = self.client.post( "/api/bulk_edit_objects/", json.dumps( { "objects": [self.t1.id], "object_type": "tags", "operation": "delete", }, ), content_type="application/json", ) self.assertEqual(response.status_code, status.HTTP_403_FORBIDDEN) self.assertTrue(Tag.objects.filter(pk=self.t1.id).exists()) def test_bulk_object_set_permissions_rejects_empty_permissions(self) -> None: """ GIVEN: - Existing objects WHEN: - bulk_edit_objects API endpoint is called with set_permissions operation and an empty permissions dict THEN: - Validation fails rather than silently applying a no-op """ response = self.client.post( "/api/bulk_edit_objects/", json.dumps( { "objects": [self.t1.id], "object_type": "tags", "operation": "set_permissions", "permissions": {}, }, ), content_type="application/json", ) self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST) def test_bulk_object_set_permissions_rejects_non_dict_permissions(self) -> None: """ GIVEN: - Existing objects WHEN: - bulk_edit_objects API endpoint is called with set_permissions operation and a non-dict permissions value THEN: - Validation fails rather than crashing """ response = self.client.post( "/api/bulk_edit_objects/", json.dumps( { "objects": [self.t1.id], "object_type": "tags", "operation": "set_permissions", "permissions": False, }, ), content_type="application/json", ) self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST) def test_bulk_object_set_permissions_rejects_unknown_action(self) -> None: """ GIVEN: - Existing objects WHEN: - bulk_edit_objects API endpoint is called with set_permissions operation and an unrecognized permission action name THEN: - Validation fails rather than silently no-oping """ response = self.client.post( "/api/bulk_edit_objects/", json.dumps( { "objects": [self.t1.id], "object_type": "tags", "operation": "set_permissions", "permissions": {"not_a_real_action": {"users": [self.user1.id]}}, }, ), content_type="application/json", ) self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST) def test_bulk_object_set_permissions_null_users_clears_users(self) -> None: """ GIVEN: - An object a user has view permission on WHEN: - bulk_edit_objects API endpoint is called with set_permissions operation, merge off, and an explicit null for the view users THEN: - Request succeeds and null is treated as an empty user list, so the existing view permission is removed """ grant_object(self.user1, self.t1, "view_tag") response = self.client.post( "/api/bulk_edit_objects/", json.dumps( { "objects": [self.t1.id], "object_type": "tags", "operation": "set_permissions", "permissions": {"view": {"users": None}}, }, ), content_type="application/json", ) self.assertEqual(response.status_code, status.HTTP_200_OK) self.assertNotIn(self.user1, get_users_with_perms(self.t1)) def test_bulk_edit_object_permissions_validation(self) -> None: """ GIVEN: - Existing objects WHEN: - bulk_edit_objects API endpoint is called with set_permissions operation with invalid params THEN: - Validation fails """ # not a list response = self.client.post( "/api/bulk_edit_objects/", json.dumps( { "objects": self.t1.id, "object_type": "tags", "operation": "set_permissions", "owner": self.user1.id, }, ), content_type="application/json", ) self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST) # not a list of ints response = self.client.post( "/api/bulk_edit_objects/", json.dumps( { "objects": ["one"], "object_type": "tags", "owner": self.user1.id, }, ), content_type="application/json", ) self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST) # duplicates response = self.client.post( "/api/bulk_edit_objects/", json.dumps( { "objects": [self.t1.id, self.t2.id, self.t1.id], "object_type": "tags", "operation": "set_permissions", "owner": self.user1.id, }, ), content_type="application/json", ) self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST) # not a valid object type response = self.client.post( "/api/bulk_edit_objects/", json.dumps( { "objects": [1], "object_type": "madeup", "operation": "set_permissions", "owner": self.user1.id, }, ), content_type="application/json", ) self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST) class TestFullPermissionsFlag(APITestCase): def setUp(self) -> None: super().setUp() self.admin = UserFactory(username="admin", superuser=True) def test_full_perms_flag(self) -> None: """ GIVEN: - API request to list documents WHEN: - full_perms flag is set to true, 1, false, or a random value THEN: - Permissions field is included or excluded accordingly """ self.client.force_authenticate(self.admin) Document.objects.create(title="Doc", checksum="xyz", owner=self.admin) resp = self.client.get("/api/documents/?full_perms=true") self.assertIn("permissions", resp.data["results"][0]) resp = self.client.get("/api/documents/?full_perms=1") self.assertIn("permissions", resp.data["results"][0]) resp = self.client.get("/api/documents/?full_perms=false") self.assertNotIn("permissions", resp.data["results"][0]) resp = self.client.get("/api/documents/?full_perms=garbage") self.assertNotIn("permissions", resp.data["results"][0])