[INFRA-501] fix(security): validate the destination entity_id, not project_id, for PROJECT_COVER duplicates

get_entity_id_field() derives the persisted project_id for a PROJECT_COVER
duplicate from entity_id, overriding whatever project_id was supplied in the
request body -- so entity_id, not project_id, is the value that actually
lands on the new row. The membership check only ever validated project_id,
so a caller could name a project they belong to there just to pass that
check, while entity_id (the real destination) pointed at a project they
were never checked against.

Composes with the prior project_id=None default fix: for PROJECT_COVER we
now set project_id = entity_id before that block runs, so the same
existence/membership checks and the default-to-source-project fallback
apply to the value that's actually persisted. Also stops get_entity_id_field's
project_id colliding with the explicit project_id kwarg passed to
FileAsset.objects.create().

Co-authored-by: Plane AI <noreply@plane.so>

apps/api/plane/app/views/asset/v2.py

@@ -834,6 +834,16 @@ class DuplicateAssetEndpoint(BaseAPIView):
                 status=status.HTTP_403_FORBIDDEN,
             )
 
+        # get_entity_id_field() (below) derives the persisted project_id for a
+        # PROJECT_COVER duplicate from entity_id, overriding whatever project_id
+        # was supplied separately -- so entity_id, not the request body's
+        # project_id, is the value that actually lands on the new row. Validate
+        # that value here too, or a caller could pass a project_id they belong to
+        # just to clear the membership check below while entity_id -- the real
+        # destination -- points at a project they were never checked against.
+        if entity_type == FileAsset.EntityTypeContext.PROJECT_COVER:
+            project_id = entity_id
+
         # A caller may redirect the copy to a different project than the source
         # (e.g. duplicating an attachment onto an issue that lives in another
         # project) by naming project_id explicitly -- that's still validated
@@ -868,6 +878,11 @@ class DuplicateAssetEndpoint(BaseAPIView):
 
         sanitized_name = sanitize_filename(original_asset.attributes.get("name")) or "unnamed"
         destination_key = f"{workspace.id}/{uuid.uuid4().hex}-{sanitized_name}"
+        entity_id_fields = self.get_entity_id_field(entity_type=entity_type, entity_id=entity_id)
+        # project_id is already validated above -- and for PROJECT_COVER it *is*
+        # entity_id -- so drop any project_id get_entity_id_field derived to avoid
+        # passing it twice to create() below.
+        entity_id_fields.pop("project_id", None)
         duplicated_asset = FileAsset.objects.create(
             attributes={
                 "name": original_asset.attributes.get("name"),
@@ -881,7 +896,7 @@ class DuplicateAssetEndpoint(BaseAPIView):
             entity_type=entity_type,
             project_id=project_id if project_id else None,
             storage_metadata=original_asset.storage_metadata,
-            **self.get_entity_id_field(entity_type=entity_type, entity_id=entity_id),
+            **entity_id_fields,
         )
         storage.copy_object(original_asset.asset, destination_key)
         # Update the is_uploaded field for all newly created assets

apps/api/plane/tests/contract/app/test_workspace_asset_routes_project_scope_app.py

@@ -354,3 +354,61 @@ class TestDuplicateAssetProjectScope:
         assert FileAsset.objects.count() == before + 1
         duplicated_asset = FileAsset.objects.get(id=response.data["asset_id"])
         assert duplicated_asset.project_id == project.id
+
+    @pytest.mark.django_db
+    def test_duplicate_project_cover_denied_when_entity_id_is_foreign_project(
+        self, session_client, workspace, project, project_asset, outsider_project
+    ):
+        """PROJECT_COVER's entity_id -- not project_id -- is the real destination.
+
+        get_entity_id_field() turns entity_id into the project_id that actually
+        gets persisted, overriding whatever project_id was supplied. A caller who
+        puts a project they belong to in project_id (clearing the membership
+        check) but a foreign project in entity_id must still be rejected, or the
+        duplicate lands in a project they were never checked against.
+        """
+        before = FileAsset.objects.count()
+        with mock.patch(S3_STORAGE_PATH) as mock_storage:
+            response = session_client.post(
+                duplicate_url(workspace.slug, project_asset.id),
+                {
+                    "entity_type": FileAsset.EntityTypeContext.PROJECT_COVER,
+                    "project_id": str(project.id),
+                    "entity_id": str(outsider_project.id),
+                },
+                format="json",
+            )
+
+        assert response.status_code == status.HTTP_403_FORBIDDEN, (
+            f"Got {response.status_code}: {getattr(response, 'data', None)!r}"
+        )
+        mock_storage.return_value.copy_object.assert_not_called()
+        assert FileAsset.objects.count() == before, (
+            "a PROJECT_COVER duplicate was persisted into a project only named "
+            "via entity_id, which was never validated"
+        )
+
+    @pytest.mark.django_db
+    def test_duplicate_project_cover_allowed_for_own_project_entity_id(
+        self, session_client, workspace, project, project_asset
+    ):
+        """Sanity check: a legitimate PROJECT_COVER duplicate still works, and the
+        persisted project_id follows entity_id as intended."""
+        before = FileAsset.objects.count()
+        with mock.patch(S3_STORAGE_PATH):
+            response = session_client.post(
+                duplicate_url(workspace.slug, project_asset.id),
+                {
+                    "entity_type": FileAsset.EntityTypeContext.PROJECT_COVER,
+                    "project_id": str(project.id),
+                    "entity_id": str(project.id),
+                },
+                format="json",
+            )
+
+        assert response.status_code == status.HTTP_200_OK, (
+            f"Got {response.status_code}: {getattr(response, 'data', None)!r}"
+        )
+        assert FileAsset.objects.count() == before + 1
+        duplicated_asset = FileAsset.objects.get(id=response.data["asset_id"])
+        assert duplicated_asset.project_id == project.id