diff --git a/src/documents/serialisers.py b/src/documents/serialisers.py index a04a8c655..c2f57581c 100644 --- a/src/documents/serialisers.py +++ b/src/documents/serialisers.py @@ -1655,11 +1655,112 @@ class DocumentListSerializer(serializers.Serializer): return documents +class SourceModeValidationMixin: + def validate_source_mode(self, source_mode: str) -> str: + if source_mode not in bulk_edit.SourceModeChoices.__dict__.values(): + raise serializers.ValidationError("Invalid source_mode") + return source_mode + + +class RotateDocumentsSerializer(DocumentListSerializer, SourceModeValidationMixin): + degrees = serializers.IntegerField(required=True) + source_mode = serializers.CharField( + required=False, + default=bulk_edit.SourceModeChoices.LATEST_VERSION, + ) + + +class MergeDocumentsSerializer(DocumentListSerializer, SourceModeValidationMixin): + metadata_document_id = serializers.IntegerField( + required=False, + allow_null=True, + ) + delete_originals = serializers.BooleanField(required=False, default=False) + archive_fallback = serializers.BooleanField(required=False, default=False) + source_mode = serializers.CharField( + required=False, + default=bulk_edit.SourceModeChoices.LATEST_VERSION, + ) + + +class EditPdfDocumentsSerializer(DocumentListSerializer, SourceModeValidationMixin): + operations = serializers.ListField(required=True) + delete_original = serializers.BooleanField(required=False, default=False) + update_document = serializers.BooleanField(required=False, default=False) + include_metadata = serializers.BooleanField(required=False, default=True) + source_mode = serializers.CharField( + required=False, + default=bulk_edit.SourceModeChoices.LATEST_VERSION, + ) + + def validate(self, attrs): + documents = attrs["documents"] + if len(documents) > 1: + raise serializers.ValidationError( + "Edit PDF method only supports one document", + ) + + operations = attrs["operations"] + if not isinstance(operations, list): + raise serializers.ValidationError("operations must be a list") + + for op in operations: + if not isinstance(op, dict): + raise serializers.ValidationError("invalid operation entry") + if "page" not in op or not isinstance(op["page"], int): + raise serializers.ValidationError("page must be an integer") + if "rotate" in op and not isinstance(op["rotate"], int): + raise serializers.ValidationError("rotate must be an integer") + if "doc" in op and not isinstance(op["doc"], int): + raise serializers.ValidationError("doc must be an integer") + + if attrs["update_document"]: + max_idx = max(op.get("doc", 0) for op in operations) + if max_idx > 0: + raise serializers.ValidationError( + "update_document only allowed with a single output document", + ) + + doc = Document.objects.get(id=documents[0]) + if doc.page_count: + for op in operations: + if op["page"] < 1 or op["page"] > doc.page_count: + raise serializers.ValidationError( + f"Page {op['page']} is out of bounds for document with {doc.page_count} pages.", + ) + return attrs + + +class RemovePasswordDocumentsSerializer( + DocumentListSerializer, + SourceModeValidationMixin, +): + password = serializers.CharField(required=True) + update_document = serializers.BooleanField(required=False, default=False) + delete_original = serializers.BooleanField(required=False, default=False) + include_metadata = serializers.BooleanField(required=False, default=True) + source_mode = serializers.CharField( + required=False, + default=bulk_edit.SourceModeChoices.LATEST_VERSION, + ) + + class BulkEditSerializer( SerializerWithPerms, DocumentListSerializer, SetPermissionsMixin, + SourceModeValidationMixin, ): + LEGACY_FILE_METHOD_ENDPOINTS = { + "rotate": "/api/documents/rotate/", + "merge": "/api/documents/merge/", + "edit_pdf": "/api/documents/edit_pdf/", + "remove_password": "/api/documents/remove_password/", + "split": "/api/documents/edit_pdf/", + "delete_pages": "/api/documents/edit_pdf/", + } + LEGACY_FILE_METHODS = tuple(LEGACY_FILE_METHOD_ENDPOINTS.keys()) + method = serializers.ChoiceField( choices=[ "set_correspondent", @@ -1672,12 +1773,7 @@ class BulkEditSerializer( "delete", "reprocess", "set_permissions", - "rotate", - "merge", - "split", - "delete_pages", - "edit_pdf", - "remove_password", + *LEGACY_FILE_METHODS, ], label="Method", write_only=True, @@ -1755,8 +1851,7 @@ class BulkEditSerializer( return bulk_edit.edit_pdf elif method == "remove_password": return bulk_edit.remove_password - else: # pragma: no cover - # This will never happen as it is handled by the ChoiceField + else: raise serializers.ValidationError("Unsupported method.") def _validate_parameters_tags(self, parameters) -> None: @@ -1866,9 +1961,7 @@ class BulkEditSerializer( "source_mode", bulk_edit.SourceModeChoices.LATEST_VERSION, ) - if source_mode not in bulk_edit.SourceModeChoices.__dict__.values(): - raise serializers.ValidationError("Invalid source_mode") - parameters["source_mode"] = source_mode + parameters["source_mode"] = self.validate_source_mode(source_mode) def _validate_parameters_split(self, parameters) -> None: if "pages" not in parameters: diff --git a/src/documents/views.py b/src/documents/views.py index 9b960a8b4..76a2269f0 100644 --- a/src/documents/views.py +++ b/src/documents/views.py @@ -181,9 +181,13 @@ from documents.serialisers import DocumentSerializer from documents.serialisers import DocumentTypeSerializer from documents.serialisers import DocumentVersionLabelSerializer from documents.serialisers import DocumentVersionSerializer +from documents.serialisers import EditPdfDocumentsSerializer from documents.serialisers import EmailSerializer +from documents.serialisers import MergeDocumentsSerializer from documents.serialisers import NotesSerializer from documents.serialisers import PostDocumentSerializer +from documents.serialisers import RemovePasswordDocumentsSerializer +from documents.serialisers import RotateDocumentsSerializer from documents.serialisers import RunTaskViewSerializer from documents.serialisers import SavedViewSerializer from documents.serialisers import SearchResultSerializer @@ -2114,6 +2118,125 @@ class SavedViewViewSet(BulkPermissionMixin, PassUserMixin, ModelViewSet): ordering_fields = ("name",) +class DocumentOperationPermissionMixin(PassUserMixin): + permission_classes = (IsAuthenticated,) + parser_classes = (parsers.JSONParser,) + METHOD_NAMES_REQUIRING_USER = { + "split", + "merge", + "rotate", + "delete_pages", + "edit_pdf", + "remove_password", + } + + def _has_document_permissions( + self, + *, + user: User, + documents: list[int], + method, + parameters: dict[str, Any], + ) -> bool: + if user.is_superuser: + return True + + document_objs = Document.objects.select_related("owner").filter( + pk__in=documents, + ) + user_is_owner_of_all_documents = all( + (doc.owner == user or doc.owner is None) for doc in document_objs + ) + + # 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 + ) + + # check ownership for methods that change original document + if ( + ( + has_perms + and method + in [ + bulk_edit.set_permissions, + bulk_edit.delete, + bulk_edit.rotate, + bulk_edit.delete_pages, + bulk_edit.edit_pdf, + bulk_edit.remove_password, + ] + ) + or ( + method in [bulk_edit.merge, bulk_edit.split] + and parameters.get("delete_originals") + ) + or (method == bulk_edit.edit_pdf and parameters.get("update_document")) + ): + has_perms = user_is_owner_of_all_documents + + # check global add permissions for methods that create documents + if ( + has_perms + and ( + method in [bulk_edit.split, bulk_edit.merge] + or ( + method in [bulk_edit.edit_pdf, bulk_edit.remove_password] + and not parameters.get("update_document") + ) + ) + and not user.has_perm("documents.add_document") + ): + has_perms = False + + # check global delete permissions for methods that delete documents + if ( + has_perms + and ( + method == bulk_edit.delete + or ( + method in [bulk_edit.merge, bulk_edit.split] + and parameters.get("delete_originals") + ) + ) + and not user.has_perm("documents.delete_document") + ): + has_perms = False + + return has_perms + + def _execute_document_action( + self, + *, + method, + validated_data: dict[str, Any], + operation_label: str, + ): + documents = validated_data["documents"] + parameters = {k: v for k, v in validated_data.items() if k != "documents"} + user = self.request.user + + if method.__name__ in self.METHOD_NAMES_REQUIRING_USER: + parameters["user"] = user + + if not self._has_document_permissions( + user=user, + documents=documents, + method=method, + parameters=parameters, + ): + return HttpResponseForbidden("Insufficient permissions") + + try: + result = method(documents, **parameters) + return Response({"result": result}) + except Exception as e: + logger.warning(f"An error occurred performing {operation_label}: {e!s}") + return HttpResponseBadRequest( + f"Error performing {operation_label}, check logs for more detail.", + ) + + @extend_schema_view( post=extend_schema( operation_id="bulk_edit", @@ -2132,7 +2255,7 @@ class SavedViewViewSet(BulkPermissionMixin, PassUserMixin, ModelViewSet): }, ), ) -class BulkEditView(PassUserMixin): +class BulkEditView(DocumentOperationPermissionMixin): MODIFIED_FIELD_BY_METHOD = { "set_correspondent": "correspondent", "set_document_type": "document_type", @@ -2154,11 +2277,21 @@ class BulkEditView(PassUserMixin): "remove_password": None, } - permission_classes = (IsAuthenticated,) serializer_class = BulkEditSerializer - parser_classes = (parsers.JSONParser,) def post(self, request, *args, **kwargs): + request_method = request.data.get("method") + api_version = int(request.version or settings.REST_FRAMEWORK["DEFAULT_VERSION"]) + if request_method in BulkEditSerializer.LEGACY_FILE_METHODS: + endpoint = BulkEditSerializer.LEGACY_FILE_METHOD_ENDPOINTS[request_method] + logger.warning( + "Deprecated bulk_edit method '%s' requested on API version %s. " + "Use '%s' instead.", + request_method, + api_version, + endpoint, + ) + serializer = self.get_serializer(data=request.data) serializer.is_valid(raise_exception=True) @@ -2166,82 +2299,15 @@ class BulkEditView(PassUserMixin): method = serializer.validated_data.get("method") parameters = serializer.validated_data.get("parameters") documents = serializer.validated_data.get("documents") - if method in [ - bulk_edit.split, - bulk_edit.merge, - bulk_edit.rotate, - bulk_edit.delete_pages, - bulk_edit.edit_pdf, - bulk_edit.remove_password, - ]: + if method.__name__ in self.METHOD_NAMES_REQUIRING_USER: parameters["user"] = user - - if not user.is_superuser: - document_objs = Document.objects.select_related("owner").filter( - pk__in=documents, - ) - user_is_owner_of_all_documents = all( - (doc.owner == user or doc.owner is None) for doc in document_objs - ) - - # 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 - ) - - # check ownership for methods that change original document - if ( - ( - has_perms - and method - in [ - bulk_edit.set_permissions, - bulk_edit.delete, - bulk_edit.rotate, - bulk_edit.delete_pages, - bulk_edit.edit_pdf, - bulk_edit.remove_password, - ] - ) - or ( - method in [bulk_edit.merge, bulk_edit.split] - and parameters["delete_originals"] - ) - or (method == bulk_edit.edit_pdf and parameters["update_document"]) - ): - has_perms = user_is_owner_of_all_documents - - # check global add permissions for methods that create documents - if ( - has_perms - and ( - method in [bulk_edit.split, bulk_edit.merge] - or ( - method in [bulk_edit.edit_pdf, bulk_edit.remove_password] - and not parameters["update_document"] - ) - ) - and not user.has_perm("documents.add_document") - ): - has_perms = False - - # check global delete permissions for methods that delete documents - if ( - has_perms - and ( - method == bulk_edit.delete - or ( - method in [bulk_edit.merge, bulk_edit.split] - and parameters["delete_originals"] - ) - ) - and not user.has_perm("documents.delete_document") - ): - has_perms = False - - if not has_perms: - return HttpResponseForbidden("Insufficient permissions") + if not self._has_document_permissions( + user=user, + documents=documents, + method=method, + parameters=parameters, + ): + return HttpResponseForbidden("Insufficient permissions") try: modified_field = self.MODIFIED_FIELD_BY_METHOD.get(method.__name__, None) @@ -2298,6 +2364,114 @@ class BulkEditView(PassUserMixin): ) +@extend_schema_view( + post=extend_schema( + operation_id="documents_rotate", + description="Rotate one or more documents", + responses={ + 200: inline_serializer( + name="RotateDocumentsResult", + fields={ + "result": serializers.CharField(), + }, + ), + }, + ), +) +class RotateDocumentsView(DocumentOperationPermissionMixin): + serializer_class = RotateDocumentsSerializer + + def post(self, request, *args, **kwargs): + serializer = self.get_serializer(data=request.data) + serializer.is_valid(raise_exception=True) + return self._execute_document_action( + method=bulk_edit.rotate, + validated_data=serializer.validated_data, + operation_label="document rotate", + ) + + +@extend_schema_view( + post=extend_schema( + operation_id="documents_merge", + description="Merge selected documents into a new document", + responses={ + 200: inline_serializer( + name="MergeDocumentsResult", + fields={ + "result": serializers.CharField(), + }, + ), + }, + ), +) +class MergeDocumentsView(DocumentOperationPermissionMixin): + serializer_class = MergeDocumentsSerializer + + def post(self, request, *args, **kwargs): + serializer = self.get_serializer(data=request.data) + serializer.is_valid(raise_exception=True) + return self._execute_document_action( + method=bulk_edit.merge, + validated_data=serializer.validated_data, + operation_label="document merge", + ) + + +@extend_schema_view( + post=extend_schema( + operation_id="documents_edit_pdf", + description="Perform PDF edit operations on a selected document", + responses={ + 200: inline_serializer( + name="EditPdfDocumentsResult", + fields={ + "result": serializers.CharField(), + }, + ), + }, + ), +) +class EditPdfDocumentsView(DocumentOperationPermissionMixin): + serializer_class = EditPdfDocumentsSerializer + + def post(self, request, *args, **kwargs): + serializer = self.get_serializer(data=request.data) + serializer.is_valid(raise_exception=True) + return self._execute_document_action( + method=bulk_edit.edit_pdf, + validated_data=serializer.validated_data, + operation_label="PDF edit", + ) + + +@extend_schema_view( + post=extend_schema( + operation_id="documents_remove_password", + description="Remove password protection from selected PDFs", + responses={ + 200: inline_serializer( + name="RemovePasswordDocumentsResult", + fields={ + "result": serializers.CharField(), + }, + ), + }, + ), +) +class RemovePasswordDocumentsView(DocumentOperationPermissionMixin): + serializer_class = RemovePasswordDocumentsSerializer + + def post(self, request, *args, **kwargs): + serializer = self.get_serializer(data=request.data) + serializer.is_valid(raise_exception=True) + return self._execute_document_action( + method=bulk_edit.remove_password, + validated_data=serializer.validated_data, + operation_label="password removal", + ) + + @extend_schema_view( post=extend_schema( description="Upload a document via the API",