From 83d5bda1ce042507dce5661f78a8e4106845b561 Mon Sep 17 00:00:00 2001 From: Manish Gupta Date: Fri, 21 Aug 2026 12:04:36 +0530 Subject: [PATCH 1/3] [INFRA-501] fix(security): enforce project membership on every workspace-level asset route The project-membership rule for asset access lived as a method on WorkspaceFileAssetEndpoint with three call sites. Every other asset route is a sibling BaseAPIView subclass and therefore could not reach it, so four workspace-level routes in the app and the whole external-API asset surface resolved assets on the workspace alone. A workspace member or guest belonging to none of the asset's projects could download another project's uploads, copy them into a project they controlled, reverse a deletion its owner performed, probe for asset ids, and flip is_uploaded to take an attachment offline. Move the rule onto the model as FileAsset.is_project_accessible_to so any surface that can load a FileAsset can ask the question, and a route added later cannot silently omit it -- which is how this gap arose. app/views/asset/v2.py - AssetRestoreEndpoint.post, AssetCheckEndpoint.get, WorkspaceAssetDownloadEndpoint.get and DuplicateAssetEndpoint.post now check the asset's project. Check answers exists=false rather than confirming a foreign asset, so it stops being a cross-project existence oracle. - DuplicateAssetEndpoint also requires active membership of the destination project from the request body; existence in the workspace was being treated as authorization. api/views/asset.py - GenericAssetEndpoint.get and .patch gained the same check. is_uploaded gates every download path, so an unscoped patch is a takedown primitive. - .post validates the body-supplied project_id against the URL workspace and the caller's membership; it was stored unvalidated, so a row in one workspace could point at a project in another -- exactly the inconsistency the access check has to defend against downstream. Also fixes three unconditional 500s in the same file: S3Storage.__init__ is (self, request=None) and never accepted is_server, so S3Storage(request=request, is_server=True) at :338, :451 and :581 raised TypeError for every caller. Passing no request is what selects the internal endpoint, making the keyword both wrong and redundant. This ships with the authorization checks on purpose: repairing the crash alone would have exposed a cross-project asset read on a route that currently only looks harmless. Contract tests cover each route from a non-member, a member, and a workspace-level asset whose project_id is NULL, so the fix cannot over-reach. The external-API positive paths patch S3Storage with autospec=True so the constructor signature is validated and the crash cannot regress unnoticed. Adds plane/tests/contract/api/conftest.py to reset the ApiKeyRateThrottle bucket around each external-API contract test. That throttle keys on the token string, which is a constant across the package, so all of those tests shared one 60/minute budget for the whole run. Adding tests here pushed the package past it and produced 429s in unrelated files. Only this throttle's key is cleared, following the existing narrowly-scoped auth-throttle helper rather than cache.clear(). Co-authored-by: Plane AI --- apps/api/plane/api/views/asset.py | 49 ++- apps/api/plane/app/views/asset/v2.py | 79 +++-- apps/api/plane/db/models/asset.py | 36 +++ apps/api/plane/tests/contract/api/conftest.py | 41 +++ .../api/test_generic_asset_project_scope.py | 297 +++++++++++++++++ ...orkspace_asset_routes_project_scope_app.py | 302 ++++++++++++++++++ 6 files changed, 771 insertions(+), 33 deletions(-) create mode 100644 apps/api/plane/tests/contract/api/conftest.py create mode 100644 apps/api/plane/tests/contract/api/test_generic_asset_project_scope.py create mode 100644 apps/api/plane/tests/contract/app/test_workspace_asset_routes_project_scope_app.py diff --git a/apps/api/plane/api/views/asset.py b/apps/api/plane/api/views/asset.py index abfa6bdc0d8..9804a53bc9c 100644 --- a/apps/api/plane/api/views/asset.py +++ b/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}" @@ -578,7 +610,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 +652,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) diff --git a/apps/api/plane/app/views/asset/v2.py b/apps/api/plane/app/views/asset/v2.py index 6d835f6a225..277d1d3bf06 100644 --- a/apps/api/plane/app/views/asset/v2.py +++ b/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, diff --git a/apps/api/plane/db/models/asset.py b/apps/api/plane/db/models/asset.py index 55efff7f41d..ebcabe9447a 100644 --- a/apps/api/plane/db/models/asset.py +++ b/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() diff --git a/apps/api/plane/tests/contract/api/conftest.py b/apps/api/plane/tests/contract/api/conftest.py new file mode 100644 index 00000000000..5b21d6e3639 --- /dev/null +++ b/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:`` 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() diff --git a/apps/api/plane/tests/contract/api/test_generic_asset_project_scope.py b/apps/api/plane/tests/contract/api/test_generic_asset_project_scope.py new file mode 100644 index 00000000000..d780016790c --- /dev/null +++ b/apps/api/plane/tests/contract/api/test_generic_asset_project_scope.py @@ -0,0 +1,297 @@ +# 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 project scoping on the external-API ``GenericAssetEndpoint`` (INFRA-501). + +The endpoint authorizes with ``WorkspaceUserPermission`` -- any active member of +the URL workspace, any role, including a GUEST who belongs to no project -- and +then resolves the asset on the workspace alone. The app surface enforces an +active ``ProjectMember`` of the asset's project for the same model; this surface +never did, so: + +* ``get`` discloses a project-bound asset's existence and upload state, and + (once the presigned stage is reachable) its contents +* ``patch`` flips ``is_uploaded``, which gates every download path, so an + attachment can be made to vanish for the project that owns it +* ``post`` stored a body-supplied ``project_id`` unvalidated, so a row could be + created in one workspace pointing at a project in another + +Both surfaces now share one rule, ``FileAsset.is_project_accessible_to``. + +Workspace-level assets (``project_id=None``) must stay reachable by any +workspace member, and a member of the asset's project must keep full access -- +both covered here so the fix cannot over-reach. + +The ``get``/``post`` positive paths patch ``S3Storage`` with ``autospec=True`` +deliberately: these call sites passed a keyword the constructor does not accept, +which raised ``TypeError`` and turned the routes into an unconditional 500 for +every caller. An autospec'd mock validates the call signature, so it fails if +that regresses; a plain mock would swallow it. +""" + +from unittest import mock +from uuid import uuid4 + +import pytest +from rest_framework import status + +from plane.db.models import ( + FileAsset, + Project, + ProjectMember, + User, + Workspace, + WorkspaceMember, +) + +S3_STORAGE_PATH = "plane.api.views.asset.S3Storage" + + +def asset_detail_url(slug, asset_id): + return f"/api/v1/workspaces/{slug}/assets/{asset_id}/" + + +def asset_list_url(slug): + return f"/api/v1/workspaces/{slug}/assets/" + + +def _user(prefix): + unique_id = uuid4().hex[:8] + user = User.objects.create( + email=f"{prefix}-{unique_id}@plane.so", + username=f"{prefix}_{unique_id}", + first_name=prefix.capitalize(), + last_name="User", + ) + user.set_password("test-password") + user.save() + return user + + +@pytest.fixture +def project_owner(db): + return _user("owner") + + +@pytest.fixture +def foreign_project(db, workspace, project_owner): + """A project in the caller's workspace that the caller is NOT a member of. + + ``create_user`` (who holds the API token) is a workspace ADMIN here, which is + the point: workspace role does not substitute for project membership, exactly + as the app surface already behaves. + """ + WorkspaceMember.objects.create( + workspace=workspace, member=project_owner, role=20, is_active=True + ) + project = Project.objects.create( + name="Foreign Project", identifier="FGN", workspace=workspace, created_by=project_owner + ) + ProjectMember.objects.create( + project=project, member=project_owner, workspace=workspace, role=20, is_active=True + ) + return project + + +@pytest.fixture +def joined_project(db, workspace, create_user): + """A project the token holder is an active member of.""" + project = Project.objects.create( + name="Joined Project", identifier="JND", workspace=workspace, created_by=create_user + ) + ProjectMember.objects.create( + project=project, member=create_user, workspace=workspace, role=20, is_active=True + ) + return project + + +def _asset(workspace, creator, project=None, name="secret.pdf"): + return FileAsset.objects.create( + attributes={"name": name, "type": "application/pdf", "size": 1024}, + asset=f"{workspace.id}/{uuid4().hex}-{name}", + size=1024, + workspace=workspace, + project=project, + created_by=creator, + entity_type=FileAsset.EntityTypeContext.ISSUE_ATTACHMENT, + is_uploaded=True, + storage_metadata={"size": 1024}, + ) + + +@pytest.fixture +def foreign_asset(db, workspace, foreign_project, project_owner): + return _asset(workspace, project_owner, foreign_project) + + +@pytest.fixture +def joined_asset(db, workspace, joined_project, create_user): + return _asset(workspace, create_user, joined_project, name="mine.pdf") + + +@pytest.fixture +def workspace_level_asset(db, workspace, create_user): + """project_id is NULL -- workspace-scoped by definition.""" + asset = _asset(workspace, create_user, None, name="logo.png") + asset.entity_type = FileAsset.EntityTypeContext.WORKSPACE_LOGO + asset.save(update_fields=["entity_type"]) + return asset + + +@pytest.fixture +def other_workspace_project(db): + """A project in an unrelated workspace, for the cross-tenant create case.""" + owner = _user("tenant") + other = Workspace.objects.create( + name="Other Workspace", owner=owner, slug=f"other-{uuid4().hex[:8]}" + ) + WorkspaceMember.objects.create(workspace=other, member=owner, role=20, is_active=True) + project = Project.objects.create( + name="Other Project", identifier="OTH", workspace=other, created_by=owner + ) + ProjectMember.objects.create( + project=project, member=owner, workspace=other, role=20, is_active=True + ) + return project + + +@pytest.mark.contract +class TestGenericAssetGetProjectScope: + @pytest.mark.django_db + def test_get_denied_for_non_project_member(self, api_key_client, workspace, foreign_asset): + with mock.patch(S3_STORAGE_PATH) as mock_storage: + # Return a real string, not the default MagicMock: if this guard ever + # regresses, the handler puts this value into a DRF Response, and JSON + # -encoding a MagicMock recurses until the process is OOM-killed. A + # regression must fail this assertion, not take the test runner down. + mock_storage.return_value.generate_presigned_url.return_value = "https://example.com/s" + response = api_key_client.get(asset_detail_url(workspace.slug, foreign_asset.id)) + + 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_get_does_not_leak_upload_state_of_foreign_asset( + self, api_key_client, workspace, foreign_asset + ): + """A not-yet-uploaded foreign asset must 403, not answer 400 'not uploaded'.""" + foreign_asset.is_uploaded = False + foreign_asset.save(update_fields=["is_uploaded"]) + + response = api_key_client.get(asset_detail_url(workspace.slug, foreign_asset.id)) + + assert response.status_code == status.HTTP_403_FORBIDDEN, ( + f"Got {response.status_code}: {getattr(response, 'data', None)!r}" + ) + + @pytest.mark.django_db + def test_get_allowed_for_project_member(self, api_key_client, workspace, joined_asset): + with mock.patch(S3_STORAGE_PATH, autospec=True) as mock_storage: + mock_storage.return_value.generate_presigned_url.return_value = "https://example.com/s" + response = api_key_client.get(asset_detail_url(workspace.slug, joined_asset.id)) + + assert response.status_code == status.HTTP_200_OK, ( + f"Got {response.status_code}: {getattr(response, 'data', None)!r}" + ) + assert response.data["asset_url"] == "https://example.com/s" + + @pytest.mark.django_db + def test_get_allowed_for_workspace_level_asset( + self, api_key_client, workspace, workspace_level_asset + ): + with mock.patch(S3_STORAGE_PATH, autospec=True) as mock_storage: + mock_storage.return_value.generate_presigned_url.return_value = "https://example.com/s" + response = api_key_client.get(asset_detail_url(workspace.slug, workspace_level_asset.id)) + + assert response.status_code == status.HTTP_200_OK, ( + f"Got {response.status_code}: {getattr(response, 'data', None)!r}" + ) + + +@pytest.mark.contract +class TestGenericAssetPatchProjectScope: + @pytest.mark.django_db + def test_patch_denied_for_non_project_member(self, api_key_client, workspace, foreign_asset): + """is_uploaded gates every download path, so this is a takedown primitive.""" + response = api_key_client.patch( + asset_detail_url(workspace.slug, foreign_asset.id), + {"is_uploaded": False}, + format="json", + ) + + assert response.status_code == status.HTTP_403_FORBIDDEN, ( + f"Got {response.status_code}: {getattr(response, 'data', None)!r}" + ) + foreign_asset.refresh_from_db() + assert foreign_asset.is_uploaded is True, "a foreign project's attachment was taken down" + + @pytest.mark.django_db + def test_patch_allowed_for_project_member(self, api_key_client, workspace, joined_asset): + response = api_key_client.patch( + asset_detail_url(workspace.slug, joined_asset.id), + {"is_uploaded": False}, + format="json", + ) + + assert response.status_code == status.HTTP_204_NO_CONTENT, ( + f"Got {response.status_code}: {getattr(response, 'data', None)!r}" + ) + joined_asset.refresh_from_db() + assert joined_asset.is_uploaded is False + + +@pytest.mark.contract +class TestGenericAssetPostProjectScope: + def _payload(self, project_id): + return { + "name": "planted.png", + "type": "image/png", + "size": 16, + "project_id": str(project_id), + } + + @pytest.mark.django_db + def test_post_denied_for_project_in_another_workspace( + self, api_key_client, workspace, other_workspace_project + ): + """The cross-tenant case: a row in workspace A pointing at a project of B.""" + before = FileAsset.objects.count() + response = api_key_client.post( + asset_list_url(workspace.slug), + self._payload(other_workspace_project.id), + format="json", + ) + + assert response.status_code == status.HTTP_404_NOT_FOUND, ( + f"Got {response.status_code}: {getattr(response, 'data', None)!r}" + ) + assert FileAsset.objects.count() == before, "a cross-workspace asset row was created" + + @pytest.mark.django_db + def test_post_denied_for_project_not_joined(self, api_key_client, workspace, foreign_project): + before = FileAsset.objects.count() + response = api_key_client.post( + asset_list_url(workspace.slug), self._payload(foreign_project.id), format="json" + ) + + assert response.status_code == status.HTTP_403_FORBIDDEN, ( + f"Got {response.status_code}: {getattr(response, 'data', None)!r}" + ) + assert FileAsset.objects.count() == before + + @pytest.mark.django_db + def test_post_allowed_for_joined_project(self, api_key_client, workspace, joined_project): + before = FileAsset.objects.count() + with mock.patch(S3_STORAGE_PATH, autospec=True) as mock_storage: + mock_storage.return_value.generate_presigned_post.return_value = {"url": "https://x"} + response = api_key_client.post( + asset_list_url(workspace.slug), self._payload(joined_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 diff --git a/apps/api/plane/tests/contract/app/test_workspace_asset_routes_project_scope_app.py b/apps/api/plane/tests/contract/app/test_workspace_asset_routes_project_scope_app.py new file mode 100644 index 00000000000..995d3ba9173 --- /dev/null +++ b/apps/api/plane/tests/contract/app/test_workspace_asset_routes_project_scope_app.py @@ -0,0 +1,302 @@ +# 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 project scoping on the workspace-level asset routes (INFRA-501). + +``WorkspaceFileAssetEndpoint`` already required an active ``ProjectMember`` of a +project-bound asset's project on get/patch/delete. Four sibling routes over the +same ``FileAsset`` model, behind the same ``level="WORKSPACE"`` authorization, +never received that check: + +* ``WorkspaceAssetDownloadEndpoint.get`` -- issues a presigned URL, i.e. the file +* ``DuplicateAssetEndpoint.post`` -- copies the asset into a caller-named project +* ``AssetRestoreEndpoint.post`` -- reverses a deletion the owner performed +* ``AssetCheckEndpoint.get`` -- existence oracle + +So a workspace member or GUEST who belonged to none of the asset's projects could +read, copy, undelete and probe another project's uploads. The rule now lives on +the model as ``FileAsset.is_project_accessible_to`` so a route added later cannot +silently omit it, which is how this gap arose in the first place. + +Workspace-level assets (workspace logo, user avatar/cover) carry +``project_id=None``, are workspace-scoped by definition, and must stay reachable +by any workspace member -- covered below so the fix cannot over-reach. +""" + +from unittest import mock +from uuid import uuid4 + +import pytest +from django.utils import timezone +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" + + +def download_url(slug, asset_id): + return f"/api/assets/v2/workspaces/{slug}/download/{asset_id}/" + + +def check_url(slug, asset_id): + return f"/api/assets/v2/workspaces/{slug}/check/{asset_id}/" + + +def restore_url(slug, asset_id): + return f"/api/assets/v2/workspaces/{slug}/restore/{asset_id}/" + + +def duplicate_url(slug, asset_id): + return f"/api/assets/v2/workspaces/{slug}/duplicate-assets/{asset_id}/" + + +def _user(prefix): + unique_id = uuid4().hex[:8] + user = User.objects.create( + email=f"{prefix}-{unique_id}@plane.so", + username=f"{prefix}_{unique_id}", + first_name=prefix.capitalize(), + last_name="User", + ) + user.set_password("test-password") + user.save() + return user + + +@pytest.fixture +def project(db, workspace, create_user): + """A project in the fixture workspace; ``create_user`` is an active ADMIN.""" + project = Project.objects.create( + name="Owner Project", identifier="OWN", workspace=workspace, created_by=create_user + ) + ProjectMember.objects.create( + project=project, member=create_user, workspace=workspace, role=20, is_active=True + ) + return project + + +@pytest.fixture +def outsider_user(db): + return _user("outsider") + + +@pytest.fixture +def outsider_client(db, workspace, outsider_user): + """A workspace GUEST who is a member of no project in the workspace.""" + WorkspaceMember.objects.create( + workspace=workspace, member=outsider_user, role=5, is_active=True + ) + client = APIClient() + client.force_authenticate(user=outsider_user) + return client + + +@pytest.fixture +def outsider_project(db, workspace, outsider_user): + """A project the outsider *does* control -- a duplicate destination.""" + project = Project.objects.create( + name="Outsider Project", identifier="OUT", workspace=workspace, created_by=outsider_user + ) + ProjectMember.objects.create( + project=project, member=outsider_user, workspace=workspace, role=20, is_active=True + ) + return project + + +@pytest.fixture +def project_asset(db, workspace, project, create_user): + """An uploaded issue attachment belonging 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 deleted_project_asset(db, project_asset): + """``project_asset`` after its own project deleted it.""" + project_asset.is_deleted = True + project_asset.deleted_at = timezone.now() + project_asset.save(update_fields=["is_deleted", "deleted_at"]) + return project_asset + + +@pytest.fixture +def workspace_logo_asset(db, workspace, create_user): + """A workspace-level asset -- ``project_id`` is NULL, so no project gate applies.""" + 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}, + ) + + +@pytest.mark.contract +class TestWorkspaceAssetDownloadProjectScope: + @pytest.mark.django_db + def test_download_denied_for_non_project_member(self, outsider_client, workspace, project_asset): + """A non-member must not be handed a presigned URL for the file.""" + with mock.patch(S3_STORAGE_PATH) as mock_storage: + # An explicit string, so a regression here fails the assertion rather + # than feeding a MagicMock into response rendering. + mock_storage.return_value.generate_presigned_url.return_value = "https://example.com/s" + response = outsider_client.get(download_url(workspace.slug, project_asset.id)) + + 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_download_allowed_for_project_member(self, session_client, workspace, project_asset): + with mock.patch(S3_STORAGE_PATH) as mock_storage: + mock_storage.return_value.generate_presigned_url.return_value = "https://example.com/s" + response = session_client.get(download_url(workspace.slug, project_asset.id)) + + assert response.status_code == status.HTTP_302_FOUND, ( + f"Got {response.status_code}: {getattr(response, 'data', None)!r}" + ) + + @pytest.mark.django_db + def test_download_workspace_level_asset_still_allowed( + self, outsider_client, workspace, workspace_logo_asset + ): + """project_id is NULL, so workspace membership alone must remain sufficient.""" + with mock.patch(S3_STORAGE_PATH) as mock_storage: + mock_storage.return_value.generate_presigned_url.return_value = "https://example.com/s" + response = outsider_client.get(download_url(workspace.slug, workspace_logo_asset.id)) + + assert response.status_code == status.HTTP_302_FOUND, ( + f"Got {response.status_code}: {getattr(response, 'data', None)!r}" + ) + + +@pytest.mark.contract +class TestAssetCheckProjectScope: + @pytest.mark.django_db + def test_check_does_not_disclose_existence_to_non_project_member( + self, outsider_client, workspace, project_asset + ): + """The oracle must answer False rather than confirming a foreign asset.""" + response = outsider_client.get(check_url(workspace.slug, project_asset.id)) + + assert response.status_code == status.HTTP_200_OK + assert response.data["exists"] is False, ( + "existence of another project's asset was disclosed to a non-member" + ) + + @pytest.mark.django_db + def test_check_reports_existence_to_project_member(self, session_client, workspace, project_asset): + response = session_client.get(check_url(workspace.slug, project_asset.id)) + + assert response.status_code == status.HTTP_200_OK + assert response.data["exists"] is True + + +@pytest.mark.contract +class TestAssetRestoreProjectScope: + @pytest.mark.django_db + def test_restore_denied_for_non_project_member( + self, outsider_client, workspace, deleted_project_asset + ): + """A non-member must not be able to reverse the owner's deletion.""" + response = outsider_client.post(restore_url(workspace.slug, deleted_project_asset.id)) + + assert response.status_code == status.HTTP_403_FORBIDDEN, ( + f"Got {response.status_code}: {getattr(response, 'data', None)!r}" + ) + deleted_project_asset.refresh_from_db() + assert deleted_project_asset.is_deleted is True + assert deleted_project_asset.deleted_at is not None + + @pytest.mark.django_db + def test_restore_allowed_for_project_member( + self, session_client, workspace, deleted_project_asset + ): + response = session_client.post(restore_url(workspace.slug, deleted_project_asset.id)) + + assert response.status_code == status.HTTP_204_NO_CONTENT, ( + f"Got {response.status_code}: {getattr(response, 'data', None)!r}" + ) + deleted_project_asset.refresh_from_db() + assert deleted_project_asset.is_deleted is False + + +@pytest.mark.contract +class TestDuplicateAssetProjectScope: + @pytest.mark.django_db + def test_duplicate_denied_when_source_not_accessible( + self, outsider_client, workspace, project_asset, outsider_project + ): + """The worst of the four: a permanent copy into a project the caller owns.""" + before = FileAsset.objects.count() + with mock.patch(S3_STORAGE_PATH) as mock_storage: + response = outsider_client.post( + duplicate_url(workspace.slug, project_asset.id), + { + "entity_type": FileAsset.EntityTypeContext.ISSUE_ATTACHMENT, + "project_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 copy of a foreign asset was created" + + @pytest.mark.django_db + def test_duplicate_denied_when_destination_project_not_joined( + self, session_client, workspace, project_asset, outsider_project + ): + """Source is reachable, destination is not: the body project_id needs its own check.""" + 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.ISSUE_ATTACHMENT, + "project_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 + + @pytest.mark.django_db + def test_duplicate_allowed_within_own_project(self, session_client, workspace, project, project_asset): + 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.ISSUE_ATTACHMENT, + "project_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 From 0effe3c2aed84f76c1b49e4af14fe3e4eae66867 Mon Sep 17 00:00:00 2001 From: Manish Gupta Date: Fri, 21 Aug 2026 13:06:53 +0530 Subject: [PATCH 2/3] [INFRA-501] fix(security): stop the external-id dedup echo disclosing foreign asset ids Review catch. GenericAssetEndpoint.post deduplicates on workspace + external_source + external_id with no project scoping, and answers a match with 409 carrying asset_id and asset_url. For an attachment, asset_url also embeds the owning project and issue ids. When the body omits project_id the create-path validation is skipped entirely, so this branch is the only gate. Knowing the asset UUID is the precondition for every asset-scoped attack on this surface, so the preceding commit closed the routes that consume a foreign id while leaving the path that hands it out -- in the same handler. Two of the reports this branch addresses name this echo as their id-recovery step. Answer 404 when the matched asset's project is not accessible, and keep the 409 echo for a match the caller can reach. 404 rather than 403 deliberately: a 403 would still confirm that some asset holds this external id pair in this workspace, turning the pair into an existence oracle. The 403 used elsewhere in this branch is fine on routes where the caller already named an 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 -- the right trade, since real integrations mint ids per source and run as a member of the target project. Contract tests: disclosure with project_id supplied, disclosure with project_id omitted, and two controls proving dedup still echoes for a project member and for a workspace-level asset whose project_id is NULL. Verified fail-before against the previous commit -- the negative case returned 409 with the foreign asset id and its project id in asset_url. Co-authored-by: Plane AI --- apps/api/plane/api/views/asset.py | 22 +++++ .../api/test_generic_asset_project_scope.py | 91 +++++++++++++++++++ 2 files changed, 113 insertions(+) diff --git a/apps/api/plane/api/views/asset.py b/apps/api/plane/api/views/asset.py index 9804a53bc9c..81451acd88a 100644 --- a/apps/api/plane/api/views/asset.py +++ b/apps/api/plane/api/views/asset.py @@ -587,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", diff --git a/apps/api/plane/tests/contract/api/test_generic_asset_project_scope.py b/apps/api/plane/tests/contract/api/test_generic_asset_project_scope.py index d780016790c..453e39292b7 100644 --- a/apps/api/plane/tests/contract/api/test_generic_asset_project_scope.py +++ b/apps/api/plane/tests/contract/api/test_generic_asset_project_scope.py @@ -295,3 +295,94 @@ def test_post_allowed_for_joined_project(self, api_key_client, workspace, joined f"Got {response.status_code}: {getattr(response, 'data', None)!r}" ) assert FileAsset.objects.count() == before + 1 + + +@pytest.mark.contract +class TestGenericAssetExternalIdDedupDisclosure: + """The 409 dedup echo must not hand out a foreign project's asset identifiers. + + The dedup lookup matches on workspace + external_source + external_id, with no + project scoping, and the 409 body carries ``asset_id`` and ``asset_url``. That + url embeds the owning project and issue ids for an attachment. Since knowing + the asset UUID is the precondition for every asset-scoped attack on this + surface, echoing a match the caller cannot access supplies exactly what the + other guards in this module exist to make useless. + """ + + EXTERNAL = {"external_id": "EXT-1", "external_source": "jira"} + + def _payload(self, project_id=None): + payload = {"name": "dedup.png", "type": "image/png", "size": 16, **self.EXTERNAL} + if project_id is not None: + payload["project_id"] = str(project_id) + return payload + + @pytest.mark.django_db + def test_dedup_does_not_disclose_foreign_asset_identifiers( + self, api_key_client, workspace, foreign_asset, joined_project + ): + """404, not 403: a 403 would still confirm the external id pair is taken.""" + foreign_asset.external_id = self.EXTERNAL["external_id"] + foreign_asset.external_source = self.EXTERNAL["external_source"] + foreign_asset.save(update_fields=["external_id", "external_source"]) + + response = api_key_client.post( + asset_list_url(workspace.slug), self._payload(joined_project.id), format="json" + ) + + assert response.status_code == status.HTTP_404_NOT_FOUND, ( + f"Got {response.status_code}: {getattr(response, 'data', None)!r}" + ) + body = str(getattr(response, "data", "")) + assert str(foreign_asset.id) not in body, "the foreign asset id was disclosed" + assert str(foreign_asset.project_id) not in body, "the foreign project id was disclosed" + + @pytest.mark.django_db + def test_dedup_does_not_disclose_when_project_id_is_omitted( + self, api_key_client, workspace, foreign_asset + ): + """Omitting project_id skips the create-path validation, so this branch is the only gate.""" + foreign_asset.external_id = self.EXTERNAL["external_id"] + foreign_asset.external_source = self.EXTERNAL["external_source"] + foreign_asset.save(update_fields=["external_id", "external_source"]) + + response = api_key_client.post(asset_list_url(workspace.slug), self._payload(), format="json") + + assert response.status_code == status.HTTP_404_NOT_FOUND, ( + f"Got {response.status_code}: {getattr(response, 'data', None)!r}" + ) + assert str(foreign_asset.id) not in str(getattr(response, "data", "")) + + @pytest.mark.django_db + def test_dedup_still_echoes_an_accessible_asset( + self, api_key_client, workspace, joined_asset, joined_project + ): + """Dedup must keep working for a caller who is a member of the match's project.""" + joined_asset.external_id = self.EXTERNAL["external_id"] + joined_asset.external_source = self.EXTERNAL["external_source"] + joined_asset.save(update_fields=["external_id", "external_source"]) + + response = api_key_client.post( + asset_list_url(workspace.slug), self._payload(joined_project.id), format="json" + ) + + assert response.status_code == status.HTTP_409_CONFLICT, ( + f"Got {response.status_code}: {getattr(response, 'data', None)!r}" + ) + assert response.data["asset_id"] == str(joined_asset.id) + + @pytest.mark.django_db + def test_dedup_still_echoes_a_workspace_level_asset( + self, api_key_client, workspace, workspace_level_asset + ): + """project_id is NULL, so there is no project dimension to gate on.""" + workspace_level_asset.external_id = self.EXTERNAL["external_id"] + workspace_level_asset.external_source = self.EXTERNAL["external_source"] + workspace_level_asset.save(update_fields=["external_id", "external_source"]) + + response = api_key_client.post(asset_list_url(workspace.slug), self._payload(), format="json") + + assert response.status_code == status.HTTP_409_CONFLICT, ( + f"Got {response.status_code}: {getattr(response, 'data', None)!r}" + ) + assert response.data["asset_id"] == str(workspace_level_asset.id) From 8d992aebf34ee3e21d0d8b58b4348ec987a327c6 Mon Sep 17 00:00:00 2001 From: Manish Gupta Date: Fri, 21 Aug 2026 14:24:12 +0530 Subject: [PATCH 3/3] [INFRA-501] test: pin absence of the asset identifier fields on a denied dedup match Review follow-up. The denial tests checked that the matched asset's UUIDs did not appear in the response body, which is weaker than it looks: asset_url is derived from entity_type, and its workspace-level form (/api/assets/v2/static//) carries no project id at all, so a substring check on the project UUID would not catch every shape of leak. Assert the asset_id and asset_url fields are absent outright, in a helper shared by both denial cases, and keep the UUID substring checks underneath it. The omitted-project_id case previously only checked the asset id, so it now covers the same ground as the case that supplies one. Both denial tests fail against the commit before the dedup fix; the two controls proving dedup still echoes for an accessible asset pass either way. Co-authored-by: Plane AI --- .../api/test_generic_asset_project_scope.py | 24 +++++++++++++++---- 1 file changed, 20 insertions(+), 4 deletions(-) diff --git a/apps/api/plane/tests/contract/api/test_generic_asset_project_scope.py b/apps/api/plane/tests/contract/api/test_generic_asset_project_scope.py index 453e39292b7..54ce292859d 100644 --- a/apps/api/plane/tests/contract/api/test_generic_asset_project_scope.py +++ b/apps/api/plane/tests/contract/api/test_generic_asset_project_scope.py @@ -317,6 +317,24 @@ def _payload(self, project_id=None): payload["project_id"] = str(project_id) return payload + @staticmethod + def _assert_discloses_nothing(response, foreign_asset): + """The denied response must carry no identifier for the matched asset. + + Asserting the keys are absent as well as the UUIDs: ``asset_url`` is + derived from ``entity_type``, and its workspace-level form + (``/api/assets/v2/static//``) carries no project id at all, so a + substring check on the project UUID alone would not catch every leak. + Absence of the field is the invariant worth pinning. + """ + data = getattr(response, "data", {}) or {} + assert "asset_url" not in data, "the foreign asset URL was disclosed" + assert "asset_id" not in data, "the foreign asset id was disclosed" + + body = str(data) + assert str(foreign_asset.id) not in body, "the foreign asset id leaked into the body" + assert str(foreign_asset.project_id) not in body, "the foreign project id leaked into the body" + @pytest.mark.django_db def test_dedup_does_not_disclose_foreign_asset_identifiers( self, api_key_client, workspace, foreign_asset, joined_project @@ -333,9 +351,7 @@ def test_dedup_does_not_disclose_foreign_asset_identifiers( assert response.status_code == status.HTTP_404_NOT_FOUND, ( f"Got {response.status_code}: {getattr(response, 'data', None)!r}" ) - body = str(getattr(response, "data", "")) - assert str(foreign_asset.id) not in body, "the foreign asset id was disclosed" - assert str(foreign_asset.project_id) not in body, "the foreign project id was disclosed" + self._assert_discloses_nothing(response, foreign_asset) @pytest.mark.django_db def test_dedup_does_not_disclose_when_project_id_is_omitted( @@ -351,7 +367,7 @@ def test_dedup_does_not_disclose_when_project_id_is_omitted( assert response.status_code == status.HTTP_404_NOT_FOUND, ( f"Got {response.status_code}: {getattr(response, 'data', None)!r}" ) - assert str(foreign_asset.id) not in str(getattr(response, "data", "")) + self._assert_discloses_nothing(response, foreign_asset) @pytest.mark.django_db def test_dedup_still_echoes_an_accessible_asset(