✨(backend) allow uploading images as attachments to a document

We only rely on S3 to store attachments for a document. Nothing
is persisted in the database as the image media urls will be
stored in the document json.

src/backend/core/api/serializers.py

@@ -1,5 +1,8 @@
 """Client serializers for the impress core app."""
 
+import mimetypes
+
+from django.conf import settings
 from django.db.models import Q
 from django.utils.translation import gettext_lazy as _
 
@@ -153,6 +156,34 @@ class Meta:
         read_only_fields = ["id", "accesses", "abilities", "created_at", "updated_at"]
 
 
+# Suppress the warning about not implementing `create` and `update` methods
+# since we don't use a model and only rely on the serializer for validation
+# pylint: disable=abstract-method
+class FileUploadSerializer(serializers.Serializer):
+    """Receive file upload requests."""
+
+    file = serializers.FileField()
+
+    def validate_file(self, file):
+        """Add file size and type constraints as defined in settings."""
+        # Validate file size
+        if file.size > settings.DOCUMENT_IMAGE_MAX_SIZE:
+            max_size = settings.DOCUMENT_IMAGE_MAX_SIZE // (1024 * 1024)
+            raise serializers.ValidationError(
+                f"File size exceeds the maximum limit of {max_size:d} MB."
+            )
+
+        # Validate file type
+        mime_type, _ = mimetypes.guess_type(file.name)
+        if mime_type not in settings.DOCUMENT_IMAGE_ALLOWED_MIME_TYPES:
+            mime_types = ", ".join(settings.DOCUMENT_IMAGE_ALLOWED_MIME_TYPES)
+            raise serializers.ValidationError(
+                f"File type '{mime_type:s}' is not allowed. Allowed types are: {mime_types:s}"
+            )
+
+        return file
+
+
 class TemplateSerializer(BaseResourceSerializer):
     """Serialize templates."""
 

src/backend/core/api/viewsets.py

@@ -1,6 +1,11 @@
 """API endpoints"""
 
+import os
+import uuid
+
+from django.conf import settings
 from django.contrib.postgres.aggregates import ArrayAgg
+from django.core.files.storage import default_storage
 from django.db.models import (
     OuterRef,
     Q,
@@ -29,6 +34,8 @@
 
 # pylint: disable=too-many-ancestors
 
+ATTACHMENTS_FOLDER = "attachments"
+
 
 class NestedGenericViewSet(viewsets.GenericViewSet):
     """
@@ -265,7 +272,7 @@ def destroy(self, request, *args, **kwargs):
         ):
             return drf_response.Response(
                 {"detail": "Cannot delete the last owner access for the resource."},
-                status=403,
+                status=status.HTTP_403_FORBIDDEN,
             )
 
         return super().destroy(request, *args, **kwargs)
@@ -390,6 +397,31 @@ def versions_detail(self, request, pk, version_id, *args, **kwargs):
             }
         )
 
+    @decorators.action(detail=True, methods=["post"], url_path="attachment-upload")
+    def attachment_upload(self, request, *args, **kwargs):
+        """Upload a file related to a given document"""
+        # Check permissions first
+        document = self.get_object()
+
+        # Validate metadata in payload
+        serializer = serializers.FileUploadSerializer(data=request.data)
+        if not serializer.is_valid():
+            return drf_response.Response(
+                serializer.errors, status=status.HTTP_400_BAD_REQUEST
+            )
+        # Extract the file extension from the original filename
+        file = serializer.validated_data["file"]
+        extension = os.path.splitext(file.name)[1]
+
+        # Generate a generic yet unique filename to store the image in object storage
+        file_id = uuid.uuid4()
+        key = f"{document.key_base}/{ATTACHMENTS_FOLDER:s}/{file_id!s}{extension:s}"
+
+        default_storage.save(key, file)
+        return drf_response.Response(
+            {"file": f"{settings.MEDIA_URL:s}{key:s}"}, status=status.HTTP_201_CREATED
+        )
+
 
 class DocumentAccessViewSet(
     ResourceAccessViewsetMixin,

src/backend/core/models.py

@@ -36,18 +36,19 @@
 
 def get_resource_roles(resource, user):
     """Compute the roles a user has on a resource."""
-    roles = []
-    if user.is_authenticated:
+    if not user.is_authenticated:
+        return []
+
+    try:
+        roles = resource.user_roles or []
+    except AttributeError:
+        teams = user.get_teams()
         try:
-            roles = resource.user_roles or []
-        except AttributeError:
-            teams = user.get_teams()
-            try:
-                roles = resource.accesses.filter(
-                    models.Q(user=user) | models.Q(team__in=teams),
-                ).values_list("role", flat=True)
-            except (models.ObjectDoesNotExist, IndexError):
-                roles = []
+            roles = resource.accesses.filter(
+                models.Q(user=user) | models.Q(team__in=teams),
+            ).values_list("role", flat=True)
+        except (models.ObjectDoesNotExist, IndexError):
+            roles = []
     return roles
 
 
@@ -403,7 +404,7 @@ def get_versions_slice(
             response = default_storage.connection.meta.client.list_object_versions(
                 Bucket=default_storage.bucket_name,
                 Prefix=self.file_key,
-                MaxKeys=settings.S3_VERSIONS_PAGE_SIZE,
+                MaxKeys=settings.DOCUMENT_VERSIONS_PAGE_SIZE,
                 **token,
             )
 
@@ -421,7 +422,7 @@ def get_versions_slice(
                     if response["NextVersionIdMarker"]:
                         return self.get_versions_slice(
                             from_version_id=response["NextVersionIdMarker"],
-                            page_size=settings.S3_VERSIONS_PAGE_SIZE,
+                            page_size=settings.DOCUMENT_VERSIONS_PAGE_SIZE,
                             from_datetime=from_datetime,
                         )
                     return {
@@ -433,9 +434,9 @@ def get_versions_slice(
         response = default_storage.connection.meta.client.list_object_versions(
             Bucket=default_storage.bucket_name,
             Prefix=self.file_key,
-            MaxKeys=min(page_size, settings.S3_VERSIONS_PAGE_SIZE)
+            MaxKeys=min(page_size, settings.DOCUMENT_VERSIONS_PAGE_SIZE)
             if page_size
-            else settings.S3_VERSIONS_PAGE_SIZE,
+            else settings.DOCUMENT_VERSIONS_PAGE_SIZE,
             **token,
         )
         return {
@@ -475,13 +476,14 @@ def get_abilities(self, user):
 
         return {
             "destroy": RoleChoices.OWNER in roles,
-            "versions_destroy": is_owner_or_admin,
-            "versions_list": can_get_versions,
-            "versions_retrieve": can_get_versions,
+            "attachment_upload": is_owner_or_admin or is_editor,
             "manage_accesses": is_owner_or_admin,
-            "update": is_owner_or_admin or is_editor,
             "partial_update": is_owner_or_admin or is_editor,
             "retrieve": can_get,
+            "update": is_owner_or_admin or is_editor,
+            "versions_destroy": is_owner_or_admin,
+            "versions_list": can_get_versions,
+            "versions_retrieve": can_get_versions,
         }
 
 

src/backend/core/tests/documents/test_api_documents_attachment_upload.py

@@ -0,0 +1,199 @@
+"""
+Test file uploads API endpoint for users in impress's core app.
+"""
+
+import re
+import uuid
+
+from django.core.files.base import ContentFile
+from django.core.files.uploadedfile import SimpleUploadedFile
+
+import pytest
+from rest_framework.test import APIClient
+
+from core import factories
+from core.tests.conftest import TEAM, USER, VIA
+
+pytestmark = pytest.mark.django_db
+
+
+def test_api_documents_attachment_upload_anonymous():
+    """Anonymous users can't upload attachments to a document."""
+    document = factories.DocumentFactory()
+    file = SimpleUploadedFile("test_file.jpg", b"Dummy content")
+
+    url = f"/api/v1.0/documents/{document.id!s}/attachment-upload/"
+    response = APIClient().post(url, {"file": file}, format="multipart")
+
+    assert response.status_code == 401
+    assert response.json() == {
+        "detail": "Authentication credentials were not provided."
+    }
+
+
+def test_api_documents_attachment_upload_authenticated_public():
+    """
+    Users who are not related to a public document should not be allowed to upload an attachment.
+    """
+    user = factories.UserFactory()
+
+    client = APIClient()
+    client.force_login(user)
+
+    document = factories.DocumentFactory(is_public=True)
+    file = SimpleUploadedFile("test_file.jpg", b"Dummy content")
+
+    url = f"/api/v1.0/documents/{document.id!s}/attachment-upload/"
+    response = client.post(url, {"file": file}, format="multipart")
+
+    assert response.status_code == 403
+    assert response.json() == {
+        "detail": "You do not have permission to perform this action."
+    }
+
+
+def test_api_documents_attachment_upload_authenticated_private():
+    """
+    Users who are not related to a private document should not be able to upload an attachment.
+    """
+    user = factories.UserFactory()
+
+    client = APIClient()
+    client.force_login(user)
+
+    document = factories.DocumentFactory(is_public=False)
+    file = SimpleUploadedFile("test_file.jpg", b"Dummy content")
+
+    url = f"/api/v1.0/documents/{document.id!s}/attachment-upload/"
+    response = client.post(url, {"file": file}, format="multipart")
+
+    assert response.status_code == 404
+    assert response.json() == {"detail": "No Document matches the given query."}
+
+
+@pytest.mark.parametrize("via", VIA)
+def test_api_documents_attachment_upload_reader(via, mock_user_get_teams):
+    """
+    Users who are simple readers on a document should not be allowed to upload an attachment.
+    """
+    user = factories.UserFactory()
+
+    client = APIClient()
+    client.force_login(user)
+
+    document = factories.DocumentFactory()
+    if via == USER:
+        factories.UserDocumentAccessFactory(document=document, user=user, role="reader")
+    elif via == TEAM:
+        mock_user_get_teams.return_value = ["lasuite", "unknown"]
+        factories.TeamDocumentAccessFactory(
+            document=document, team="lasuite", role="reader"
+        )
+
+    file = SimpleUploadedFile("test_file.jpg", b"Dummy content")
+
+    url = f"/api/v1.0/documents/{document.id!s}/attachment-upload/"
+    response = client.post(url, {"file": file}, format="multipart")
+
+    assert response.status_code == 403
+    assert response.json() == {
+        "detail": "You do not have permission to perform this action."
+    }
+
+
+@pytest.mark.parametrize("role", ["editor", "administrator", "owner"])
+@pytest.mark.parametrize("via", VIA)
+def test_api_documents_attachment_upload_success(
+    via, role, mock_user_get_teams, settings
+):
+    """
+    Editors, administrators and owners of a document should be able to upload an attachment.
+    """
+    user = factories.UserFactory()
+
+    client = APIClient()
+    client.force_login(user)
+
+    document = factories.DocumentFactory()
+    if via == USER:
+        factories.UserDocumentAccessFactory(document=document, user=user, role=role)
+    elif via == TEAM:
+        mock_user_get_teams.return_value = ["lasuite", "unknown"]
+        factories.TeamDocumentAccessFactory(
+            document=document, team="lasuite", role=role
+        )
+
+    file = SimpleUploadedFile("test_file.jpg", b"Dummy content")
+
+    url = f"/api/v1.0/documents/{document.id!s}/attachment-upload/"
+    response = client.post(url, {"file": file}, format="multipart")
+
+    assert response.status_code == 201
+
+    pattern = re.compile(rf"^{settings.MEDIA_URL}{document.id!s}/attachments/(.*)\.jpg")
+    match = pattern.search(response.json()["file"])
+    file_id = match.group(1)
+
+    # Validate that file_id is a valid UUID
+    uuid.UUID(file_id)
+
+
+def test_api_documents_attachment_upload_invalid(client):
+    """Attempt to upload without a file should return an explicit error."""
+    user = factories.UserFactory()
+
+    client = APIClient()
+    client.force_login(user)
+
+    document = factories.DocumentFactory(users=[(user, "owner")])
+    url = f"/api/v1.0/documents/{document.id!s}/attachment-upload/"
+
+    response = client.post(url, {}, format="multipart")
+
+    assert response.status_code == 400
+    assert response.json() == {"file": ["No file was submitted."]}
+
+
+def test_api_documents_attachment_upload_size_limit_exceeded(settings):
+    """The uploaded file should not exceeed the maximum size in settings."""
+    settings.DOCUMENT_IMAGE_MAX_SIZE = 1048576  # 1 MB for test
+
+    user = factories.UserFactory()
+    client = APIClient()
+    client.force_login(user)
+
+    document = factories.DocumentFactory(users=[(user, "owner")])
+    url = f"/api/v1.0/documents/{document.id!s}/attachment-upload/"
+
+    # Create a temporary file larger than the allowed size
+    content = b"a" * (1048576 + 1)
+    file = ContentFile(content, name="test.jpg")
+
+    response = client.post(url, {"file": file}, format="multipart")
+
+    assert response.status_code == 400
+    assert response.json() == {"file": ["File size exceeds the maximum limit of 1 MB."]}
+
+
+def test_api_documents_attachment_upload_type_not_allowed(settings):
+    """The uploaded file should be of a whitelisted type."""
+    settings.DOCUMENT_IMAGE_ALLOWED_MIME_TYPES = ["image/jpeg", "image/png"]
+
+    user = factories.UserFactory()
+    client = APIClient()
+    client.force_login(user)
+
+    document = factories.DocumentFactory(users=[(user, "owner")])
+    url = f"/api/v1.0/documents/{document.id!s}/attachment-upload/"
+
+    # Create a temporary file with a not allowed type (e.g., text file)
+    file = ContentFile(b"a" * 1048576, name="test.txt")
+
+    response = client.post(url, {"file": file}, format="multipart")
+
+    assert response.status_code == 400
+    assert response.json() == {
+        "file": [
+            "File type 'text/plain' is not allowed. Allowed types are: image/jpeg, image/png"
+        ]
+    }