fix(api): enforce object-level authz on asset/attachment endpoints + WorkspaceAdmin API permissions

apps/api/plane/api/permissions/__init__.py

@@ -0,0 +1,5 @@
+# Copyright (c) 2023-present Plane Software, Inc. and contributors
+# SPDX-License-Identifier: AGPL-3.0-only
+# See the LICENSE file for details.
+
+from .workspace import WorkspaceAdminOnlyPermission, WorkspaceAdminWriteMemberReadPermission

apps/api/plane/api/permissions/workspace.py

@@ -0,0 +1,70 @@
+# Copyright (c) 2023-present Plane Software, Inc. and contributors
+# SPDX-License-Identifier: AGPL-3.0-only
+# See the LICENSE file for details.
+
+from rest_framework.permissions import BasePermission, SAFE_METHODS
+
+from plane.app.permissions import ROLE
+from plane.db.models import WorkspaceMember
+
+
+def get_workspace_slug(view):
+    """Resolve the workspace slug from the view, returning None when it is absent.
+
+    Accessing ``view.workspace_slug`` directly would raise ``AttributeError`` (and a
+    500) on a view that does not expose it; returning None lets the caller deny cleanly.
+    """
+    return getattr(view, "workspace_slug", None)
+
+
+class WorkspaceAdminOnlyPermission(BasePermission):
+    """
+    Permission class for external APIs that restricts access to workspace admins only.
+    """
+
+    def has_permission(self, request, view):
+        """Allow only active workspace admins."""
+        if request.user.is_anonymous:
+            return False
+
+        workspace_slug = get_workspace_slug(view)
+        if not workspace_slug:
+            return False
+
+        return WorkspaceMember.objects.filter(
+            member=request.user,
+            workspace__slug=workspace_slug,
+            role=ROLE.ADMIN.value,
+            is_active=True,
+        ).exists()
+
+
+class WorkspaceAdminWriteMemberReadPermission(BasePermission):
+    """
+    Permission class for external APIs that allows workspace members to read
+    but restricts write operations to workspace admins only.
+    """
+
+    def has_permission(self, request, view):
+        """Allow active members to read and restrict writes to active admins."""
+        if request.user.is_anonymous:
+            return False
+
+        workspace_slug = get_workspace_slug(view)
+        if not workspace_slug:
+            return False
+
+        if request.method in SAFE_METHODS:
+            return WorkspaceMember.objects.filter(
+                member=request.user,
+                workspace__slug=workspace_slug,
+                role__in=[ROLE.ADMIN.value, ROLE.MEMBER.value],
+                is_active=True,
+            ).exists()
+
+        return WorkspaceMember.objects.filter(
+            member=request.user,
+            workspace__slug=workspace_slug,
+            role=ROLE.ADMIN.value,
+            is_active=True,
+        ).exists()

apps/api/plane/api/views/issue.py

@@ -1767,6 +1767,7 @@ class IssueAttachmentListCreateAPIEndpoint(BaseAPIView):
 
     serializer_class = IssueAttachmentSerializer
     model = FileAsset
+    permission_classes = [ProjectEntityPermission]
     use_read_replica = True
 
     @issue_attachment_docs(
@@ -1982,6 +1983,7 @@ class IssueAttachmentDetailAPIEndpoint(BaseAPIView):
 
     serializer_class = IssueAttachmentSerializer
     model = FileAsset
+    permission_classes = [ProjectEntityPermission]
     use_read_replica = True
 
     @issue_attachment_docs(

apps/api/plane/app/permissions/base.py

@@ -33,7 +33,11 @@ def _wrapped_view(instance, request, *args, **kwargs):
                         status=status.HTTP_403_FORBIDDEN,
                     )
 
-                obj = model.objects.filter(id=kwargs["pk"], created_by=request.user).exists()
+                obj = model.objects.filter(
+                    id=kwargs["pk"],
+                    created_by=request.user,
+                    workspace__slug=kwargs["slug"],
+                ).exists()
                 if obj:
                     return view_func(instance, request, *args, **kwargs)
 

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

@@ -9,10 +9,22 @@
 
 # Module imports
 from ..base import BaseAPIView, BaseViewSet
-from plane.db.models import FileAsset, Workspace
+from plane.db.models import FileAsset, Workspace, WorkspaceMember
 from plane.app.serializers import FileAssetSerializer
 
 
+def _is_workspace_member(user, *, workspace_id=None, slug=None):
+    """Return True when the user is an active member of the given workspace."""
+    if user.is_anonymous:
+        return False
+    filters = {"member": user, "is_active": True}
+    if workspace_id is not None:
+        filters["workspace_id"] = workspace_id
+    if slug is not None:
+        filters["workspace__slug"] = slug
+    return WorkspaceMember.objects.filter(**filters).exists()
+
+
 class FileAssetEndpoint(BaseAPIView):
     parser_classes = (MultiPartParser, FormParser, JSONParser)
 
@@ -21,6 +33,11 @@ class FileAssetEndpoint(BaseAPIView):
     """
 
     def get(self, request, workspace_id, asset_key):
+        if not _is_workspace_member(request.user, workspace_id=workspace_id):
+            return Response(
+                {"error": "You don't have the required permissions."},
+                status=status.HTTP_403_FORBIDDEN,
+            )
         asset_key = str(workspace_id) + "/" + asset_key
         files = FileAsset.objects.filter(asset=asset_key)
         if files.exists():
@@ -33,6 +50,11 @@ def get(self, request, workspace_id, asset_key):
             )
 
     def post(self, request, slug):
+        if not _is_workspace_member(request.user, slug=slug):
+            return Response(
+                {"error": "You don't have the required permissions."},
+                status=status.HTTP_403_FORBIDDEN,
+            )
         serializer = FileAssetSerializer(data=request.data)
         if serializer.is_valid():
             # Get the workspace
@@ -42,6 +64,11 @@ def post(self, request, slug):
         return Response(serializer.errors, status=status.HTTP_400_BAD_REQUEST)
 
     def delete(self, request, workspace_id, asset_key):
+        if not _is_workspace_member(request.user, workspace_id=workspace_id):
+            return Response(
+                {"error": "You don't have the required permissions."},
+                status=status.HTTP_403_FORBIDDEN,
+            )
         asset_key = str(workspace_id) + "/" + asset_key
         file_asset = FileAsset.objects.get(asset=asset_key)
         file_asset.is_deleted = True
@@ -51,6 +78,11 @@ def delete(self, request, workspace_id, asset_key):
 
 class FileAssetViewSet(BaseViewSet):
     def restore(self, request, workspace_id, asset_key):
+        if not _is_workspace_member(request.user, workspace_id=workspace_id):
+            return Response(
+                {"error": "You don't have the required permissions."},
+                status=status.HTTP_403_FORBIDDEN,
+            )
         asset_key = str(workspace_id) + "/" + asset_key
         file_asset = FileAsset.objects.get(asset=asset_key)
         file_asset.is_deleted = False

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

@@ -18,7 +18,14 @@
 
 # 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
@@ -202,6 +209,52 @@ def delete(self, request, asset_id):
 class WorkspaceFileAssetEndpoint(BaseAPIView):
     """This endpoint is used to upload cover images/logos etc for workspace, projects and users."""
 
+    def is_workspace_admin(self, request, slug):
+        return WorkspaceMember.objects.filter(
+            member=request.user,
+            workspace__slug=slug,
+            role=ROLE.ADMIN.value,
+            is_active=True,
+        ).exists()
+
+    def is_project_admin(self, request, slug, project_id):
+        if not project_id:
+            return False
+        return ProjectMember.objects.filter(
+            member=request.user,
+            workspace__slug=slug,
+            project_id=project_id,
+            role=ROLE.ADMIN.value,
+            is_active=True,
+        ).exists()
+
+    def has_project_access(self, request, slug, project_id):
+        if not project_id:
+            return False
+        return ProjectMember.objects.filter(
+            member=request.user,
+            workspace__slug=slug,
+            project_id=project_id,
+            is_active=True,
+        ).exists()
+
+    def authorize_entity_mutation(self, request, slug, entity_type, project_id):
+        """Logo/cover mutations require admin on the owning scope. Returns an error
+        response when the caller is not authorized, otherwise None."""
+        if entity_type == FileAsset.EntityTypeContext.WORKSPACE_LOGO and not self.is_workspace_admin(request, slug):
+            return Response(
+                {"error": "You don't have the required permissions."},
+                status=status.HTTP_403_FORBIDDEN,
+            )
+        if entity_type == FileAsset.EntityTypeContext.PROJECT_COVER and not self.is_project_admin(
+            request, slug, project_id
+        ):
+            return Response(
+                {"error": "You don't have the required permissions."},
+                status=status.HTTP_403_FORBIDDEN,
+            )
+        return None
+
     def get_entity_id_field(self, entity_type, entity_id):
         # Workspace Logo
         if entity_type == FileAsset.EntityTypeContext.WORKSPACE_LOGO:
@@ -327,6 +380,16 @@ def post(self, request, slug):
                 status=status.HTTP_400_BAD_REQUEST,
             )
 
+        # Logo/cover assets may only be created by an admin of the owning scope
+        project_identifier = entity_identifier if entity_type == FileAsset.EntityTypeContext.PROJECT_COVER else None
+        if entity_type == FileAsset.EntityTypeContext.PROJECT_COVER and not Project.objects.filter(
+            id=project_identifier, workspace__slug=slug
+        ).exists():
+            return Response({"error": "Project not found"}, status=status.HTTP_404_NOT_FOUND)
+        unauthorized = self.authorize_entity_mutation(request, slug, entity_type, project_identifier)
+        if unauthorized:
+            return unauthorized
+
         # Check if the file type is allowed
         allowed_types = [
             "image/jpeg",
@@ -382,6 +445,10 @@ def post(self, request, slug):
     def patch(self, request, slug, asset_id):
         # get the asset id
         asset = FileAsset.objects.get(id=asset_id, workspace__slug=slug)
+        # Logo/cover mutations require admin on the owning scope
+        unauthorized = self.authorize_entity_mutation(request, slug, asset.entity_type, asset.project_id)
+        if unauthorized:
+            return unauthorized
         # get the storage metadata
         asset.is_uploaded = True
         # get the storage metadata
@@ -403,6 +470,10 @@ def patch(self, request, slug, asset_id):
     @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)
+        # Logo/cover mutations require admin on the owning scope
+        unauthorized = self.authorize_entity_mutation(request, slug, asset.entity_type, asset.project_id)
+        if unauthorized:
+            return unauthorized
         asset.is_deleted = True
         asset.deleted_at = timezone.now()
         # get the entity and save the asset id for the request field
@@ -415,6 +486,14 @@ def get(self, request, slug, asset_id):
         # get the asset id
         asset = FileAsset.objects.get(id=asset_id, workspace__slug=slug)
 
+        # Project-bound assets must not be readable by workspace members who are
+        # not members of the owning project (e.g. secret projects).
+        if asset.project_id and not self.has_project_access(request, slug, asset.project_id):
+            return Response(
+                {"error": "The requested asset could not be found."},
+                status=status.HTTP_404_NOT_FOUND,
+            )
+
         # Check if the asset is uploaded
         if not asset.is_uploaded:
             return Response(
@@ -646,8 +725,9 @@ def post(self, request, slug, project_id, entity_id):
         if not asset_ids:
             return Response({"error": "No asset ids provided."}, status=status.HTTP_400_BAD_REQUEST)
 
-        # get the asset id
-        assets = FileAsset.objects.filter(id__in=asset_ids, workspace__slug=slug)
+        # get the asset id — scope to the URL project so assets from sibling
+        # projects in the same workspace cannot be hijacked/reassigned.
+        assets = FileAsset.objects.filter(id__in=asset_ids, workspace__slug=slug, project_id=project_id)
 
         # Get the first asset
         asset = assets.first()
@@ -757,7 +837,8 @@ def post(self, request, slug, asset_id):
                 return Response({"error": "Project not found"}, status=status.HTTP_404_NOT_FOUND)
 
         storage = S3Storage(request=request)
-        # Scope the source asset lookup to workspaces the caller is a member of
+        # Scope the source asset lookup to workspaces the caller is a member of,
+        # so a known asset UUID cannot be copied out of another tenant.
         user_workspace_ids = WorkspaceMember.objects.filter(
             member=request.user,
             is_active=True,
@@ -771,6 +852,15 @@ def post(self, request, slug, asset_id):
         if not original_asset:
             return Response({"error": "Asset not found"}, status=status.HTTP_404_NOT_FOUND)
 
+        # If the source asset belongs to a project, the caller must be a member of
+        # that project (guards secret-project assets from cross-project duplication).
+        if original_asset.project_id and not ProjectMember.objects.filter(
+            member=request.user,
+            project_id=original_asset.project_id,
+            is_active=True,
+        ).exists():
+            return Response({"error": "Asset not found"}, status=status.HTTP_404_NOT_FOUND)
+
         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(
@@ -812,6 +902,22 @@ def get(self, request, slug, asset_id):
                 status=status.HTTP_404_NOT_FOUND,
             )
 
+        # Project-bound assets must not be downloadable by workspace members who
+        # are not members of the owning project (e.g. secret projects).
+        if (
+            asset.project_id
+            and not ProjectMember.objects.filter(
+                member=request.user,
+                workspace__slug=slug,
+                project_id=asset.project_id,
+                is_active=True,
+            ).exists()
+        ):
+            return Response(
+                {"error": "The requested asset could not be found."},
+                status=status.HTTP_404_NOT_FOUND,
+            )
+
         storage = S3Storage(request=request)
         signed_url = storage.generate_presigned_url(
             object_name=asset.asset.name,

apps/api/plane/app/views/issue/attachment.py

@@ -148,7 +148,7 @@ def post(self, request, slug, project_id, issue_id):
 
     @allow_permission([ROLE.ADMIN], creator=True, model=FileAsset)
     def delete(self, request, slug, project_id, issue_id, pk):
-        issue_attachment = FileAsset.objects.get(pk=pk, workspace__slug=slug, project_id=project_id)
+        issue_attachment = FileAsset.objects.get(pk=pk, workspace__slug=slug, project_id=project_id, issue_id=issue_id)
         issue_attachment.is_deleted = True
         issue_attachment.deleted_at = timezone.now()
         issue_attachment.save()
@@ -171,7 +171,7 @@ def delete(self, request, slug, project_id, issue_id, pk):
     def get(self, request, slug, project_id, issue_id, pk=None):
         if pk:
             # Get the asset
-            asset = FileAsset.objects.get(id=pk, workspace__slug=slug, project_id=project_id)
+            asset = FileAsset.objects.get(id=pk, workspace__slug=slug, project_id=project_id, issue_id=issue_id)
 
             # Check if the asset is uploaded
             if not asset.is_uploaded:
@@ -202,7 +202,7 @@ def get(self, request, slug, project_id, issue_id, pk=None):
 
     @allow_permission([ROLE.ADMIN, ROLE.MEMBER, ROLE.GUEST])
     def patch(self, request, slug, project_id, issue_id, pk):
-        issue_attachment = FileAsset.objects.get(pk=pk, workspace__slug=slug, project_id=project_id)
+        issue_attachment = FileAsset.objects.get(pk=pk, workspace__slug=slug, project_id=project_id, issue_id=issue_id)
         serializer = IssueAttachmentSerializer(issue_attachment)
 
         # Send this activity only if the attachment is not uploaded before