From 683f2ac250d4db33f829fe6bc6d40fa8c4e89a93 Mon Sep 17 00:00:00 2001 From: Trenton H <797416+stumpylog@users.noreply.github.com> Date: Tue, 15 Sep 2026 12:23:37 -0700 Subject: [PATCH] Fix: validate legacy bulk_edit owner/rotate/split parameters (#14120) BulkEditSerializer's hand-parsed validators only caught the exceptions their happy paths raised, so wrong-typed input produced a 500 or was passed through to the task: - owner: a nonexistent or wrong-typed id raised an uncaught error, and a boolean was accepted (Django coerces True to pk 1). It is now validated with PrimaryKeyRelatedField and the validated pk is passed on. - rotate: null raised TypeError, while true, "90" and 45 were accepted and failed later in QPDF. Degrees are now an IntegerField plus a multiple-of-90 check, passing an int on. The dedicated rotate endpoint and edit_pdf operations share the same multiple-of-90 check. - split: null raised AttributeError, "0" silently became the last page, "3-1" gave an empty group that crashed the task, and a range like "1-5000000" was expanded into a list during the request with no upper bound. Each range is now checked against 1 <= start <= end <= page_count before it is built. --- src/documents/serialisers.py | 79 ++++---- src/documents/tests/test_api_bulk_edit.py | 211 ++++++++++++++++++++++ 2 files changed, 254 insertions(+), 36 deletions(-) diff --git a/src/documents/serialisers.py b/src/documents/serialisers.py index 908ab66e9..10296a392 100644 --- a/src/documents/serialisers.py +++ b/src/documents/serialisers.py @@ -1669,6 +1669,13 @@ class SourceModeValidationMixin: return source_mode +def _validate_rotation_degrees(degrees: int, field: str = "degrees") -> int: + # QPDF refuses any other angle, which would otherwise fail inside the task + if degrees % 90 != 0: + raise serializers.ValidationError(f"{field} must be a multiple of 90") + return degrees + + class RotateDocumentsSerializer(DocumentSelectionSerializer, SourceModeValidationMixin): degrees = serializers.IntegerField(required=True) source_mode = serializers.CharField( @@ -1677,6 +1684,9 @@ class RotateDocumentsSerializer(DocumentSelectionSerializer, SourceModeValidatio ) from_webui = serializers.BooleanField(required=False, default=False) + def validate_degrees(self, value: int) -> int: + return _validate_rotation_degrees(value) + class MergeDocumentsSerializer(DocumentListSerializer, SourceModeValidationMixin): metadata_document_id = serializers.IntegerField( @@ -1744,9 +1754,7 @@ class PdfEditOperationSerializer(serializers.Serializer[dict[str, int]]): doc = serializers.IntegerField(required=False, min_value=0) def validate_rotate(self, value: int) -> int: - if value % 90 != 0: - raise serializers.ValidationError("rotate must be a multiple of 90") - return value + return _validate_rotation_degrees(value, field="rotate") class EditPdfDocumentsSerializer(DocumentListSerializer, SourceModeValidationMixin): @@ -2031,11 +2039,14 @@ class BulkEditSerializer( else: raise serializers.ValidationError("remove_custom_fields not specified") - def _validate_owner(self, owner): - ownerUser = User.objects.get(pk=owner) - if ownerUser is None: - raise serializers.ValidationError("Specified owner cannot be found") - return ownerUser + def _validate_owner(self, owner) -> User: + owner_field = serializers.PrimaryKeyRelatedField(queryset=User.objects.all()) + try: + return owner_field.run_validation(owner) + except serializers.ValidationError as e: + raise serializers.ValidationError( + "Specified owner cannot be found", + ) from e def _validate_parameters_set_permissions(self, parameters) -> None: if "set_permissions" not in parameters: @@ -2049,19 +2060,18 @@ class BulkEditSerializer( set_permissions, ) if "owner" in parameters and parameters["owner"] is not None: - self._validate_owner(parameters["owner"]) + parameters["owner"] = self._validate_owner(parameters["owner"]).pk if "merge" not in parameters: parameters["merge"] = False def _validate_parameters_rotate(self, parameters) -> None: - try: - if ( - "degrees" not in parameters - or not float(parameters["degrees"]).is_integer() - ): - raise serializers.ValidationError("invalid rotation degrees") - except ValueError: + if "degrees" not in parameters: raise serializers.ValidationError("invalid rotation degrees") + try: + degrees = serializers.IntegerField().run_validation(parameters["degrees"]) + except serializers.ValidationError as e: + raise serializers.ValidationError("invalid rotation degrees") from e + parameters["degrees"] = _validate_rotation_degrees(degrees) def _validate_source_mode(self, parameters) -> None: source_mode = parameters.get( @@ -2070,28 +2080,25 @@ class BulkEditSerializer( ) parameters["source_mode"] = self.validate_source_mode(source_mode) - def _validate_parameters_split(self, parameters) -> None: + def _validate_parameters_split(self, parameters, document_id) -> None: if "pages" not in parameters: raise serializers.ValidationError("pages not specified") - try: - pages = [] - docs = parameters["pages"].split(",") - for doc in docs: - if "-" in doc: - pages.append( - [ - x - for x in range( - int(doc.split("-")[0]), - int(doc.split("-")[1]) + 1, - ) - ], - ) - else: - pages.append([int(doc)]) - parameters["pages"] = pages - except ValueError: + if not isinstance(parameters["pages"], str): raise serializers.ValidationError("invalid pages specified") + page_count = Document.objects.get(id=document_id).page_count + pages = [] + for group in parameters["pages"].split(","): + start, is_range, end = group.partition("-") + try: + first = int(start) + last = int(end) if is_range else first + except ValueError as e: + raise serializers.ValidationError("invalid pages specified") from e + # Bound the range before building it, a huge one would exhaust memory + if not 1 <= first <= last or (page_count and last > page_count): + raise serializers.ValidationError("invalid pages specified") + pages.append(list(range(first, last + 1))) + parameters["pages"] = pages if "delete_originals" in parameters: if not isinstance(parameters["delete_originals"], bool): @@ -2218,7 +2225,7 @@ class BulkEditSerializer( raise serializers.ValidationError( "Split method only supports one document", ) - self._validate_parameters_split(parameters) + self._validate_parameters_split(parameters, attrs["documents"][0]) elif method == bulk_edit.delete_pages: if len(attrs["documents"]) > 1: raise serializers.ValidationError( diff --git a/src/documents/tests/test_api_bulk_edit.py b/src/documents/tests/test_api_bulk_edit.py index dafbf4365..0c13cce40 100644 --- a/src/documents/tests/test_api_bulk_edit.py +++ b/src/documents/tests/test_api_bulk_edit.py @@ -1200,6 +1200,48 @@ class TestBulkEditAPI(DirectoriesMixin, APITestCase): self.assertIn(expected_message, response.content) m.assert_not_called() + @mock.patch("documents.serialisers.bulk_edit.set_permissions") + def test_set_permissions_rejects_invalid_owner(self, m) -> None: + """ + GIVEN: + - A set_permissions bulk edit with an owner that is a nonexistent + id, a boolean, a list, a dict, or a non-numeric string + WHEN: + - API to bulk edit is called + THEN: + - API returns HTTP 400 + - set_permissions is not called + """ + self.setup_mock(m, "set_permissions") + + for bad_owner in ( + 999999, + True, + ["not", "an", "id"], + {"nested": "dict"}, + "not-a-number", + ): + with self.subTest(owner=bad_owner): + response = self.client.post( + "/api/documents/bulk_edit/", + json.dumps( + { + "documents": [self.doc2.id], + "method": "set_permissions", + "parameters": { + "set_permissions": { + "view": {"users": [self.user.id]}, + }, + "owner": bad_owner, + }, + }, + ), + content_type="application/json", + ) + self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST) + self.assertIn(b"Specified owner cannot be found", response.content) + m.assert_not_called() + @mock.patch("documents.serialisers.bulk_edit.set_permissions") def test_set_permissions_null_is_a_noop(self, m) -> None: """ @@ -1236,6 +1278,37 @@ class TestBulkEditAPI(DirectoriesMixin, APITestCase): {"view": {}, "change": {}}, ) + @mock.patch("documents.serialisers.bulk_edit.set_permissions") + def test_set_permissions_passes_validated_owner_id(self, m) -> None: + """ + GIVEN: + - A set_permissions bulk edit with the owner id given as a string + WHEN: + - API to bulk edit is called + THEN: + - set_permissions receives the owner as an integer id + """ + self.setup_mock(m, "set_permissions") + + response = self.client.post( + "/api/documents/bulk_edit/", + json.dumps( + { + "documents": [self.doc2.id], + "method": "set_permissions", + "parameters": { + "set_permissions": {"view": {"users": [self.user.id]}}, + "owner": str(self.user.id), + }, + }, + ), + content_type="application/json", + ) + + self.assertEqual(response.status_code, status.HTTP_200_OK) + m.assert_called_once() + self.assertEqual(m.call_args.kwargs["owner"], self.user.id) + @mock.patch("documents.serialisers.bulk_edit.set_permissions") def test_set_permissions_merge(self, m) -> None: self.setup_mock(m, "set_permissions") @@ -1507,8 +1580,146 @@ class TestBulkEditAPI(DirectoriesMixin, APITestCase): content_type="application/json", ) self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST) + + response = self.client.post( + "/api/documents/rotate/", + json.dumps( + { + "documents": [self.doc2.id, self.doc3.id], + "degrees": 45, + }, + ), + content_type="application/json", + ) + self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST) + self.assertIn(b"degrees must be a multiple of 90", response.content) m.assert_not_called() + @mock.patch("documents.serialisers.bulk_edit.rotate") + def test_bulk_edit_rotate_rejects_invalid_degrees(self, m) -> None: + """ + GIVEN: + - A legacy rotate bulk edit with degrees that are null, not an + integer, or not a multiple of 90 + WHEN: + - API to bulk edit is called + THEN: + - API returns HTTP 400 + - rotate is not called + """ + self.setup_mock(m, "rotate") + + for degrees, expected_message in ( + (None, b"invalid rotation degrees"), + (True, b"invalid rotation degrees"), + ("foo", b"invalid rotation degrees"), + (90.5, b"invalid rotation degrees"), + (45, b"degrees must be a multiple of 90"), + ): + with self.subTest(degrees=degrees): + response = self.client.post( + "/api/documents/bulk_edit/", + json.dumps( + { + "documents": [self.doc2.id], + "method": "rotate", + "parameters": {"degrees": degrees}, + }, + ), + content_type="application/json", + ) + self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST) + self.assertIn(expected_message, response.content) + m.assert_not_called() + + @mock.patch("documents.serialisers.bulk_edit.rotate") + def test_bulk_edit_rotate_passes_integer_degrees(self, m) -> None: + """ + GIVEN: + - A legacy rotate bulk edit with degrees given as a string + WHEN: + - API to bulk edit is called + THEN: + - rotate receives the degrees as an integer + """ + self.setup_mock(m, "rotate") + + response = self.client.post( + "/api/documents/bulk_edit/", + json.dumps( + { + "documents": [self.doc2.id], + "method": "rotate", + "parameters": {"degrees": "-90"}, + }, + ), + content_type="application/json", + ) + + self.assertEqual(response.status_code, status.HTTP_200_OK) + m.assert_called_once() + self.assertEqual(m.call_args.kwargs["degrees"], -90) + + @mock.patch("documents.serialisers.bulk_edit.split") + def test_bulk_edit_split_rejects_invalid_pages(self, m) -> None: + """ + GIVEN: + - A legacy split bulk edit of a 5 page document with pages that + are not a string, not numeric, zero, a reversed range, or past + the last page + WHEN: + - API to bulk edit is called + THEN: + - API returns HTTP 400 + - split is not called + """ + self.setup_mock(m, "split") + + for pages in (None, "", "a", "1-2-3", "0", "3-1", "1-6", "1-5000000"): + with self.subTest(pages=pages): + response = self.client.post( + "/api/documents/bulk_edit/", + json.dumps( + { + "documents": [self.doc2.id], + "method": "split", + "parameters": {"pages": pages}, + }, + ), + content_type="application/json", + ) + self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST) + self.assertIn(b"invalid pages specified", response.content) + m.assert_not_called() + + @mock.patch("documents.serialisers.bulk_edit.split") + def test_bulk_edit_split_parses_pages(self, m) -> None: + """ + GIVEN: + - A legacy split bulk edit with single pages and ranges + WHEN: + - API to bulk edit is called + THEN: + - split receives one page list per comma separated group + """ + self.setup_mock(m, "split") + + response = self.client.post( + "/api/documents/bulk_edit/", + json.dumps( + { + "documents": [self.doc2.id], + "method": "split", + "parameters": {"pages": "1,2-4,5"}, + }, + ), + content_type="application/json", + ) + + self.assertEqual(response.status_code, status.HTTP_200_OK) + m.assert_called_once() + self.assertEqual(m.call_args.kwargs["pages"], [[1], [2, 3, 4], [5]]) + @mock.patch("documents.views.bulk_edit.rotate") def test_rotate_insufficient_permissions(self, m) -> None: self.doc1.owner = User.objects.get(username="temp_admin")