fixup! ✨(backend) add duplicate action to the document API endpoint

Author: sampaccoud

src/backend/core/api/viewsets.py

@@ -30,7 +30,7 @@
 from core import authentication, enums, models
 from core.services.ai_services import AIService
 from core.services.collaboration_services import CollaborationService
-from core.utils import extract_attachments
+from core.utils import extract_attachments, filter_descendants
 
 from . import permissions, serializers, utils
 from .filters import DocumentFilter, ListDocumentFilter
@@ -891,7 +891,8 @@ def tree(self, request, pk, *args, **kwargs):
         )
         return drf.response.Response(
             utils.nest_tree(serializer.data, self.queryset.model.steplen)
-        
+        )
+
     @drf.decorators.action(
         detail=True,
         methods=["post"],
@@ -907,9 +908,11 @@ def duplicate(self, request, *args, **kwargs):
         Optionally duplicates accesses if `with_accesses` is set to true
         in the payload.
         """
-        serializer = serializers.DocumentDuplicationSerializer(data=request.GET)
+        serializer = serializers.DocumentDuplicationSerializer(
+            data=request.data, partial=True
+        )
         serializer.is_valid(raise_exception=True)
-        with_accesses = serializer.validated_data["with_accesses"]
+        with_accesses = serializer.validated_data.get("with_accesses", False)
 
         # Get document while checking permissions
         document = self.get_object()
@@ -1114,7 +1117,7 @@ def attachment_upload(self, request, *args, **kwargs):
 
         # Generate a generic yet unique filename to store the image in object storage
         file_id = uuid.uuid4()
-        extension = serializer.validated_data["expected_extension"]
+        ext = serializer.validated_data["expected_extension"]
 
         # Prepare metadata for storage
         extra_args = {
@@ -1126,7 +1129,7 @@ def attachment_upload(self, request, *args, **kwargs):
             extra_args["Metadata"]["is_unsafe"] = "true"
             file_unsafe = "-unsafe"
 
-        key = f"{document.key_base}/{ATTACHMENTS_FOLDER:s}/{file_id!s}{file_unsafe}.{extension:s}"
+        key = f"{document.key_base}/{enums.ATTACHMENTS_FOLDER:s}/{file_id!s}{file_unsafe}.{ext:s}"
 
         file_name = serializer.validated_data["file_name"]
         if (
@@ -1221,12 +1224,27 @@ def media_auth(self, request, *args, **kwargs):
         user = request.user
         key = f"{url_params['pk']:s}/{url_params['attachment']:s}"
 
-        if (
-            not self.queryset.readable(user)
-            .filter(attachments__contains=[key])
-            .exists()
-        ):
-            logger.debug("User '%s' lacks permission for image", user)
+        # Look for a document to which the user has access and that includes this attachment
+        # We must look into all descendants of any document to which the user has access per se
+        readable_per_se_paths = (
+            self.queryset.readable_per_se(user)
+            .order_by("path")
+            .values_list("path", flat=True)
+        )
+
+        attachments_documents = (
+            self.queryset.filter(attachments__contains=[key])
+            .only("path")
+            .order_by("path")
+        )
+        readable_attachments_paths = filter_descendants(
+            [doc.path for doc in attachments_documents],
+            readable_per_se_paths,
+            skip_sorting=True,
+        )
+
+        if not readable_attachments_paths:
+            logger.debug("User '%s' lacks permission for attachment", user)
             raise drf.exceptions.PermissionDenied()
 
         # Generate S3 authorization headers using the extracted URL parameters

src/backend/core/enums.py

@@ -12,10 +12,10 @@
 UUID_REGEX = (
     r"[a-fA-F0-9]{8}-[a-fA-F0-9]{4}-[a-fA-F0-9]{4}-[a-fA-F0-9]{4}-[a-fA-F0-9]{12}"
 )
-FILE_EXT_REGEX = r"\.[a-zA-Z]{3,4}"
+FILE_EXT_REGEX = r"\.[a-zA-Z0-9]{1,10}"
 MEDIA_STORAGE_URL_PATTERN = re.compile(
     f"{settings.MEDIA_URL:s}(?P<pk>{UUID_REGEX:s})/"
-    f"(?P<key>{ATTACHMENTS_FOLDER:s}/{UUID_REGEX:s}(?:-unsafe)?{FILE_EXT_REGEX:s})$"
+    f"(?P<attachment>{ATTACHMENTS_FOLDER:s}/{UUID_REGEX:s}(?:-unsafe)?{FILE_EXT_REGEX:s})$"
 )
 MEDIA_STORAGE_URL_EXTRACT = re.compile(
     f"{settings.MEDIA_URL:s}({UUID_REGEX}/{ATTACHMENTS_FOLDER}/{UUID_REGEX}{FILE_EXT_REGEX})"

src/backend/core/migrations/0018_remove_is_public_add_field_attachments_and_duplicated_from.py

@@ -0,0 +1,77 @@
+# Generated by Django 5.1.4 on 2025-01-18 11:53
+import re
+
+import django.contrib.postgres.fields
+import django.db.models.deletion
+from django.core.files.storage import default_storage
+from django.db import migrations, models
+
+from botocore.exceptions import ClientError
+
+import core.models
+from core.utils import extract_attachments
+
+
+def populate_attachments_on_all_documents(apps, schema_editor):
+    """Populate "attachments" field on all existing documents in the database."""
+    Document = apps.get_model("core", "Document")
+
+    for document in Document.objects.all():
+        try:
+            response = default_storage.connection.meta.client.get_object(
+                Bucket=default_storage.bucket_name, Key=f"{document.pk!s}/file"
+            )
+        except (FileNotFoundError, ClientError):
+            pass
+        else:
+            content = response["Body"].read().decode("utf-8")
+            document.attachments = extract_attachments(content)
+            document.save(update_fields=["attachments"])
+
+
+class Migration(migrations.Migration):
+    dependencies = [
+        ("core", "0019_alter_user_language_default_to_null"),
+    ]
+
+    operations = [
+        # v2.0.0 was released so we can now remove BC field "is_public"
+        migrations.RemoveField(
+            model_name="document",
+            name="is_public",
+        ),
+        migrations.AlterModelManagers(
+            name="user",
+            managers=[
+                ("objects", core.models.UserManager()),
+            ],
+        ),
+        migrations.AddField(
+            model_name="document",
+            name="attachments",
+            field=django.contrib.postgres.fields.ArrayField(
+                base_field=models.CharField(max_length=255),
+                blank=True,
+                default=list,
+                editable=False,
+                null=True,
+                size=None,
+            ),
+        ),
+        migrations.AddField(
+            model_name="document",
+            name="duplicated_from",
+            field=models.ForeignKey(
+                blank=True,
+                editable=False,
+                null=True,
+                on_delete=django.db.models.deletion.SET_NULL,
+                related_name="duplicates",
+                to="core.document",
+            ),
+        ),
+        migrations.RunPython(
+            populate_attachments_on_all_documents,
+            reverse_code=migrations.RunPython.noop,
+        ),
+    ]

src/backend/core/migrations/0020_remove_is_public_add_field_attachments_and_duplicated_from.py

@@ -428,10 +428,12 @@ class DocumentQuerySet(MP_NodeQuerySet):
 
     def readable_per_se(self, user):
         """
-        Filters the queryset to return documents that the given user has
-        permission to read.
+        Filters the queryset to return documents on which the given user has
+        direct access, team access or link access. This will not return all the
+        documents that a user can read because it can be obtained via an ancestor.
         :param user: The user for whom readable documents are to be fetched.
-        :return: A queryset of documents readable by the user.
+        :return: A queryset of documents for which the user has direct access,
+            team access or link access.
         """
         if user.is_authenticated:
             return self.filter(
@@ -460,7 +462,9 @@ def readable_per_se(self, user):
         """
         Filters documents based on user permissions using the custom queryset.
         :param user: The user for whom readable documents are to be fetched.
-        :return: A queryset of documents readable by the user.
+        :return: A queryset of documents for which the user has direct access,
+            team access or link access. This will not return all the documents
+            that a user can read because it can be obtained via an ancestor.
         """
         return self.get_queryset().readable_per_se(user)
 

src/backend/core/models.py

@@ -313,16 +313,16 @@ def test_api_documents_attachment_upload_fix_extension(
     match = pattern.search(file_path)
     file_id = match.group(1)
 
-    assert "-unsafe" in file_id
-    # Validate that file_id is a valid UUID
-    file_id = file_id.replace("-unsafe", "")
-    uuid.UUID(file_id)
-
     document.refresh_from_db()
     assert document.attachments == [
         f"{document.id!s}/attachments/{file_id!s}.{extension:s}"
     ]
 
+    assert "-unsafe" in file_id
+    # Validate that file_id is a valid UUID
+    file_id = file_id.replace("-unsafe", "")
+    uuid.UUID(file_id)
+
     # Now, check the metadata of the uploaded file
     key = file_path.replace("/media", "")
     file_head = default_storage.connection.meta.client.head_object(
@@ -373,14 +373,14 @@ def test_api_documents_attachment_upload_unsafe():
     match = pattern.search(file_path)
     file_id = match.group(1)
 
+    document.refresh_from_db()
+    assert document.attachments == [f"{document.id!s}/attachments/{file_id!s}.exe"]
+
     assert "-unsafe" in file_id
     # Validate that file_id is a valid UUID
     file_id = file_id.replace("-unsafe", "")
     uuid.UUID(file_id)
 
-    document.refresh_from_db()
-    assert document.attachments == [f"{document.id!s}/attachments/{file_id!s}.exe"]
-
     # Now, check the metadata of the uploaded file
     key = file_path.replace("/media", "")
     file_head = default_storage.connection.meta.client.head_object(

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

@@ -182,7 +182,9 @@ def test_api_documents_duplicate_with_accesses():
 
     # Duplicate the document via the API endpoint requesting to duplicate accesses
     response = client.post(
-        f"/api/v1.0/documents/{document.id!s}/duplicate/?with_accesses=true"
+        f"/api/v1.0/documents/{document.id!s}/duplicate/",
+        {"with_accesses": True},
+        format="json",
     )
 
     assert response.status_code == 201

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

@@ -81,8 +81,6 @@ def test_api_documents_media_auth_anonymous_public():
 
 def test_api_documents_media_auth_extensions():
     """Files with extensions of any format should work."""
-    document = factories.DocumentFactory(link_reach="public")
-
     extensions = [
         "c",
         "go",
@@ -91,10 +89,15 @@ def test_api_documents_media_auth_extensions():
         "woff2",
         "appimage",
     ]
+    document_id = uuid4()
+    keys = []
     for ext in extensions:
-        filename = f"{uuid.uuid4()!s}.{ext:s}"
-        key = f"{document.pk!s}/attachments/{filename:s}"
+        filename = f"{uuid4()!s}.{ext:s}"
+        keys.append(f"{document_id!s}/attachments/{filename:s}")
+
+    factories.DocumentFactory(link_reach="public", attachments=keys)
 
+    for key in keys:
         original_url = f"http://localhost/media/{key:s}"
         response = APIClient().get(
             "/api/v1.0/documents/media-auth/", HTTP_X_ORIGINAL_URL=original_url
@@ -127,7 +130,7 @@ def test_api_documents_media_auth_anonymous_attachments():
     """
     Declaring a media key as original attachment on a document to which
     a user has access should give them access to the attachment file
-    regarless of their access rights on the original document.
+    regardless of their access rights on the original document.
     """
     document_id = uuid4()
     filename = f"{uuid4()!s}.jpg"
@@ -150,7 +153,8 @@ def test_api_documents_media_auth_anonymous_attachments():
 
     # Let's now add a document to which the anonymous user has access and
     # pointing to the attachment
-    factories.DocumentFactory(link_reach="public", attachments=[key])
+    parent = factories.DocumentFactory(link_reach="public")
+    factories.DocumentFactory(parent=parent, link_reach="restricted", attachments=[key])
 
     response = APIClient().get(
         "/api/v1.0/documents/media-auth/", HTTP_X_ORIGINAL_URL=media_url