[WEB-8066] fix: scope workspace asset get/patch/delete to project membership

WorkspaceFileAssetEndpoint is authorized at the WORKSPACE level, so any
workspace member/guest could reach get/patch/delete for a project-bound
asset (issue attachment/description, comment description, page description)
of a project they are not a member of — an incomplete fix of the GHSA-qw87
asset-IDOR cluster (GHSA-h7mc-p9mm-2r4w / GHSA-cjph-cgm5-8pw8).

Add project_membership_denied(): for project-bound assets (project_id set)
require an active ProjectMember of the asset's project, else 403. Workspace-
level entity types (WORKSPACE_LOGO, USER_AVATAR, USER_COVER) have project_id
NULL and remain accessible to any workspace member. Mirrors ProjectAssetEndpoint
(level=PROJECT). Guard runs before the is_uploaded check / mutation so a
non-member gets a uniform 403 and cannot probe upload state.

Contract regression tests cover denied get/patch/delete for a non-project
member, the positive project-member path, and the workspace-level exemption;
fail-before verified.

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

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

@@ -18,7 +18,7 @@ from rest_framework.permissions import AllowAny
 
 # Module imports
 from ..base import BaseAPIView
-from plane.db.models import FileAsset, Workspace, Project, User, WorkspaceMember
+from plane.db.models import FileAsset, Workspace, Project, User, WorkspaceMember, ProjectMember
 from plane.settings.storage import S3Storage
 from plane.app.permissions import allow_permission, ROLE
 from plane.utils.cache import invalidate_cache_directly
@@ -312,6 +312,30 @@ class WorkspaceFileAssetEndpoint(BaseAPIView):
         else:
             return
 
+    def project_membership_denied(self, request, asset):
+        """Enforce project-level access on a workspace-scoped asset lookup.
+
+        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
+        exempt. Returns a 403 Response when access is denied, else None.
+        """
+        if asset.project_id is None:
+            return None
+        is_project_member = ProjectMember.objects.filter(
+            member=request.user,
+            project_id=asset.project_id,
+            is_active=True,
+        ).exists()
+        if not is_project_member:
+            return Response(
+                {"error": "You don't have access to this asset."},
+                status=status.HTTP_403_FORBIDDEN,
+            )
+        return None
+
     @allow_permission([ROLE.ADMIN, ROLE.MEMBER, ROLE.GUEST], level="WORKSPACE")
     def post(self, request, slug):
         name = sanitize_filename(request.data.get("name")) or "unnamed"
@@ -393,6 +417,10 @@ class WorkspaceFileAssetEndpoint(BaseAPIView):
     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
+        denied = self.project_membership_denied(request, asset)
+        if denied is not None:
+            return denied
         # get the storage metadata
         asset.is_uploaded = True
         # get the storage metadata
@@ -414,6 +442,10 @@ class WorkspaceFileAssetEndpoint(BaseAPIView):
     @allow_permission([ROLE.ADMIN, ROLE.MEMBER, ROLE.GUEST], level="WORKSPACE")
     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
+        denied = self.project_membership_denied(request, asset)
+        if denied is not None:
+            return denied
         asset.is_deleted = True
         asset.deleted_at = timezone.now()
         # get the entity and save the asset id for the request field
@@ -425,6 +457,10 @@ class WorkspaceFileAssetEndpoint(BaseAPIView):
     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
+        denied = self.project_membership_denied(request, asset)
+        if denied is not None:
+            return denied
 
         # Check if the asset is uploaded
         if not asset.is_uploaded:

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

@@ -0,0 +1,209 @@
+# Copyright (c) 2023-present Plane Software, Inc. and contributors
+# SPDX-License-Identifier: AGPL-3.0-only
+# See the LICENSE file for details.
+
+"""Contract tests for ``WorkspaceFileAssetEndpoint`` project-scoping.
+
+Regression coverage for GHSA-h7mc-p9mm-2r4w / GHSA-cjph-cgm5-8pw8 (WEB-8066),
+an incomplete fix of the GHSA-qw87 asset-IDOR cluster.
+
+The endpoint is authorized at the WORKSPACE level, so any workspace member or
+guest previously reached ``get``/``patch``/``delete`` for a project-bound asset
+(issue attachment / description, comment description, page description) even
+when they were not a member of that asset's project. The fix requires an active
+``ProjectMember`` of ``asset.project_id`` for project-bound assets, while
+leaving workspace-level assets (WORKSPACE_LOGO, USER_AVATAR, USER_COVER, whose
+``project_id`` is NULL) accessible to any workspace member.
+"""
+
+from unittest import mock
+from uuid import uuid4
+
+import pytest
+from rest_framework import status
+from rest_framework.test import APIClient
+
+from plane.db.models import (
+    FileAsset,
+    Project,
+    ProjectMember,
+    User,
+    WorkspaceMember,
+)
+
+S3_STORAGE_PATH = "plane.app.views.asset.v2.S3Storage"
+
+
+@pytest.fixture
+def project(db, workspace, create_user):
+    """A project in the fixture workspace; ``create_user`` is an active member."""
+    project = Project.objects.create(
+        name="Test Project",
+        identifier="TP",
+        workspace=workspace,
+        created_by=create_user,
+    )
+    ProjectMember.objects.create(
+        project=project, member=create_user, workspace=workspace, role=20
+    )
+    return project
+
+
+@pytest.fixture
+def outsider_user(db):
+    """A user who is a workspace member but NOT a member of ``project``."""
+    unique_id = uuid4().hex[:8]
+    user = User.objects.create(
+        email=f"outsider-{unique_id}@plane.so",
+        username=f"outsider_{unique_id}",
+        first_name="Outsider",
+        last_name="User",
+    )
+    user.set_password("test-password")
+    user.save()
+    return user
+
+
+@pytest.fixture
+def outsider_client(db, workspace, outsider_user):
+    """Session client for a workspace member who is not in ``project``."""
+    WorkspaceMember.objects.create(
+        workspace=workspace, member=outsider_user, role=15
+    )
+    client = APIClient()
+    client.force_authenticate(user=outsider_user)
+    return client
+
+
+@pytest.fixture
+def project_asset(db, workspace, project, create_user):
+    """An uploaded issue attachment that belongs to ``project``."""
+    return FileAsset.objects.create(
+        attributes={"name": "secret.pdf", "type": "application/pdf", "size": 1024},
+        asset=f"{workspace.id}/secret.pdf",
+        size=1024,
+        workspace=workspace,
+        project=project,
+        created_by=create_user,
+        entity_type=FileAsset.EntityTypeContext.ISSUE_ATTACHMENT,
+        is_uploaded=True,
+        storage_metadata={"size": 1024},
+    )
+
+
+@pytest.fixture
+def workspace_logo_asset(db, workspace, create_user):
+    """A workspace-level asset (project_id is NULL) — exempt from project scope."""
+    return FileAsset.objects.create(
+        attributes={"name": "logo.png", "type": "image/png", "size": 256},
+        asset=f"{workspace.id}/logo.png",
+        size=256,
+        workspace=workspace,
+        created_by=create_user,
+        entity_type=FileAsset.EntityTypeContext.WORKSPACE_LOGO,
+        is_uploaded=True,
+        storage_metadata={"size": 256},
+    )
+
+
+def detail_url(slug, asset_id):
+    return f"/api/assets/v2/workspaces/{slug}/{asset_id}/"
+
+
+@pytest.mark.contract
+class TestWorkspaceFileAssetProjectScope:
+    """A workspace member who is not in the asset's project must be blocked."""
+
+    @pytest.mark.django_db
+    def test_get_project_asset_denied_for_non_project_member(
+        self, outsider_client, workspace, project_asset
+    ):
+        """GET on a project asset by a non-project-member must 403, not mint a
+        presigned download URL."""
+        url = detail_url(workspace.slug, project_asset.id)
+
+        with mock.patch(S3_STORAGE_PATH) as mock_storage:
+            mock_storage.return_value.generate_presigned_url.return_value = (
+                "https://signed.example/download"
+            )
+            response = outsider_client.get(url)
+
+        assert response.status_code == status.HTTP_403_FORBIDDEN, (
+            f"Got {response.status_code}: {getattr(response, 'data', None)!r}"
+        )
+        mock_storage.return_value.generate_presigned_url.assert_not_called()
+
+    @pytest.mark.django_db
+    def test_patch_project_asset_denied_for_non_project_member(
+        self, outsider_client, workspace, project_asset
+    ):
+        """PATCH on a project asset by a non-project-member must 403 and leave
+        the asset untouched."""
+        url = detail_url(workspace.slug, project_asset.id)
+        project_asset.is_uploaded = False
+        project_asset.save(update_fields=["is_uploaded"])
+
+        response = outsider_client.patch(
+            url, {"attributes": {"name": "hacked.pdf"}}, format="json"
+        )
+
+        assert response.status_code == status.HTTP_403_FORBIDDEN, (
+            f"Got {response.status_code}: {getattr(response, 'data', None)!r}"
+        )
+        project_asset.refresh_from_db()
+        assert project_asset.is_uploaded is False
+        assert project_asset.attributes.get("name") == "secret.pdf"
+
+    @pytest.mark.django_db
+    def test_delete_project_asset_denied_for_non_project_member(
+        self, outsider_client, workspace, project_asset
+    ):
+        """DELETE on a project asset by a non-project-member must 403 and must
+        not soft-delete the asset."""
+        url = detail_url(workspace.slug, project_asset.id)
+
+        response = outsider_client.delete(url)
+
+        assert response.status_code == status.HTTP_403_FORBIDDEN, (
+            f"Got {response.status_code}: {getattr(response, 'data', None)!r}"
+        )
+        project_asset.refresh_from_db()
+        assert project_asset.is_deleted is False
+
+    @pytest.mark.django_db
+    def test_get_project_asset_allowed_for_project_member(
+        self, session_client, workspace, project_asset
+    ):
+        """Positive control: an active project member can still download the
+        asset, so the fix does not over-block legitimate callers."""
+        url = detail_url(workspace.slug, project_asset.id)
+
+        with mock.patch(S3_STORAGE_PATH) as mock_storage:
+            mock_storage.return_value.generate_presigned_url.return_value = (
+                "https://signed.example/download"
+            )
+            response = session_client.get(url)
+
+        assert response.status_code == status.HTTP_302_FOUND, (
+            f"Got {response.status_code}: {getattr(response, 'data', None)!r}"
+        )
+        mock_storage.return_value.generate_presigned_url.assert_called_once()
+
+    @pytest.mark.django_db
+    def test_get_workspace_level_asset_allowed_for_non_project_member(
+        self, outsider_client, workspace, workspace_logo_asset
+    ):
+        """Exemption control: a workspace-level asset (project_id NULL) stays
+        accessible to any workspace member."""
+        url = detail_url(workspace.slug, workspace_logo_asset.id)
+
+        with mock.patch(S3_STORAGE_PATH) as mock_storage:
+            mock_storage.return_value.generate_presigned_url.return_value = (
+                "https://signed.example/download"
+            )
+            response = outsider_client.get(url)
+
+        assert response.status_code == status.HTTP_302_FOUND, (
+            f"Got {response.status_code}: {getattr(response, 'data', None)!r}"
+        )
+        mock_storage.return_value.generate_presigned_url.assert_called_once()