[INFRA-501] fix(security): enforce project membership on every workspace-level asset route

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

@@ -18,7 +18,7 @@
 from plane.bgtasks.storage_metadata_task import get_asset_object_metadata
 from plane.settings.storage import S3Storage
 from plane.utils.path_validator import sanitize_filename
-from plane.db.models import FileAsset, User, Workspace
+from plane.db.models import FileAsset, Project, ProjectMember, User, Workspace
 from plane.app.permissions import WorkspaceUserPermission
 from plane.api.views.base import BaseAPIView
 from plane.api.serializers import (
@@ -335,7 +335,7 @@ def post(self, request):
         )
 
         # Get the presigned URL
-        storage = S3Storage(request=request, is_server=True)
+        storage = S3Storage(request=request)
         # Generate a presigned URL to share an S3 object
         presigned_url = storage.generate_presigned_post(object_name=asset_key, file_type=type, file_size=size_limit)
         # Return the presigned URL
@@ -437,6 +437,16 @@ def get(self, request, slug, asset_id):
             # Get the asset
             asset = FileAsset.objects.get(id=asset_id, workspace_id=workspace.id, is_deleted=False)
 
+            # WorkspaceUserPermission admits any active member of the workspace,
+            # including a GUEST who belongs to no project. The asset lookup binds
+            # the workspace and nothing else, so the project dimension has to be
+            # enforced here -- as the app surface already does for the same model.
+            if not asset.is_project_accessible_to(request.user):
+                return Response(
+                    {"error": "You don't have access to this asset."},
+                    status=status.HTTP_403_FORBIDDEN,
+                )
+
             # Check if the asset exists and is uploaded
             if not asset.is_uploaded:
                 return Response(
@@ -448,7 +458,7 @@ def get(self, request, slug, asset_id):
             # Force attachment disposition for script-capable MIME types (e.g. SVG)
             # to prevent same-origin XSS when the asset URL shares the app's origin
             # (default MinIO self-hosted setup).
-            storage = S3Storage(request=request, is_server=True)
+            storage = S3Storage(request=request)
             asset_mime_type = (asset.attributes.get("type") or "").split(";")[0].strip().lower()
             disposition = (
                 "attachment" if asset_mime_type in settings.SCRIPT_CAPABLE_MIME_TYPES else "inline"
@@ -542,6 +552,28 @@ def post(self, request, slug):
         # Get the workspace
         workspace = Workspace.objects.get(slug=slug)
 
+        # project_id arrives in the request body and was stored unvalidated, so a
+        # member of one workspace could mint an asset row pointing at a project in
+        # another. Bind it to the URL workspace and to the caller's membership;
+        # rows where workspace_id != project.workspace_id are the inconsistency
+        # is_project_accessible_to has to defend against downstream.
+        if project_id:
+            if not Project.objects.filter(id=project_id, workspace_id=workspace.id).exists():
+                return Response(
+                    {"error": "Project not found.", "status": False},
+                    status=status.HTTP_404_NOT_FOUND,
+                )
+            if not ProjectMember.objects.filter(
+                member=request.user,
+                workspace_id=workspace.id,
+                project_id=project_id,
+                is_active=True,
+            ).exists():
+                return Response(
+                    {"error": "You don't have access to this project.", "status": False},
+                    status=status.HTTP_403_FORBIDDEN,
+                )
+
         # asset key
         asset_key = f"{workspace.id}/{uuid.uuid4().hex}-{name}"
 
@@ -555,6 +587,28 @@ def post(self, request, slug):
             ).first()
 
             if existing_asset:
+                # The dedup lookup is scoped to the workspace only -- and when the
+                # body omits project_id the validation above is skipped entirely --
+                # so the match may belong to a project the caller cannot see.
+                # Echoing it would hand over that asset's id, and asset_url
+                # additionally embeds the owning project and issue ids. Knowing the
+                # asset UUID is the precondition for every asset-scoped attack on
+                # this surface, so this branch must not supply it.
+                #
+                # 404 rather than 403 on purpose: a 403 would still confirm that
+                # some asset carries this external id pair in this workspace, which
+                # turns the pair into an existence oracle. The elsewhere-consistent
+                # 403 is fine on routes where the caller already named the asset id;
+                # here they named only an external id, so a match is new
+                # information. The cost is that a caller who guesses a pair held by
+                # a project they cannot see cannot create their own asset under it,
+                # which is the right trade -- real integrations mint ids per source
+                # and run as a member of the target project.
+                if not existing_asset.is_project_accessible_to(request.user):
+                    return Response(
+                        {"error": "Asset not found.", "status": False},
+                        status=status.HTTP_404_NOT_FOUND,
+                    )
                 return Response(
                     {
                         "message": "Asset with same external id and source already exists",
@@ -578,7 +632,7 @@ def post(self, request, slug):
         )
 
         # Get the presigned URL
-        storage = S3Storage(request=request, is_server=True)
+        storage = S3Storage(request=request)
         presigned_url = storage.generate_presigned_post(object_name=asset_key, file_type=type, file_size=size_limit)
 
         return Response(
@@ -620,6 +674,15 @@ def patch(self, request, slug, asset_id):
         try:
             asset = FileAsset.objects.get(id=asset_id, workspace__slug=slug, is_deleted=False)
 
+            # is_uploaded gates every download path, so an unscoped write here
+            # lets any workspace member make another project's attachment vanish
+            # for its own members, or mark a never-uploaded asset complete.
+            if not asset.is_project_accessible_to(request.user):
+                return Response(
+                    {"error": "You don't have access to this asset."},
+                    status=status.HTTP_403_FORBIDDEN,
+                )
+
             # Update is_uploaded status
             asset.is_uploaded = request.data.get("is_uploaded", asset.is_uploaded)
 

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

@@ -313,30 +313,6 @@ def entity_asset_delete(self, entity_type, asset, request):
         else:
             return
 
-    def has_project_asset_access(self, request, asset):
-        """Return whether the user may access a workspace-scoped asset.
-
-        This endpoint is authorized at the WORKSPACE level, so a workspace
-        member/guest could otherwise reach an asset that belongs to a project
-        they are not a member of. For project-bound assets, require an active
-        ProjectMember of the asset's project. Workspace-level entity types
-        (WORKSPACE_LOGO, USER_AVATAR, USER_COVER) have project_id=None and are
-        always allowed.
-        """
-        if asset.project_id is None:
-            return True
-        # Scope the membership lookup to the asset's workspace as well as its
-        # project, mirroring allow_permission's PROJECT branch. This prevents a
-        # member of the same project in a different workspace from passing the
-        # check should an asset row ever be inconsistent (asset.workspace_id !=
-        # asset.project.workspace_id).
-        return ProjectMember.objects.filter(
-            member=request.user,
-            workspace_id=asset.workspace_id,
-            project_id=asset.project_id,
-            is_active=True,
-        ).exists()
-
     @allow_permission([ROLE.ADMIN, ROLE.MEMBER, ROLE.GUEST], level="WORKSPACE")
     def post(self, request, slug):
         name = sanitize_filename(request.data.get("name")) or "unnamed"
@@ -419,7 +395,7 @@ def patch(self, request, slug, asset_id):
         # get the asset id
         asset = FileAsset.objects.get(id=asset_id, workspace__slug=slug)
         # enforce project-level access for project-bound assets
-        if not self.has_project_asset_access(request, asset):
+        if not asset.is_project_accessible_to(request.user):
             return Response(
                 {"error": "You don't have access to this asset."},
                 status=status.HTTP_403_FORBIDDEN,
@@ -446,7 +422,7 @@ def patch(self, request, slug, asset_id):
     def delete(self, request, slug, asset_id):
         asset = FileAsset.objects.get(id=asset_id, workspace__slug=slug)
         # enforce project-level access for project-bound assets
-        if not self.has_project_asset_access(request, asset):
+        if not asset.is_project_accessible_to(request.user):
             return Response(
                 {"error": "You don't have access to this asset."},
                 status=status.HTTP_403_FORBIDDEN,
@@ -463,7 +439,7 @@ def get(self, request, slug, asset_id):
         # get the asset id
         asset = FileAsset.objects.get(id=asset_id, workspace__slug=slug)
         # enforce project-level access for project-bound assets
-        if not self.has_project_asset_access(request, asset):
+        if not asset.is_project_accessible_to(request.user):
             return Response(
                 {"error": "You don't have access to this asset."},
                 status=status.HTTP_403_FORBIDDEN,
@@ -539,6 +515,14 @@ class AssetRestoreEndpoint(BaseAPIView):
     @allow_permission([ROLE.ADMIN, ROLE.MEMBER, ROLE.GUEST], level="WORKSPACE")
     def post(self, request, slug, asset_id):
         asset = FileAsset.all_objects.get(id=asset_id, workspace__slug=slug)
+        # Authorized at the WORKSPACE level, so without this a workspace member
+        # who is not in the asset's project could reverse a deletion performed
+        # by that project's own members.
+        if not asset.is_project_accessible_to(request.user):
+            return Response(
+                {"error": "You don't have access to this asset."},
+                status=status.HTTP_403_FORBIDDEN,
+            )
         asset.is_deleted = False
         asset.deleted_at = None
         asset.save(update_fields=["is_deleted", "deleted_at"])
@@ -772,8 +756,12 @@ class AssetCheckEndpoint(BaseAPIView):
 
     @allow_permission([ROLE.ADMIN, ROLE.MEMBER, ROLE.GUEST], level="WORKSPACE")
     def get(self, request, slug, asset_id):
-        asset = FileAsset.all_objects.filter(id=asset_id, workspace__slug=slug, deleted_at__isnull=True).exists()
-        return Response({"exists": asset}, status=status.HTTP_200_OK)
+        asset = FileAsset.all_objects.filter(id=asset_id, workspace__slug=slug, deleted_at__isnull=True).first()
+        # Report existence only to callers who could otherwise reach the asset.
+        # Reporting it unconditionally makes this route an existence oracle for
+        # every project in the workspace, including ones the caller cannot see.
+        exists = asset is not None and asset.is_project_accessible_to(request.user)
+        return Response({"exists": exists}, status=status.HTTP_200_OK)
 
 
 class DuplicateAssetEndpoint(BaseAPIView):
@@ -829,6 +817,20 @@ def post(self, request, slug, asset_id):
             # check if project exists in the workspace
             if not Project.objects.filter(id=project_id, workspace=workspace).exists():
                 return Response({"error": "Project not found"}, status=status.HTTP_404_NOT_FOUND)
+            # project_id is the *destination* and comes from the request body.
+            # Existence in the workspace is not authorization: require the caller
+            # to be an active member of the project the copy will land in, or a
+            # workspace member could deposit assets into any project.
+            if not ProjectMember.objects.filter(
+                member=request.user,
+                workspace=workspace,
+                project_id=project_id,
+                is_active=True,
+            ).exists():
+                return Response(
+                    {"error": "You don't have access to this project."},
+                    status=status.HTTP_403_FORBIDDEN,
+                )
 
         storage = S3Storage(request=request)
         # Restrict the source asset to the same destination workspace to prevent cross-workspace asset copying
@@ -841,6 +843,15 @@ def post(self, request, slug, asset_id):
         if not original_asset:
             return Response({"error": "Asset not found"}, status=status.HTTP_404_NOT_FOUND)
 
+        # The source lookup binds the workspace but not the project, so without
+        # this a non-member could copy a project's asset into a project they do
+        # control -- a permanent copy that outlives the original being deleted.
+        if not original_asset.is_project_accessible_to(request.user):
+            return Response(
+                {"error": "You don't have access to this asset."},
+                status=status.HTTP_403_FORBIDDEN,
+            )
+
         sanitized_name = sanitize_filename(original_asset.attributes.get("name")) or "unnamed"
         destination_key = f"{workspace.id}/{uuid.uuid4().hex}-{sanitized_name}"
         duplicated_asset = FileAsset.objects.create(
@@ -882,6 +893,16 @@ def get(self, request, slug, asset_id):
                 status=status.HTTP_404_NOT_FOUND,
             )
 
+        # The workspace-level twin of ProjectAssetDownloadEndpoint, which binds
+        # project_id through level="PROJECT". Here the project is not in the URL,
+        # so it has to be enforced against the asset itself -- otherwise the
+        # presigned URL hands the file to a non-member of its project.
+        if not asset.is_project_accessible_to(request.user):
+            return Response(
+                {"error": "You don't have access to this asset."},
+                status=status.HTTP_403_FORBIDDEN,
+            )
+
         storage = S3Storage(request=request)
         signed_url = storage.generate_presigned_url(
             object_name=asset.asset.name,

apps/api/plane/db/models/asset.py

@@ -14,6 +14,7 @@
 from plane.utils.path_validator import sanitize_filename
 
 from .base import BaseModel
+from .project import ProjectMember
 
 
 def get_upload_path(instance, filename):
@@ -101,3 +102,38 @@ def asset_url(self):
             return f"/api/assets/v2/workspaces/{self.workspace.slug}/projects/{self.project_id}/{self.id}/"
 
         return None
+
+    def is_project_accessible_to(self, user):
+        """Return whether ``user`` clears the project dimension of access to this asset.
+
+        This is the project-membership half of asset authorization and nothing
+        more. Callers are still responsible for establishing that ``user`` may
+        act in this asset's workspace at all -- typically the endpoint's
+        permission class or ``allow_permission(..., level="WORKSPACE")``.
+
+        It exists as a model method rather than a view helper because the
+        workspace-level asset routes are spread across several unrelated
+        ``BaseAPIView`` subclasses in both the app and the external API, and a
+        helper bound to one of those classes is unreachable from the others.
+        That is precisely how the earlier project-scoping fix came to cover
+        three handlers and miss the rest: a route added later has no way to
+        inherit the rule. Keeping it on the model means every surface that can
+        load a ``FileAsset`` can also ask the question.
+
+        Assets with no project (workspace logos, user avatars and covers) carry
+        ``project_id=None`` and are workspace-level by definition, so they clear
+        this check; workspace authorization is the only gate that applies.
+        """
+        if self.project_id is None:
+            return True
+        # Scope the membership lookup to this asset's workspace as well as its
+        # project. A project id alone would let a member of a same-id project in
+        # a different workspace pass, should a row ever be inconsistent
+        # (workspace_id != project.workspace_id) -- a state the external API's
+        # create path could previously produce.
+        return ProjectMember.objects.filter(
+            member=user,
+            workspace_id=self.workspace_id,
+            project_id=self.project_id,
+            is_active=True,
+        ).exists()

apps/api/plane/tests/contract/api/conftest.py

@@ -0,0 +1,41 @@
+# Copyright (c) 2023-present Plane Software, Inc. and contributors
+# SPDX-License-Identifier: AGPL-3.0-only
+# See the LICENSE file for details.
+
+"""Shared fixtures for the external-API contract tests.
+
+Every test in this package authenticates with the ``api_token`` fixture, whose
+token string is a constant. ``ApiKeyRateThrottle.get_cache_key`` keys on that
+string, so all of these tests share one throttle bucket for the whole run --
+``API_KEY_RATE_LIMIT`` defaults to 60/minute, and the suite finishes well inside
+a minute. Once the package as a whole crosses 60 requests, whichever test issues
+the next one fails with 429, which shows up as an unrelated test breaking in a
+file nobody touched.
+
+Resetting the bucket around each test keeps the failure attributable and stops
+the package having an effective cap on how many API calls its tests may make in
+total. Only this throttle's key is removed, mirroring the narrowly-scoped
+``_clear_auth_throttle_keys`` helper in the app authentication tests, rather than
+calling ``cache.clear()`` and disturbing unrelated cached state.
+"""
+
+import pytest
+from django.core.cache import cache
+
+
+def _clear_api_key_throttle_keys():
+    """Delete only ApiKeyRateThrottle history keys from the shared cache.
+
+    ``ApiKeyRateThrottle`` overrides ``get_cache_key`` to return
+    ``f"{self.scope}:{api_key}"``, so its entries are ``api_key:<token>`` and do
+    not carry DRF's usual ``throttle_`` prefix.
+    """
+    cache.delete_pattern("api_key:*")
+
+
+@pytest.fixture(autouse=True)
+def _reset_api_key_throttle_cache():
+    """Give every external-API contract test a clean throttle bucket."""
+    _clear_api_key_throttle_keys()
+    yield
+    _clear_api_key_throttle_keys()