From 5faa618a24cb0e77b8d672cd39cc2bf5c531ccf8 Mon Sep 17 00:00:00 2001
From: kirtimanmishrazipstack <kirtiman.mishra@zipstack.com>
Date: Fri, 7 Aug 2026 19:08:29 +0530
Subject: [PATCH 1/5] UN-2651 [FIX] Show execution logs to group-shared and
 org-shared users

The executions list resolved visibility through direct memberships only,
so a deployment reached via a group share or shared_to_org opened fine
while its Logs page came back empty. Defer to each resource's own
for_user, which spans every sharing path the resource list itself honours
(owner, co-owner, direct share, group share, shared_to_org).

The per-execution logs and export endpoints had the mirrored problem: no
scoping at all. IsOwner sat in permission_classes but implements only
has_object_permission, which DRF never invokes on list/export, so any org
member holding an execution id could read and CSV-export its logs. Gate
the queryset on the executions the caller can see instead, and deny a
missing execution the same way as an inaccessible one so the response
does not confirm which ids exist.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
---
 .../tests/test_shared_execution_access.py     | 108 ++++++++++++++++++
 .../workflow_v2/execution_log_view.py         |  15 ++-
 .../workflow_v2/models/execution.py           |  24 ++--
 3 files changed, 129 insertions(+), 18 deletions(-)
 create mode 100644 backend/workflow_manager/execution/tests/test_shared_execution_access.py

diff --git a/backend/workflow_manager/execution/tests/test_shared_execution_access.py b/backend/workflow_manager/execution/tests/test_shared_execution_access.py
new file mode 100644
index 0000000000..c290435ae1
--- /dev/null
+++ b/backend/workflow_manager/execution/tests/test_shared_execution_access.py
@@ -0,0 +1,108 @@
+"""Executions and their logs follow the sharing paths of the resource they belong
+to — memberships, group shares and ``shared_to_org`` (UN-2651).
+"""
+
+import secrets
+from unittest.mock import patch
+
+from api_v2.models import APIDeployment
+from django.contrib.contenttypes.models import ContentType
+from django.utils import timezone
+from rest_framework.exceptions import PermissionDenied
+from rest_framework.test import APIRequestFactory
+from tenant_account_v2.models import ResourceGroupShare
+from tenant_account_v2.tests import GroupSharingTestBase
+
+from workflow_manager.workflow_v2.enums import ExecutionStatus
+from workflow_manager.workflow_v2.execution_log_view import WorkflowExecutionLogViewSet
+from workflow_manager.workflow_v2.models.execution import WorkflowExecution
+from workflow_manager.workflow_v2.models.execution_log import ExecutionLog
+
+_ADMIN_PREDICATE = (
+    "tenant_account_v2.organization_member_service."
+    "OrganizationMemberService.is_user_organization_admin"
+)
+
+
+class SharedExecutionAccessTests(GroupSharingTestBase):
+    """``self.member`` belongs to ``self.group``; ``self.outsider`` is an org
+    member with no share of any kind. Neither owns anything here.
+    """
+
+    def setUp(self) -> None:
+        super().setUp()
+        # Pin admin resolution: "admins see everything" would mask the paths here.
+        patcher = patch(_ADMIN_PREDICATE, return_value=False)
+        patcher.start()
+        self.addCleanup(patcher.stop)
+
+    def _api_deployment(self, *, shared_to_org: bool = False) -> APIDeployment:
+        # api_name must be short — it defaults to a UUID longer than the column.
+        return APIDeployment.objects.create(
+            api_name=f"api-{secrets.token_hex(4)}",
+            workflow=self.workflow,
+            organization=self.org,
+            created_by=self.owner,
+            shared_to_org=shared_to_org,
+        )
+
+    def _execution(self, deployment: APIDeployment) -> WorkflowExecution:
+        return WorkflowExecution.objects.create(
+            workflow=self.workflow,
+            pipeline_id=deployment.id,
+            execution_mode=WorkflowExecution.Mode.INSTANT,
+            execution_method=WorkflowExecution.Method.DIRECT,
+            execution_type=WorkflowExecution.Type.COMPLETE,
+            status=ExecutionStatus.COMPLETED,
+        )
+
+    def _share_with_group(self, deployment: APIDeployment) -> None:
+        ResourceGroupShare.objects.create(
+            group=self.group,
+            content_type=ContentType.objects.get_for_model(APIDeployment),
+            object_id=str(deployment.id),
+            organization=self.org,
+        )
+
+    def _visible_to(self, user, execution: WorkflowExecution) -> bool:
+        return (
+            WorkflowExecution.objects.for_user(user).filter(pk=execution.pk).exists()
+        )
+
+    def _log_queryset(self, user, execution_id):
+        """Run the log viewset's queryset build for ``user`` — the access gate."""
+        request = APIRequestFactory().get("/")
+        request.user = user
+        view = WorkflowExecutionLogViewSet()
+        view.request = request
+        view.kwargs = {"pk": str(execution_id)}
+        return view.get_queryset()
+
+    def test_org_wide_share_exposes_the_deployment_executions(self) -> None:
+        execution = self._execution(self._api_deployment(shared_to_org=True))
+        self.assertTrue(self._visible_to(self.outsider, execution))
+
+    def test_group_share_exposes_the_deployment_executions(self) -> None:
+        deployment = self._api_deployment()
+        self._share_with_group(deployment)
+        execution = self._execution(deployment)
+
+        self.assertTrue(self._visible_to(self.member, execution))
+        # Same org, not in the group — still nothing.
+        self.assertFalse(self._visible_to(self.outsider, execution))
+
+    def test_unshared_deployment_stays_invisible(self) -> None:
+        execution = self._execution(self._api_deployment())
+        self.assertFalse(self._visible_to(self.outsider, execution))
+
+    def test_logs_denied_when_the_execution_is_not_accessible(self) -> None:
+        execution = self._execution(self._api_deployment())
+        with self.assertRaises(PermissionDenied):
+            self._log_queryset(self.outsider, execution.id)
+
+    def test_logs_readable_once_the_deployment_is_shared(self) -> None:
+        execution = self._execution(self._api_deployment(shared_to_org=True))
+        log = ExecutionLog.objects.create(
+            wf_execution=execution, data={"log": "hello"}, event_time=timezone.now()
+        )
+        self.assertIn(log, list(self._log_queryset(self.outsider, execution.id)))
diff --git a/backend/workflow_manager/workflow_v2/execution_log_view.py b/backend/workflow_manager/workflow_v2/execution_log_view.py
index 2d8575080d..52a123dfe8 100644
--- a/backend/workflow_manager/workflow_v2/execution_log_view.py
+++ b/backend/workflow_manager/workflow_v2/execution_log_view.py
@@ -7,14 +7,15 @@
 from django.db.models.query import QuerySet
 from django.http import HttpResponse
 from django.utils import timezone
-from permissions.permission import IsOwner
 from rest_framework import status, viewsets
+from rest_framework.exceptions import PermissionDenied
 from rest_framework.permissions import IsAuthenticated
 from rest_framework.response import Response
 from rest_framework.versioning import URLPathVersioning
 from utils.pagination import CustomPagination
 
 from workflow_manager.workflow_v2.filters import ExecutionLogFilter
+from workflow_manager.workflow_v2.models.execution import WorkflowExecution
 from workflow_manager.workflow_v2.models.execution_log import ExecutionLog
 from workflow_manager.workflow_v2.serializers import WorkflowExecutionLogSerializer
 
@@ -28,7 +29,7 @@
 
 class WorkflowExecutionLogViewSet(viewsets.ModelViewSet):
     versioning_class = URLPathVersioning
-    permission_classes = [IsAuthenticated, IsOwner]
+    permission_classes = [IsAuthenticated]
     serializer_class = WorkflowExecutionLogSerializer
     pagination_class = CustomPagination
     ordering_fields = ["event_time"]
@@ -38,6 +39,16 @@ class WorkflowExecutionLogViewSet(viewsets.ModelViewSet):
     def get_queryset(self) -> QuerySet:
         execution_id = self.kwargs.get("pk")
 
+        # The URL's execution id is all that addresses these logs, so it is what
+        # gets authorized (UN-2651). Unknown ids are denied like inaccessible ones
+        # so the response does not reveal which ids exist.
+        if (
+            not WorkflowExecution.objects.for_user(self.request.user)
+            .filter(pk=execution_id)
+            .exists()
+        ):
+            raise PermissionDenied("You do not have access to logs for this execution.")
+
         # Query by execution_id for backward compatibility
         # Remove filter after execution_id is removed
         return ExecutionLog.objects.filter(
diff --git a/backend/workflow_manager/workflow_v2/models/execution.py b/backend/workflow_manager/workflow_v2/models/execution.py
index dc63a19ac7..f3e13188f4 100644
--- a/backend/workflow_manager/workflow_v2/models/execution.py
+++ b/backend/workflow_manager/workflow_v2/models/execution.py
@@ -9,7 +9,6 @@
 from pipeline_v2.models import Pipeline
 from tags.models import Tag
 from tenant_account_v2.organization_member_service import OrganizationMemberService
-from tenant_account_v2.sharing_helpers import resources_visible_via_memberships
 from usage_v2.constants import UsageKeys
 from usage_v2.helper import UsageHelper
 from usage_v2.models import Usage
@@ -35,8 +34,8 @@ def for_user(self, user) -> QuerySet:
         """Filter user's workflow executions with proper access control.
 
         Returns executions where the user has access to:
-        - The workflow (created by user OR shared with user) AND/OR
-        - The pipeline/API deployment (created by user OR shared with user)
+        - The workflow AND/OR
+        - The pipeline/API deployment
 
         This handles independent sharing scenarios:
         1. Workflow shared but not API deployment -> User can see workflow-only executions
@@ -64,23 +63,16 @@ def for_user(self, user) -> QuerySet:
                 return self.filter(workflow__organization=org)
             return self.all()
 
-        # Filter for workflow access (owner or direct viewer via membership).
-        # ``created_by`` is audit-only (UN-2202); VIEWER rows replaced shared_users.
-        # ``object_id`` is varchar, so resolve the ids via the cast helper rather
-        # than a ``memberships`` JOIN (Postgres refuses ``uuid = varchar``).
-        workflow_filter = Q(
-            workflow_id__in=resources_visible_via_memberships(Workflow, user)
-        )
+        # Defer to each resource's own ``for_user`` so execution visibility matches
+        # the resource list: memberships, group shares and ``shared_to_org``
+        # (UN-2651). Those managers are org-scoped, so no explicit org arg.
+        workflow_filter = Q(workflow_id__in=Workflow.objects.for_user(user).values("pk"))
 
         # Filter for API deployments the user can access
-        api_filter = Q(
-            pipeline_id__in=resources_visible_via_memberships(APIDeployment, user)
-        )
+        api_filter = Q(pipeline_id__in=APIDeployment.objects.for_user(user).values("pk"))
 
         # Filter for Pipelines the user can access
-        pipeline_filter = Q(
-            pipeline_id__in=resources_visible_via_memberships(Pipeline, user)
-        )
+        pipeline_filter = Q(pipeline_id__in=Pipeline.objects.for_user(user).values("pk"))
 
         # Combine deployment filters
         deployment_filter = api_filter | pipeline_filter

From ef3c01d43a28378e11b46d7165e7614abdb9a583 Mon Sep 17 00:00:00 2001
From: kirtimanmishrazipstack <kirtiman.mishra@zipstack.com>
Date: Tue, 11 Aug 2026 17:25:44 +0530
Subject: [PATCH 2/5] UN-2651 [FIX] Test that unknown execution ids are denied
 like inaccessible ones

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
---
 .../execution/tests/test_shared_execution_access.py        | 7 +++++++
 1 file changed, 7 insertions(+)

diff --git a/backend/workflow_manager/execution/tests/test_shared_execution_access.py b/backend/workflow_manager/execution/tests/test_shared_execution_access.py
index c290435ae1..7d3de069be 100644
--- a/backend/workflow_manager/execution/tests/test_shared_execution_access.py
+++ b/backend/workflow_manager/execution/tests/test_shared_execution_access.py
@@ -3,6 +3,7 @@
 """
 
 import secrets
+import uuid
 from unittest.mock import patch
 
 from api_v2.models import APIDeployment
@@ -100,6 +101,12 @@ def test_logs_denied_when_the_execution_is_not_accessible(self) -> None:
         with self.assertRaises(PermissionDenied):
             self._log_queryset(self.outsider, execution.id)
 
+    def test_logs_denied_when_the_execution_is_unknown(self) -> None:
+        # Same denial as an inaccessible execution — the response must not
+        # reveal which execution ids exist.
+        with self.assertRaises(PermissionDenied):
+            self._log_queryset(self.outsider, uuid.uuid4())
+
     def test_logs_readable_once_the_deployment_is_shared(self) -> None:
         execution = self._execution(self._api_deployment(shared_to_org=True))
         log = ExecutionLog.objects.create(

From 8e2649ca3ce74960d2257a5aa9de3a20a37fb07a Mon Sep 17 00:00:00 2001
From: kirtimanmishrazipstack <kirtiman.mishra@zipstack.com>
Date: Wed, 12 Aug 2026 11:32:41 +0530
Subject: [PATCH 3/5] UN-2651 [FIX] Pin the execution-log access gate at the
 HTTP layer
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Review follow-up on the sharing tests and the log viewset.

- Test fixtures built API deployments with no OWNER membership row, so they
  were visible to nobody and the negative test constrained nothing. The
  fixture now creates the row the deployment viewset creates, and asserts
  the owner sees the execution — which is what makes the denial a control.
- Executions were only ever built with pipeline_id set, leaving the
  workflow-level and Pipeline branches of for_user unexercised. _execution
  now takes an optional resource, with tests for both.
- Tests called get_queryset() directly, so the 403 and the whole export
  action were unverified. Both endpoints now go through as_view().
- Unknown and inaccessible execution ids are compared on status and body,
  which is what an enumerator observes.
- Added a cross-organization test: for_user is the only tenant boundary on
  /execution/, since the viewset replaces filter_backends.
- The log viewset is read-only; the queryset gate does not cover writes.
- Memoize the org-admin predicate on the user instance: for_user resolved
  it and then delegated to three managers that each re-resolved it.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
---
 .../organization_member_service.py            |  23 ++-
 .../tests/test_shared_execution_access.py     | 165 ++++++++++++++++--
 .../workflow_v2/execution_log_view.py         |   5 +-
 .../workflow_v2/models/execution.py           |   9 +-
 4 files changed, 180 insertions(+), 22 deletions(-)

diff --git a/backend/tenant_account_v2/organization_member_service.py b/backend/tenant_account_v2/organization_member_service.py
index 7ba92d9dbd..416456b143 100644
--- a/backend/tenant_account_v2/organization_member_service.py
+++ b/backend/tenant_account_v2/organization_member_service.py
@@ -7,6 +7,12 @@
 
 logger = logging.getLogger(__name__)
 
+# Memo attribute for the admin predicate, set on the ``User`` instance. Django
+# rebuilds that instance per request, so the memo lives exactly one request —
+# the same lifetime as the ``request``-keyed cache in ``permissions.permission``,
+# which the model managers cannot reach.
+_ADMIN_MEMO_ATTR = "_unstract_is_org_admin"
+
 
 class OrganizationMemberService:
     @staticmethod
@@ -23,11 +29,19 @@ def is_user_organization_admin(user: Any) -> bool:
         Service accounts are not org admins — they have their own bypass
         path in the relevant permissions / managers. Returns False on any
         lookup failure (anonymous user, no membership row, DB unavailable).
+
+        The result is memoized on ``user`` for the life of that instance:
+        ``WorkflowExecutionManager.for_user`` resolves this predicate and then
+        delegates to three resource managers that each re-resolve it, which was
+        four uncached membership lookups per call on a polled endpoint.
         """
         if not user or not getattr(user, "is_authenticated", False):
             return False
         if getattr(user, "is_service_account", False):
             return False
+        memo = getattr(user, _ADMIN_MEMO_ATTR, None)
+        if memo is not None:
+            return memo
         try:
             member = OrganizationMember.objects.get(user=user.id)  # type: ignore
         except OrganizationMember.DoesNotExist:
@@ -42,7 +56,14 @@ def is_user_organization_admin(user: Any) -> bool:
         # Delegate so admin-role string handling matches the active auth plugin.
         from account_v2.authentication_controller import AuthenticationController
 
-        return AuthenticationController().is_admin_by_role(member.role)
+        is_admin = AuthenticationController().is_admin_by_role(member.role)
+        # Failure paths above deliberately stay uncached — a transient DB error
+        # must not pin this user to "not an admin" for the rest of the request.
+        try:
+            setattr(user, _ADMIN_MEMO_ATTR, is_admin)
+        except AttributeError:
+            pass  # Immutable user object (e.g. AnonymousUser subclass) — skip the memo.
+        return is_admin
 
     @staticmethod
     def get_user_by_user_id(user_id: str) -> OrganizationMember | None:
diff --git a/backend/workflow_manager/execution/tests/test_shared_execution_access.py b/backend/workflow_manager/execution/tests/test_shared_execution_access.py
index 7d3de069be..fc2b300e04 100644
--- a/backend/workflow_manager/execution/tests/test_shared_execution_access.py
+++ b/backend/workflow_manager/execution/tests/test_shared_execution_access.py
@@ -6,18 +6,22 @@
 import uuid
 from unittest.mock import patch
 
+from account_v2.models import Organization
 from api_v2.models import APIDeployment
 from django.contrib.contenttypes.models import ContentType
 from django.utils import timezone
+from permissions.roles import ResourceRole
+from pipeline_v2.models import Pipeline
 from rest_framework.exceptions import PermissionDenied
-from rest_framework.test import APIRequestFactory
+from rest_framework.test import APIRequestFactory, force_authenticate
 from tenant_account_v2.models import ResourceGroupShare
-from tenant_account_v2.tests import GroupSharingTestBase
+from tenant_account_v2.tests import GroupSharingTestBase, _add_viewers
 
 from workflow_manager.workflow_v2.enums import ExecutionStatus
 from workflow_manager.workflow_v2.execution_log_view import WorkflowExecutionLogViewSet
 from workflow_manager.workflow_v2.models.execution import WorkflowExecution
 from workflow_manager.workflow_v2.models.execution_log import ExecutionLog
+from workflow_manager.workflow_v2.models.workflow import Workflow
 
 _ADMIN_PREDICATE = (
     "tenant_account_v2.organization_member_service."
@@ -27,7 +31,7 @@
 
 class SharedExecutionAccessTests(GroupSharingTestBase):
     """``self.member`` belongs to ``self.group``; ``self.outsider`` is an org
-    member with no share of any kind. Neither owns anything here.
+    member with no share of any kind. ``self.owner`` owns every fixture here.
     """
 
     def setUp(self) -> None:
@@ -39,29 +43,48 @@ def setUp(self) -> None:
 
     def _api_deployment(self, *, shared_to_org: bool = False) -> APIDeployment:
         # api_name must be short — it defaults to a UUID longer than the column.
-        return APIDeployment.objects.create(
+        deployment = APIDeployment.objects.create(
             api_name=f"api-{secrets.token_hex(4)}",
             workflow=self.workflow,
             organization=self.org,
             created_by=self.owner,
             shared_to_org=shared_to_org,
         )
-
-    def _execution(self, deployment: APIDeployment) -> WorkflowExecution:
+        # Creator access flows through an OWNER row, not ``created_by`` (UN-2202);
+        # mirrors what ``APIDeploymentViewSet.perform_create`` does.
+        deployment.memberships.create(user=self.owner, role=ResourceRole.OWNER)
+        return deployment
+
+    def _pipeline(self, *, shared_to_org: bool = False) -> Pipeline:
+        pipeline = Pipeline.objects.create(
+            pipeline_name=f"pipe-{secrets.token_hex(4)}",
+            workflow=self.workflow,
+            organization=self.org,
+            created_by=self.owner,
+            shared_to_org=shared_to_org,
+        )
+        pipeline.memberships.create(user=self.owner, role=ResourceRole.OWNER)
+        return pipeline
+
+    def _execution(self, resource=None) -> WorkflowExecution:
+        """``resource=None`` builds a workflow-level execution (``pipeline_id``
+        NULL), which is the only shape that exercises the workflow branch of
+        ``for_user``.
+        """
         return WorkflowExecution.objects.create(
             workflow=self.workflow,
-            pipeline_id=deployment.id,
+            pipeline_id=resource.id if resource else None,
             execution_mode=WorkflowExecution.Mode.INSTANT,
             execution_method=WorkflowExecution.Method.DIRECT,
             execution_type=WorkflowExecution.Type.COMPLETE,
             status=ExecutionStatus.COMPLETED,
         )
 
-    def _share_with_group(self, deployment: APIDeployment) -> None:
+    def _share_with_group(self, resource) -> None:
         ResourceGroupShare.objects.create(
             group=self.group,
-            content_type=ContentType.objects.get_for_model(APIDeployment),
-            object_id=str(deployment.id),
+            content_type=ContentType.objects.get_for_model(type(resource)),
+            object_id=str(resource.pk),
             organization=self.org,
         )
 
@@ -79,6 +102,33 @@ def _log_queryset(self, user, execution_id):
         view.kwargs = {"pk": str(execution_id)}
         return view.get_queryset()
 
+    def _call(self, action: str, user, execution_id, **query):
+        """Drive the real endpoint, so ``dispatch`` translates the gate to HTTP."""
+        view = WorkflowExecutionLogViewSet.as_view({"get": action})
+        request = APIRequestFactory().get("/", query)
+        force_authenticate(request, user=user)
+        return view(request, pk=str(execution_id))
+
+    # --- visibility: who sees which executions ---------------------------------
+
+    def test_unshared_deployment_is_visible_only_to_its_owner(self) -> None:
+        execution = self._execution(self._api_deployment())
+        # The owner assertion is what makes the denial below a real control:
+        # without the OWNER membership row the deployment would be visible to
+        # nobody, and the denial would pass with the sharing filter deleted.
+        self.assertTrue(self._visible_to(self.owner, execution))
+        self.assertFalse(self._visible_to(self.outsider, execution))
+
+    def test_direct_viewer_sees_the_deployment_executions(self) -> None:
+        deployment = self._api_deployment()
+        _add_viewers(deployment, self.outsider)
+        self.assertTrue(self._visible_to(self.outsider, self._execution(deployment)))
+
+    def test_co_owner_sees_the_deployment_executions(self) -> None:
+        deployment = self._api_deployment()
+        deployment.memberships.create(user=self.member, role=ResourceRole.OWNER)
+        self.assertTrue(self._visible_to(self.member, self._execution(deployment)))
+
     def test_org_wide_share_exposes_the_deployment_executions(self) -> None:
         execution = self._execution(self._api_deployment(shared_to_org=True))
         self.assertTrue(self._visible_to(self.outsider, execution))
@@ -92,20 +142,61 @@ def test_group_share_exposes_the_deployment_executions(self) -> None:
         # Same org, not in the group — still nothing.
         self.assertFalse(self._visible_to(self.outsider, execution))
 
-    def test_unshared_deployment_stays_invisible(self) -> None:
-        execution = self._execution(self._api_deployment())
+    def test_workflow_level_execution_follows_the_workflow_share(self) -> None:
+        execution = self._execution()  # pipeline_id NULL
+        self.assertTrue(self._visible_to(self.owner, execution))
+        self.assertFalse(self._visible_to(self.member, execution))
+
+        self._share_with_group(self.workflow)
+        self.assertTrue(self._visible_to(self.member, execution))
         self.assertFalse(self._visible_to(self.outsider, execution))
 
-    def test_logs_denied_when_the_execution_is_not_accessible(self) -> None:
-        execution = self._execution(self._api_deployment())
+    def test_pipeline_execution_follows_the_pipeline_share(self) -> None:
+        private = self._execution(self._pipeline())
+        self.assertTrue(self._visible_to(self.owner, private))
+        self.assertFalse(self._visible_to(self.outsider, private))
+
+        shared = self._execution(self._pipeline(shared_to_org=True))
+        self.assertTrue(self._visible_to(self.outsider, shared))
+
+    def test_org_wide_share_does_not_cross_organizations(self) -> None:
+        """``shared_to_org`` means *this* org — the tenant boundary for
+        ``/execution/`` is the manager, since the view drops the org filter
+        backend.
+        """
+        other_org = Organization.objects.create(
+            name="org-b", display_name="Org B", organization_id="org-b"
+        )
+        other_workflow = Workflow.objects.create(
+            workflow_name="wf-b", organization=other_org, created_by=self.owner
+        )
+        foreign = APIDeployment.objects.create(
+            api_name=f"api-{secrets.token_hex(4)}",
+            workflow=other_workflow,
+            organization=other_org,
+            created_by=self.owner,
+            shared_to_org=True,
+        )
+        execution = WorkflowExecution.objects.create(
+            workflow=other_workflow,
+            pipeline_id=foreign.id,
+            execution_mode=WorkflowExecution.Mode.INSTANT,
+            execution_method=WorkflowExecution.Method.DIRECT,
+            execution_type=WorkflowExecution.Type.COMPLETE,
+            status=ExecutionStatus.COMPLETED,
+        )
+
+        # UserContext still points at org A throughout.
+        self.assertFalse(self._visible_to(self.outsider, execution))
         with self.assertRaises(PermissionDenied):
             self._log_queryset(self.outsider, execution.id)
 
-    def test_logs_denied_when_the_execution_is_unknown(self) -> None:
-        # Same denial as an inaccessible execution — the response must not
-        # reveal which execution ids exist.
+    # --- the log endpoints ------------------------------------------------------
+
+    def test_logs_denied_when_the_execution_is_not_accessible(self) -> None:
+        execution = self._execution(self._api_deployment())
         with self.assertRaises(PermissionDenied):
-            self._log_queryset(self.outsider, uuid.uuid4())
+            self._log_queryset(self.outsider, execution.id)
 
     def test_logs_readable_once_the_deployment_is_shared(self) -> None:
         execution = self._execution(self._api_deployment(shared_to_org=True))
@@ -113,3 +204,41 @@ def test_logs_readable_once_the_deployment_is_shared(self) -> None:
             wf_execution=execution, data={"log": "hello"}, event_time=timezone.now()
         )
         self.assertIn(log, list(self._log_queryset(self.outsider, execution.id)))
+
+    def test_both_endpoints_return_403_for_an_inaccessible_execution(self) -> None:
+        execution = self._execution(self._api_deployment())
+        for action in ("list", "export"):
+            with self.subTest(action=action):
+                response = self._call(action, self.outsider, execution.id)
+                self.assertEqual(response.status_code, 403)
+
+    def test_both_endpoints_serve_a_shared_execution(self) -> None:
+        execution = self._execution(self._api_deployment(shared_to_org=True))
+        ExecutionLog.objects.create(
+            wf_execution=execution, data={"log": "hello"}, event_time=timezone.now()
+        )
+
+        listed = self._call("list", self.outsider, execution.id)
+        self.assertEqual(listed.status_code, 200)
+        listed.render()
+        self.assertIn(b"hello", listed.content)
+
+        exported = self._call(
+            "export", self.outsider, execution.id, file_format="csv"
+        )
+        self.assertEqual(exported.status_code, 200)
+        self.assertIn(b"hello", exported.content)
+
+    def test_unknown_execution_is_indistinguishable_from_an_inaccessible_one(
+        self,
+    ) -> None:
+        """An enumerator observes the response, not the exception — so the
+        status and body must match, not merely the exception type.
+        """
+        inaccessible = self._execution(self._api_deployment())
+        denied = self._call("list", self.outsider, inaccessible.id)
+        unknown = self._call("list", self.outsider, uuid.uuid4())
+
+        self.assertEqual(denied.status_code, 403)
+        self.assertEqual(unknown.status_code, denied.status_code)
+        self.assertEqual(unknown.data, denied.data)
diff --git a/backend/workflow_manager/workflow_v2/execution_log_view.py b/backend/workflow_manager/workflow_v2/execution_log_view.py
index 52a123dfe8..0cb904d852 100644
--- a/backend/workflow_manager/workflow_v2/execution_log_view.py
+++ b/backend/workflow_manager/workflow_v2/execution_log_view.py
@@ -27,7 +27,10 @@
 MAX_SYNC_EXPORT_ROWS = 50_000
 
 
-class WorkflowExecutionLogViewSet(viewsets.ModelViewSet):
+class WorkflowExecutionLogViewSet(viewsets.ReadOnlyModelViewSet):
+    # Read-only on purpose: the access gate lives in ``get_queryset``, which write
+    # handlers never call. ``ExecutionLog`` rows are written by workers and their
+    # fields are ``editable=False``, so there is nothing to expose.
     versioning_class = URLPathVersioning
     permission_classes = [IsAuthenticated]
     serializer_class = WorkflowExecutionLogSerializer
diff --git a/backend/workflow_manager/workflow_v2/models/execution.py b/backend/workflow_manager/workflow_v2/models/execution.py
index f3e13188f4..b7526e1651 100644
--- a/backend/workflow_manager/workflow_v2/models/execution.py
+++ b/backend/workflow_manager/workflow_v2/models/execution.py
@@ -43,7 +43,10 @@ def for_user(self, user) -> QuerySet:
         3. Both shared -> User can see all executions
         4. Neither shared -> User cannot see executions
 
-        Service accounts see all executions (org-scoped by view).
+        Service accounts and org admins see every execution in the current
+        organization. That org scoping is enforced here, not by the view:
+        ``ExecutionViewSet`` replaces ``filter_backends``, which drops
+        ``OrganizationFilterBackend``.
 
         Args:
             user: The user to filter executions for
@@ -65,7 +68,9 @@ def for_user(self, user) -> QuerySet:
 
         # Defer to each resource's own ``for_user`` so execution visibility matches
         # the resource list: memberships, group shares and ``shared_to_org``
-        # (UN-2651). Those managers are org-scoped, so no explicit org arg.
+        # (UN-2651). Those managers org-scope themselves via ``UserContext``, so
+        # this is correct on request paths only — a worker or management command
+        # with no org context gets an empty queryset (fail-closed).
         workflow_filter = Q(workflow_id__in=Workflow.objects.for_user(user).values("pk"))
 
         # Filter for API deployments the user can access

From 92a0651679f97b9a715a1aa6da07633c423a30bf Mon Sep 17 00:00:00 2001
From: kirtimanmishrazipstack <kirtiman.mishra@zipstack.com>
Date: Wed, 12 Aug 2026 14:23:20 +0530
Subject: [PATCH 4/5] UN-2651 [FIX] Gate every execution/<id>/ route behind one
 shared check
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

The logs endpoint was gated on its own, but `<id>/files/` takes the same
execution id, backs the same screen, and returned file names, per-file errors
and the latest log line to any authenticated org member. `/workflow/<id>/
execution/` carried the same dead `IsOwner` this PR retires elsewhere.

Both now go through `assert_execution_accessible`, which also logs the denial
with the user id — the server is the only place left that can tell an unknown
id from an inaccessible one, since the response deliberately cannot.

Also: fail closed rather than returning every tenant's executions when no
organization is in context; key the admin memo by organization, since the
predicate it caches is org-dependent; and correct two comments that described
the inverse of the mechanism they sat on.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
---
 .../organization_member_service.py            |  22 +--
 backend/workflow_manager/execution/access.py  |  37 +++++
 .../tests/test_shared_execution_access.py     | 147 +++++++++++++++++-
 .../workflow_manager/file_execution/views.py  |   6 +
 .../workflow_v2/execution_log_view.py         |  32 ++--
 .../workflow_v2/execution_view.py             |  17 +-
 .../workflow_v2/models/execution.py           |  34 ++--
 7 files changed, 250 insertions(+), 45 deletions(-)
 create mode 100644 backend/workflow_manager/execution/access.py

diff --git a/backend/tenant_account_v2/organization_member_service.py b/backend/tenant_account_v2/organization_member_service.py
index 416456b143..ed29d767ab 100644
--- a/backend/tenant_account_v2/organization_member_service.py
+++ b/backend/tenant_account_v2/organization_member_service.py
@@ -2,15 +2,17 @@
 from typing import Any
 
 from utils.cache_service import CacheService
+from utils.user_context import UserContext
 
 from tenant_account_v2.models import OrganizationMember
 
 logger = logging.getLogger(__name__)
 
-# Memo attribute for the admin predicate, set on the ``User`` instance. Django
-# rebuilds that instance per request, so the memo lives exactly one request —
-# the same lifetime as the ``request``-keyed cache in ``permissions.permission``,
-# which the model managers cannot reach.
+# Memo for the admin predicate, set on the ``User`` instance as
+# ``(organization_identifier, is_admin)``. The answer is a function of the user
+# *and* the current org — ``OrganizationMember.objects`` is org-scoped — and
+# ``set_user_organization`` can move a live ``request.user`` between orgs, so the
+# org id is part of the key rather than an assumption about instance lifetime.
 _ADMIN_MEMO_ATTR = "_unstract_is_org_admin"
 
 
@@ -30,7 +32,7 @@ def is_user_organization_admin(user: Any) -> bool:
         path in the relevant permissions / managers. Returns False on any
         lookup failure (anonymous user, no membership row, DB unavailable).
 
-        The result is memoized on ``user`` for the life of that instance:
+        The result is memoized on ``user``, keyed by the current organization:
         ``WorkflowExecutionManager.for_user`` resolves this predicate and then
         delegates to three resource managers that each re-resolve it, which was
         four uncached membership lookups per call on a polled endpoint.
@@ -39,9 +41,10 @@ def is_user_organization_admin(user: Any) -> bool:
             return False
         if getattr(user, "is_service_account", False):
             return False
+        org_id = UserContext.get_organization_identifier()
         memo = getattr(user, _ADMIN_MEMO_ATTR, None)
-        if memo is not None:
-            return memo
+        if memo is not None and memo[0] == org_id:
+            return memo[1]
         try:
             member = OrganizationMember.objects.get(user=user.id)  # type: ignore
         except OrganizationMember.DoesNotExist:
@@ -59,10 +62,7 @@ def is_user_organization_admin(user: Any) -> bool:
         is_admin = AuthenticationController().is_admin_by_role(member.role)
         # Failure paths above deliberately stay uncached — a transient DB error
         # must not pin this user to "not an admin" for the rest of the request.
-        try:
-            setattr(user, _ADMIN_MEMO_ATTR, is_admin)
-        except AttributeError:
-            pass  # Immutable user object (e.g. AnonymousUser subclass) — skip the memo.
+        setattr(user, _ADMIN_MEMO_ATTR, (org_id, is_admin))
         return is_admin
 
     @staticmethod
diff --git a/backend/workflow_manager/execution/access.py b/backend/workflow_manager/execution/access.py
new file mode 100644
index 0000000000..5a30012dd8
--- /dev/null
+++ b/backend/workflow_manager/execution/access.py
@@ -0,0 +1,37 @@
+"""Shared access gate for the ``execution/<pk>/...`` routes (UN-2651).
+
+Every route under that prefix is addressed by a ``WorkflowExecution`` id and
+nothing else, so the id is what has to be authorized. The logs endpoint and the
+file-execution endpoint back the same screen for the same id — keeping one gate
+is what stops them drifting apart again.
+"""
+
+import logging
+
+from rest_framework.exceptions import PermissionDenied
+from utils.user_context import UserContext
+
+from workflow_manager.workflow_v2.models.execution import WorkflowExecution
+
+logger = logging.getLogger(__name__)
+
+
+def assert_execution_accessible(user, execution_id) -> None:
+    """Raise ``PermissionDenied`` unless ``user`` may read ``execution_id``.
+
+    Unknown ids are denied exactly like inaccessible ones, so the response is
+    not an existence oracle. The two are still told apart in the log line —
+    the server is now the only place that can, and the distinction is what
+    separates a stale bookmark from someone walking the id space.
+    """
+    if WorkflowExecution.objects.for_user(user).filter(pk=execution_id).exists():
+        return
+
+    logger.warning(
+        "Execution access denied: user=%s execution=%s org=%s exists=%s",
+        getattr(user, "id", None),
+        execution_id,
+        UserContext.get_organization_identifier(),
+        WorkflowExecution.objects.filter(pk=execution_id).exists(),
+    )
+    raise PermissionDenied("You do not have access to this execution.")
diff --git a/backend/workflow_manager/execution/tests/test_shared_execution_access.py b/backend/workflow_manager/execution/tests/test_shared_execution_access.py
index fc2b300e04..7198a8435a 100644
--- a/backend/workflow_manager/execution/tests/test_shared_execution_access.py
+++ b/backend/workflow_manager/execution/tests/test_shared_execution_access.py
@@ -15,10 +15,13 @@
 from rest_framework.exceptions import PermissionDenied
 from rest_framework.test import APIRequestFactory, force_authenticate
 from tenant_account_v2.models import ResourceGroupShare
-from tenant_account_v2.tests import GroupSharingTestBase, _add_viewers
+from tenant_account_v2.tests import GroupSharingTestBase, _add_viewers, _make_user
+from utils.user_context import UserContext
 
+from workflow_manager.file_execution.views import FileCentricExecutionViewSet
 from workflow_manager.workflow_v2.enums import ExecutionStatus
 from workflow_manager.workflow_v2.execution_log_view import WorkflowExecutionLogViewSet
+from workflow_manager.workflow_v2.execution_view import WorkflowExecutionViewSet
 from workflow_manager.workflow_v2.models.execution import WorkflowExecution
 from workflow_manager.workflow_v2.models.execution_log import ExecutionLog
 from workflow_manager.workflow_v2.models.workflow import Workflow
@@ -31,7 +34,8 @@
 
 class SharedExecutionAccessTests(GroupSharingTestBase):
     """``self.member`` belongs to ``self.group``; ``self.outsider`` is an org
-    member with no share of any kind. ``self.owner`` owns every fixture here.
+    member with no share of any kind. ``self.owner`` owns every fixture in org A
+    (the cross-org test builds its own, deliberately unowned).
     """
 
     def setUp(self) -> None:
@@ -51,7 +55,7 @@ def _api_deployment(self, *, shared_to_org: bool = False) -> APIDeployment:
             shared_to_org=shared_to_org,
         )
         # Creator access flows through an OWNER row, not ``created_by`` (UN-2202);
-        # mirrors what ``APIDeploymentViewSet.perform_create`` does.
+        # mirrors what ``APIDeploymentViewSet.create`` does after ``perform_create``.
         deployment.memberships.create(user=self.owner, role=ResourceRole.OWNER)
         return deployment
 
@@ -159,6 +163,27 @@ def test_pipeline_execution_follows_the_pipeline_share(self) -> None:
         shared = self._execution(self._pipeline(shared_to_org=True))
         self.assertTrue(self._visible_to(self.outsider, shared))
 
+    def test_workflow_share_does_not_expose_unshared_deployment_runs(self) -> None:
+        """Pins the ``pipeline_id__isnull=True`` conjunct on the workflow branch.
+
+        Workflow access must not leak into runs of a deployment that was
+        deliberately not shared — the "every path has to be revoked" behaviour.
+        """
+        deployment = self._api_deployment()  # shared with nobody
+        execution = self._execution(deployment)
+        self._share_with_group(self.workflow)
+
+        self.assertIn(self.workflow, Workflow.objects.for_user(self.member))
+        self.assertFalse(self._visible_to(self.member, execution))
+
+    def test_deployment_share_does_not_expose_workflow_level_runs(self) -> None:
+        """The other half: a deployment share says nothing about the workflow's
+        own runs, which is why the two branches are not symmetric.
+        """
+        deployment = self._api_deployment(shared_to_org=True)
+        self.assertTrue(self._visible_to(self.outsider, self._execution(deployment)))
+        self.assertFalse(self._visible_to(self.outsider, self._execution()))
+
     def test_org_wide_share_does_not_cross_organizations(self) -> None:
         """``shared_to_org`` means *this* org — the tenant boundary for
         ``/execution/`` is the manager, since the view drops the org filter
@@ -186,6 +211,11 @@ def test_org_wide_share_does_not_cross_organizations(self) -> None:
             status=ExecutionStatus.COMPLETED,
         )
 
+        # Control: the same shape inside org A *is* visible, so the denial below
+        # is about the org boundary and not about the fixture being malformed.
+        local = self._execution(self._api_deployment(shared_to_org=True))
+        self.assertTrue(self._visible_to(self.outsider, local))
+
         # UserContext still points at org A throughout.
         self.assertFalse(self._visible_to(self.outsider, execution))
         with self.assertRaises(PermissionDenied):
@@ -242,3 +272,114 @@ def test_unknown_execution_is_indistinguishable_from_an_inaccessible_one(
         self.assertEqual(denied.status_code, 403)
         self.assertEqual(unknown.status_code, denied.status_code)
         self.assertEqual(unknown.data, denied.data)
+
+    # --- the sibling routes on the same execution id ---------------------------
+
+    def _call_files(self, user, execution_id):
+        view = FileCentricExecutionViewSet.as_view({"get": "list"})
+        request = APIRequestFactory().get("/")
+        force_authenticate(request, user=user)
+        return view(request, pk=str(execution_id))
+
+    def test_file_executions_follow_the_same_gate_as_the_logs(self) -> None:
+        """``<id>/files/`` carries file names, per-file errors and the latest log
+        line for the same id — 403 on logs and 200 here would defeat the point.
+        """
+        execution = self._execution(self._api_deployment())
+        self.assertEqual(self._call_files(self.outsider, execution.id).status_code, 403)
+
+        shared = self._execution(self._api_deployment(shared_to_org=True))
+        self.assertEqual(self._call_files(self.outsider, shared.id).status_code, 200)
+
+    def test_workflow_execution_list_is_scoped_to_accessible_workflows(self) -> None:
+        """``/workflow/<id>/execution/`` had the same dead ``IsOwner`` gate."""
+        execution = self._execution()  # workflow-level, owner-only
+
+        view = WorkflowExecutionViewSet.as_view({"get": "list"})
+        request = APIRequestFactory().get("/")
+        force_authenticate(request, user=self.outsider)
+        response = view(request, pk=str(self.workflow.id))
+        response.render()
+
+        self.assertEqual(response.status_code, 200)
+        self.assertNotIn(str(execution.id).encode(), response.content)
+
+
+class ExecutionBypassRoleTests(GroupSharingTestBase):
+    """The two branches that skip sharing entirely.
+
+    The admin predicate is patched to a deterministic one — only ``self.admin``
+    — rather than left to resolve for real: the admin *role string* belongs to
+    the active authentication plugin (OSS reads ``"admin"``, the auth0 plugin
+    reads ``"unstract_admin"``), so a test that depended on it would pass in OSS
+    CI and fail on any checkout carrying the plugin. What is under test here is
+    what ``for_user`` does once the predicate is True.
+    """
+
+    def setUp(self) -> None:
+        super().setUp()
+        patcher = patch(
+            _ADMIN_PREDICATE,
+            side_effect=lambda user: getattr(user, "email", None) == self.admin.email,
+        )
+        patcher.start()
+        self.addCleanup(patcher.stop)
+
+    def _execution(self) -> WorkflowExecution:
+        return WorkflowExecution.objects.create(
+            workflow=self.workflow,
+            execution_mode=WorkflowExecution.Mode.INSTANT,
+            execution_method=WorkflowExecution.Method.DIRECT,
+            execution_type=WorkflowExecution.Type.COMPLETE,
+            status=ExecutionStatus.COMPLETED,
+        )
+
+    def _visible_to(self, user, execution) -> bool:
+        return WorkflowExecution.objects.for_user(user).filter(pk=execution.pk).exists()
+
+    def test_org_admin_sees_executions_of_workflows_never_shared_with_them(self) -> None:
+        execution = self._execution()
+        self.assertTrue(self._visible_to(self.admin, execution))
+        # Control: same user, non-admin role → back to the sharing rules.
+        self.assertFalse(self._visible_to(self.outsider, execution))
+
+    def test_service_account_sees_executions_without_any_membership(self) -> None:
+        service = _make_user("svc@example.com", is_service_account=True)
+        self.assertTrue(self._visible_to(service, self._execution()))
+
+    def test_bypass_roles_stay_inside_the_current_organization(self) -> None:
+        """The org filter in ``_org_scoped`` is the only tenant boundary these
+        two branches have — the view drops ``OrganizationFilterBackend``.
+        """
+        other_org = Organization.objects.create(
+            name="org-c", display_name="Org C", organization_id="org-c"
+        )
+        other_workflow = Workflow.objects.create(
+            workflow_name="wf-c", organization=other_org, created_by=self.owner
+        )
+        foreign = WorkflowExecution.objects.create(
+            workflow=other_workflow,
+            execution_mode=WorkflowExecution.Mode.INSTANT,
+            execution_method=WorkflowExecution.Method.DIRECT,
+            execution_type=WorkflowExecution.Type.COMPLETE,
+            status=ExecutionStatus.COMPLETED,
+        )
+        service = _make_user("svc2@example.com", is_service_account=True)
+
+        # Control first: the local execution is visible to both bypass roles.
+        local = self._execution()
+        self.assertTrue(self._visible_to(self.admin, local))
+        self.assertTrue(self._visible_to(service, local))
+
+        self.assertFalse(self._visible_to(self.admin, foreign))
+        self.assertFalse(self._visible_to(service, foreign))
+
+    def test_no_organization_in_context_returns_nothing(self) -> None:
+        """Fail closed rather than returning every tenant's executions."""
+        execution = self._execution()
+        UserContext.set_organization_identifier(None)
+        self.addCleanup(
+            UserContext.set_organization_identifier, self.org.organization_id
+        )
+
+        self.assertFalse(self._visible_to(self.admin, execution))
diff --git a/backend/workflow_manager/file_execution/views.py b/backend/workflow_manager/file_execution/views.py
index cadfa25706..fed8ea2689 100644
--- a/backend/workflow_manager/file_execution/views.py
+++ b/backend/workflow_manager/file_execution/views.py
@@ -3,6 +3,7 @@
 from rest_framework.permissions import IsAuthenticated
 from utils.pagination import CustomPagination
 
+from workflow_manager.execution.access import assert_execution_accessible
 from workflow_manager.file_execution.filter import FileExecutionFilter
 from workflow_manager.file_execution.models import (
     WorkflowFileExecution as FileExecution,
@@ -22,6 +23,11 @@ class FileCentricExecutionViewSet(viewsets.ReadOnlyModelViewSet):
     def get_queryset(self):
         execution_id = self.kwargs.get("pk")
 
+        # Same execution id, same screen, same gate as ``<pk>/logs/`` (UN-2651).
+        # ``status_msg`` resolves to the latest ``ExecutionLog.data["log"]``, so
+        # this response carries log text as well as file names and errors.
+        assert_execution_accessible(self.request.user, execution_id)
+
         # Subquery to get latest non-DEBUG/WARN log data per file execution
         # Avoids N+1 queries when serializing status_msg
         latest_log_subquery = (
diff --git a/backend/workflow_manager/workflow_v2/execution_log_view.py b/backend/workflow_manager/workflow_v2/execution_log_view.py
index 0cb904d852..fa371629d7 100644
--- a/backend/workflow_manager/workflow_v2/execution_log_view.py
+++ b/backend/workflow_manager/workflow_v2/execution_log_view.py
@@ -8,14 +8,13 @@
 from django.http import HttpResponse
 from django.utils import timezone
 from rest_framework import status, viewsets
-from rest_framework.exceptions import PermissionDenied
 from rest_framework.permissions import IsAuthenticated
 from rest_framework.response import Response
 from rest_framework.versioning import URLPathVersioning
 from utils.pagination import CustomPagination
 
+from workflow_manager.execution.access import assert_execution_accessible
 from workflow_manager.workflow_v2.filters import ExecutionLogFilter
-from workflow_manager.workflow_v2.models.execution import WorkflowExecution
 from workflow_manager.workflow_v2.models.execution_log import ExecutionLog
 from workflow_manager.workflow_v2.serializers import WorkflowExecutionLogSerializer
 
@@ -28,9 +27,12 @@
 
 
 class WorkflowExecutionLogViewSet(viewsets.ReadOnlyModelViewSet):
-    # Read-only on purpose: the access gate lives in ``get_queryset``, which write
-    # handlers never call. ``ExecutionLog`` rows are written by workers and their
-    # fields are ``editable=False``, so there is nothing to expose.
+    # Read-only on purpose. ``create`` is the one write handler that never calls
+    # ``get_queryset``, so it would sit outside the gate below entirely; the rest
+    # would run the gate but have nothing legitimate to do. ``ExecutionLog`` rows
+    # are written by the ``consume_log_history`` Celery task, and the serializer
+    # is ``fields = "__all__"`` — ``data`` and ``event_time`` are not
+    # ``editable=False``, so a write verb would expose both.
     versioning_class = URLPathVersioning
     permission_classes = [IsAuthenticated]
     serializer_class = WorkflowExecutionLogSerializer
@@ -43,17 +45,15 @@ def get_queryset(self) -> QuerySet:
         execution_id = self.kwargs.get("pk")
 
         # The URL's execution id is all that addresses these logs, so it is what
-        # gets authorized (UN-2651). Unknown ids are denied like inaccessible ones
-        # so the response does not reveal which ids exist.
-        if (
-            not WorkflowExecution.objects.for_user(self.request.user)
-            .filter(pk=execution_id)
-            .exists()
-        ):
-            raise PermissionDenied("You do not have access to logs for this execution.")
-
-        # Query by execution_id for backward compatibility
-        # Remove filter after execution_id is removed
+        # gets authorized — same gate as ``<pk>/files/`` (UN-2651).
+        assert_execution_accessible(self.request.user, execution_id)
+
+        # ``execution_id`` is the deprecated pre-``wf_execution`` column, kept for
+        # rows written before the FK existed. In request context it matches
+        # nothing: ``OrgAwareManager`` joins the org through ``wf_execution``, so
+        # the legacy-only rows this branch targets are dropped before the OR is
+        # evaluated. It stays for the unscoped contexts (Celery, shell) and goes
+        # when those rows are rotated out.
         return ExecutionLog.objects.filter(
             Q(wf_execution_id=execution_id) | Q(execution_id=execution_id)
         )
diff --git a/backend/workflow_manager/workflow_v2/execution_view.py b/backend/workflow_manager/workflow_v2/execution_view.py
index 47d99d93f5..771264c5e0 100644
--- a/backend/workflow_manager/workflow_v2/execution_view.py
+++ b/backend/workflow_manager/workflow_v2/execution_view.py
@@ -1,7 +1,7 @@
 import logging
 
-from permissions.permission import IsOwner
 from rest_framework import viewsets
+from rest_framework.permissions import IsAuthenticated
 from rest_framework.versioning import URLPathVersioning
 
 from workflow_manager.workflow_v2.models.execution import WorkflowExecution
@@ -10,9 +10,9 @@
 logger = logging.getLogger(__name__)
 
 
-class WorkflowExecutionViewSet(viewsets.ModelViewSet):
+class WorkflowExecutionViewSet(viewsets.ReadOnlyModelViewSet):
     versioning_class = URLPathVersioning
-    permission_classes = [IsOwner]
+    permission_classes = [IsAuthenticated]
     serializer_class = WorkflowExecutionSerializer
 
     CREATED_AT_FIELD_DESC = "-created_at"
@@ -20,7 +20,14 @@ class WorkflowExecutionViewSet(viewsets.ModelViewSet):
     def get_queryset(self):
         # Get the uuid:pk from the URL path
         workflow_id = self.kwargs.get("pk")
-        queryset = WorkflowExecution.objects.filter(workflow_id=workflow_id).order_by(
-            self.CREATED_AT_FIELD_DESC
+        # ``IsOwner`` used to stand here, but it only implements
+        # ``has_object_permission``, which DRF never invokes on ``list`` — the
+        # same dead gate this PR removes from the log viewset. ``for_user`` is
+        # the real one: without it any org member could enumerate every
+        # execution of a workflow never shared with them (UN-2651).
+        queryset = (
+            WorkflowExecution.objects.for_user(self.request.user)
+            .filter(workflow_id=workflow_id)
+            .order_by(self.CREATED_AT_FIELD_DESC)
         )
         return queryset
diff --git a/backend/workflow_manager/workflow_v2/models/execution.py b/backend/workflow_manager/workflow_v2/models/execution.py
index b7526e1651..d3590833ac 100644
--- a/backend/workflow_manager/workflow_v2/models/execution.py
+++ b/backend/workflow_manager/workflow_v2/models/execution.py
@@ -55,22 +55,17 @@ def for_user(self, user) -> QuerySet:
             QuerySet of executions that the user has permission to access
         """
         if getattr(user, "is_service_account", False):
-            org = UserContext.get_organization()
-            if org:
-                return self.filter(workflow__organization=org)
-            return self.all()
+            return self._org_scoped("service account", user)
 
         if OrganizationMemberService.is_user_organization_admin(user):
-            org = UserContext.get_organization()
-            if org:
-                return self.filter(workflow__organization=org)
-            return self.all()
+            return self._org_scoped("org admin", user)
 
         # Defer to each resource's own ``for_user`` so execution visibility matches
         # the resource list: memberships, group shares and ``shared_to_org``
         # (UN-2651). Those managers org-scope themselves via ``UserContext``, so
-        # this is correct on request paths only — a worker or management command
-        # with no org context gets an empty queryset (fail-closed).
+        # this is meaningful on request paths only. Off the request path they
+        # resolve to ``organization_id IS NULL`` rather than to nothing, which
+        # matches orphan rows — call ``for_user`` from a request, not a worker.
         workflow_filter = Q(workflow_id__in=Workflow.objects.for_user(user).values("pk"))
 
         # Filter for API deployments the user can access
@@ -89,6 +84,25 @@ def for_user(self, user) -> QuerySet:
 
         return self.filter(final_filter).distinct()
 
+    def _org_scoped(self, actor: str, user) -> QuerySet:
+        """Every execution in the current organization, for the bypass roles.
+
+        Fails closed when there is no organization in context. This used to be
+        ``self.all()`` — every execution in every tenant — which is unreachable
+        from the three request-path callers but is a bad default for a manager
+        the view now trusts as its only tenant boundary.
+        """
+        org = UserContext.get_organization()
+        if org:
+            return self.filter(workflow__organization=org)
+        logger.warning(
+            "for_user called with no organization in context (%s, user=%s); "
+            "returning no executions",
+            actor,
+            getattr(user, "id", None),
+        )
+        return self.none()
+
     def clean_invalid_workflows(self):
         """Remove execution records with invalid workflow references.
 

From c2e28a989e8a36a915208dc25a02e98d8c3a636a Mon Sep 17 00:00:00 2001
From: kirtimanmishrazipstack <kirtiman.mishra@zipstack.com>
Date: Wed, 12 Aug 2026 14:52:53 +0530
Subject: [PATCH 5/5] UN-2651 [MISC] Trim the comments added by the access-gate
 change

Several of them narrated the mechanism line by line or recorded what the code
used to be. That kind of comment goes stale on the next refactor and adds noise
for anything reading the file whole. Kept the reason, dropped the retelling.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
---
 .../organization_member_service.py            | 13 ++---
 backend/workflow_manager/execution/access.py  | 12 ++---
 .../tests/test_shared_execution_access.py     | 50 +++++++------------
 .../workflow_manager/file_execution/views.py  |  5 +-
 .../workflow_v2/execution_log_view.py         | 20 +++-----
 .../workflow_v2/execution_view.py             |  7 +--
 .../workflow_v2/models/execution.py           | 18 +++----
 7 files changed, 42 insertions(+), 83 deletions(-)

diff --git a/backend/tenant_account_v2/organization_member_service.py b/backend/tenant_account_v2/organization_member_service.py
index ed29d767ab..241b7112a6 100644
--- a/backend/tenant_account_v2/organization_member_service.py
+++ b/backend/tenant_account_v2/organization_member_service.py
@@ -9,10 +9,8 @@
 logger = logging.getLogger(__name__)
 
 # Memo for the admin predicate, set on the ``User`` instance as
-# ``(organization_identifier, is_admin)``. The answer is a function of the user
-# *and* the current org — ``OrganizationMember.objects`` is org-scoped — and
-# ``set_user_organization`` can move a live ``request.user`` between orgs, so the
-# org id is part of the key rather than an assumption about instance lifetime.
+# ``(organization_identifier, is_admin)``. The answer depends on the org, and a
+# live user can be moved between orgs, so the org id is part of the key.
 _ADMIN_MEMO_ATTR = "_unstract_is_org_admin"
 
 
@@ -32,10 +30,9 @@ def is_user_organization_admin(user: Any) -> bool:
         path in the relevant permissions / managers. Returns False on any
         lookup failure (anonymous user, no membership row, DB unavailable).
 
-        The result is memoized on ``user``, keyed by the current organization:
-        ``WorkflowExecutionManager.for_user`` resolves this predicate and then
-        delegates to three resource managers that each re-resolve it, which was
-        four uncached membership lookups per call on a polled endpoint.
+        The result is memoized on ``user``, keyed by the current organization —
+        execution filtering resolves this predicate four times per call
+        otherwise, on a polled endpoint.
         """
         if not user or not getattr(user, "is_authenticated", False):
             return False
diff --git a/backend/workflow_manager/execution/access.py b/backend/workflow_manager/execution/access.py
index 5a30012dd8..fad7eb8302 100644
--- a/backend/workflow_manager/execution/access.py
+++ b/backend/workflow_manager/execution/access.py
@@ -1,9 +1,7 @@
 """Shared access gate for the ``execution/<pk>/...`` routes (UN-2651).
 
-Every route under that prefix is addressed by a ``WorkflowExecution`` id and
-nothing else, so the id is what has to be authorized. The logs endpoint and the
-file-execution endpoint back the same screen for the same id — keeping one gate
-is what stops them drifting apart again.
+Those routes are addressed by a ``WorkflowExecution`` id and nothing else, so
+the id is what gets authorized. One gate keeps them from drifting apart.
 """
 
 import logging
@@ -19,10 +17,8 @@
 def assert_execution_accessible(user, execution_id) -> None:
     """Raise ``PermissionDenied`` unless ``user`` may read ``execution_id``.
 
-    Unknown ids are denied exactly like inaccessible ones, so the response is
-    not an existence oracle. The two are still told apart in the log line —
-    the server is now the only place that can, and the distinction is what
-    separates a stale bookmark from someone walking the id space.
+    Unknown ids are denied like inaccessible ones so the response is not an
+    existence oracle; the log line keeps the distinction.
     """
     if WorkflowExecution.objects.for_user(user).filter(pk=execution_id).exists():
         return
diff --git a/backend/workflow_manager/execution/tests/test_shared_execution_access.py b/backend/workflow_manager/execution/tests/test_shared_execution_access.py
index 7198a8435a..1fb92ff2d3 100644
--- a/backend/workflow_manager/execution/tests/test_shared_execution_access.py
+++ b/backend/workflow_manager/execution/tests/test_shared_execution_access.py
@@ -34,8 +34,7 @@
 
 class SharedExecutionAccessTests(GroupSharingTestBase):
     """``self.member`` belongs to ``self.group``; ``self.outsider`` is an org
-    member with no share of any kind. ``self.owner`` owns every fixture in org A
-    (the cross-org test builds its own, deliberately unowned).
+    member with no share of any kind. ``self.owner`` owns the org A fixtures.
     """
 
     def setUp(self) -> None:
@@ -54,8 +53,7 @@ def _api_deployment(self, *, shared_to_org: bool = False) -> APIDeployment:
             created_by=self.owner,
             shared_to_org=shared_to_org,
         )
-        # Creator access flows through an OWNER row, not ``created_by`` (UN-2202);
-        # mirrors what ``APIDeploymentViewSet.create`` does after ``perform_create``.
+        # Creator access flows through an OWNER row, not ``created_by`` (UN-2202).
         deployment.memberships.create(user=self.owner, role=ResourceRole.OWNER)
         return deployment
 
@@ -98,7 +96,7 @@ def _visible_to(self, user, execution: WorkflowExecution) -> bool:
         )
 
     def _log_queryset(self, user, execution_id):
-        """Run the log viewset's queryset build for ``user`` — the access gate."""
+        """Run the log viewset's queryset build for ``user``."""
         request = APIRequestFactory().get("/")
         request.user = user
         view = WorkflowExecutionLogViewSet()
@@ -117,9 +115,7 @@ def _call(self, action: str, user, execution_id, **query):
 
     def test_unshared_deployment_is_visible_only_to_its_owner(self) -> None:
         execution = self._execution(self._api_deployment())
-        # The owner assertion is what makes the denial below a real control:
-        # without the OWNER membership row the deployment would be visible to
-        # nobody, and the denial would pass with the sharing filter deleted.
+        # Positive control: without it the denial passes even with no filter.
         self.assertTrue(self._visible_to(self.owner, execution))
         self.assertFalse(self._visible_to(self.outsider, execution))
 
@@ -164,11 +160,7 @@ def test_pipeline_execution_follows_the_pipeline_share(self) -> None:
         self.assertTrue(self._visible_to(self.outsider, shared))
 
     def test_workflow_share_does_not_expose_unshared_deployment_runs(self) -> None:
-        """Pins the ``pipeline_id__isnull=True`` conjunct on the workflow branch.
-
-        Workflow access must not leak into runs of a deployment that was
-        deliberately not shared — the "every path has to be revoked" behaviour.
-        """
+        """Workflow access must not leak into runs of an unshared deployment."""
         deployment = self._api_deployment()  # shared with nobody
         execution = self._execution(deployment)
         self._share_with_group(self.workflow)
@@ -177,18 +169,15 @@ def test_workflow_share_does_not_expose_unshared_deployment_runs(self) -> None:
         self.assertFalse(self._visible_to(self.member, execution))
 
     def test_deployment_share_does_not_expose_workflow_level_runs(self) -> None:
-        """The other half: a deployment share says nothing about the workflow's
-        own runs, which is why the two branches are not symmetric.
+        """The other direction: a deployment share says nothing about the
+        workflow's own runs.
         """
         deployment = self._api_deployment(shared_to_org=True)
         self.assertTrue(self._visible_to(self.outsider, self._execution(deployment)))
         self.assertFalse(self._visible_to(self.outsider, self._execution()))
 
     def test_org_wide_share_does_not_cross_organizations(self) -> None:
-        """``shared_to_org`` means *this* org — the tenant boundary for
-        ``/execution/`` is the manager, since the view drops the org filter
-        backend.
-        """
+        """``shared_to_org`` means *this* org."""
         other_org = Organization.objects.create(
             name="org-b", display_name="Org B", organization_id="org-b"
         )
@@ -211,8 +200,7 @@ def test_org_wide_share_does_not_cross_organizations(self) -> None:
             status=ExecutionStatus.COMPLETED,
         )
 
-        # Control: the same shape inside org A *is* visible, so the denial below
-        # is about the org boundary and not about the fixture being malformed.
+        # Control: the same shape inside org A is visible.
         local = self._execution(self._api_deployment(shared_to_org=True))
         self.assertTrue(self._visible_to(self.outsider, local))
 
@@ -282,8 +270,8 @@ def _call_files(self, user, execution_id):
         return view(request, pk=str(execution_id))
 
     def test_file_executions_follow_the_same_gate_as_the_logs(self) -> None:
-        """``<id>/files/`` carries file names, per-file errors and the latest log
-        line for the same id — 403 on logs and 200 here would defeat the point.
+        """``<id>/files/`` carries log text for the same id, so it must not
+        answer 200 where the logs answer 403.
         """
         execution = self._execution(self._api_deployment())
         self.assertEqual(self._call_files(self.outsider, execution.id).status_code, 403)
@@ -292,7 +280,7 @@ def test_file_executions_follow_the_same_gate_as_the_logs(self) -> None:
         self.assertEqual(self._call_files(self.outsider, shared.id).status_code, 200)
 
     def test_workflow_execution_list_is_scoped_to_accessible_workflows(self) -> None:
-        """``/workflow/<id>/execution/`` had the same dead ``IsOwner`` gate."""
+        """The execution list is scoped the same way as the logs."""
         execution = self._execution()  # workflow-level, owner-only
 
         view = WorkflowExecutionViewSet.as_view({"get": "list"})
@@ -308,12 +296,10 @@ def test_workflow_execution_list_is_scoped_to_accessible_workflows(self) -> None
 class ExecutionBypassRoleTests(GroupSharingTestBase):
     """The two branches that skip sharing entirely.
 
-    The admin predicate is patched to a deterministic one — only ``self.admin``
-    — rather than left to resolve for real: the admin *role string* belongs to
-    the active authentication plugin (OSS reads ``"admin"``, the auth0 plugin
-    reads ``"unstract_admin"``), so a test that depended on it would pass in OSS
-    CI and fail on any checkout carrying the plugin. What is under test here is
-    what ``for_user`` does once the predicate is True.
+    The admin predicate is patched rather than resolved for real: the admin role
+    string comes from the active authentication plugin, so resolving it makes
+    the result depend on which plugins are installed. Under test is what
+    ``for_user`` does once the predicate is True.
     """
 
     def setUp(self) -> None:
@@ -348,9 +334,7 @@ def test_service_account_sees_executions_without_any_membership(self) -> None:
         self.assertTrue(self._visible_to(service, self._execution()))
 
     def test_bypass_roles_stay_inside_the_current_organization(self) -> None:
-        """The org filter in ``_org_scoped`` is the only tenant boundary these
-        two branches have — the view drops ``OrganizationFilterBackend``.
-        """
+        """The org filter is the only tenant boundary these two branches have."""
         other_org = Organization.objects.create(
             name="org-c", display_name="Org C", organization_id="org-c"
         )
diff --git a/backend/workflow_manager/file_execution/views.py b/backend/workflow_manager/file_execution/views.py
index fed8ea2689..13c0f1b97d 100644
--- a/backend/workflow_manager/file_execution/views.py
+++ b/backend/workflow_manager/file_execution/views.py
@@ -23,9 +23,8 @@ class FileCentricExecutionViewSet(viewsets.ReadOnlyModelViewSet):
     def get_queryset(self):
         execution_id = self.kwargs.get("pk")
 
-        # Same execution id, same screen, same gate as ``<pk>/logs/`` (UN-2651).
-        # ``status_msg`` resolves to the latest ``ExecutionLog.data["log"]``, so
-        # this response carries log text as well as file names and errors.
+        # Same id and same data class as ``<pk>/logs/`` — ``status_msg`` carries
+        # log text — so the same gate applies (UN-2651).
         assert_execution_accessible(self.request.user, execution_id)
 
         # Subquery to get latest non-DEBUG/WARN log data per file execution
diff --git a/backend/workflow_manager/workflow_v2/execution_log_view.py b/backend/workflow_manager/workflow_v2/execution_log_view.py
index fa371629d7..3ee690c64f 100644
--- a/backend/workflow_manager/workflow_v2/execution_log_view.py
+++ b/backend/workflow_manager/workflow_v2/execution_log_view.py
@@ -27,12 +27,8 @@
 
 
 class WorkflowExecutionLogViewSet(viewsets.ReadOnlyModelViewSet):
-    # Read-only on purpose. ``create`` is the one write handler that never calls
-    # ``get_queryset``, so it would sit outside the gate below entirely; the rest
-    # would run the gate but have nothing legitimate to do. ``ExecutionLog`` rows
-    # are written by the ``consume_log_history`` Celery task, and the serializer
-    # is ``fields = "__all__"`` — ``data`` and ``event_time`` are not
-    # ``editable=False``, so a write verb would expose both.
+    # Read-only: rows are written by the log-consumer task, and ``create`` is the
+    # one handler that would skip the gate in ``get_queryset``.
     versioning_class = URLPathVersioning
     permission_classes = [IsAuthenticated]
     serializer_class = WorkflowExecutionLogSerializer
@@ -44,16 +40,12 @@ class WorkflowExecutionLogViewSet(viewsets.ReadOnlyModelViewSet):
     def get_queryset(self) -> QuerySet:
         execution_id = self.kwargs.get("pk")
 
-        # The URL's execution id is all that addresses these logs, so it is what
-        # gets authorized — same gate as ``<pk>/files/`` (UN-2651).
+        # The execution id in the URL is all that addresses these logs (UN-2651).
         assert_execution_accessible(self.request.user, execution_id)
 
-        # ``execution_id`` is the deprecated pre-``wf_execution`` column, kept for
-        # rows written before the FK existed. In request context it matches
-        # nothing: ``OrgAwareManager`` joins the org through ``wf_execution``, so
-        # the legacy-only rows this branch targets are dropped before the OR is
-        # evaluated. It stays for the unscoped contexts (Celery, shell) and goes
-        # when those rows are rotated out.
+        # ``execution_id`` is the deprecated pre-``wf_execution`` column. Org
+        # scoping joins through ``wf_execution``, so this term matches nothing in
+        # request context; it stays until those rows are rotated out.
         return ExecutionLog.objects.filter(
             Q(wf_execution_id=execution_id) | Q(execution_id=execution_id)
         )
diff --git a/backend/workflow_manager/workflow_v2/execution_view.py b/backend/workflow_manager/workflow_v2/execution_view.py
index 771264c5e0..80701105f3 100644
--- a/backend/workflow_manager/workflow_v2/execution_view.py
+++ b/backend/workflow_manager/workflow_v2/execution_view.py
@@ -20,11 +20,8 @@ class WorkflowExecutionViewSet(viewsets.ReadOnlyModelViewSet):
     def get_queryset(self):
         # Get the uuid:pk from the URL path
         workflow_id = self.kwargs.get("pk")
-        # ``IsOwner`` used to stand here, but it only implements
-        # ``has_object_permission``, which DRF never invokes on ``list`` — the
-        # same dead gate this PR removes from the log viewset. ``for_user`` is
-        # the real one: without it any org member could enumerate every
-        # execution of a workflow never shared with them (UN-2651).
+        # ``for_user`` is the gate — an object-level permission class does not
+        # run on ``list`` (UN-2651).
         queryset = (
             WorkflowExecution.objects.for_user(self.request.user)
             .filter(workflow_id=workflow_id)
diff --git a/backend/workflow_manager/workflow_v2/models/execution.py b/backend/workflow_manager/workflow_v2/models/execution.py
index d3590833ac..a0ec40ae64 100644
--- a/backend/workflow_manager/workflow_v2/models/execution.py
+++ b/backend/workflow_manager/workflow_v2/models/execution.py
@@ -44,9 +44,8 @@ def for_user(self, user) -> QuerySet:
         4. Neither shared -> User cannot see executions
 
         Service accounts and org admins see every execution in the current
-        organization. That org scoping is enforced here, not by the view:
-        ``ExecutionViewSet`` replaces ``filter_backends``, which drops
-        ``OrganizationFilterBackend``.
+        organization. This manager is the only org scoping on that path — the
+        view drops the organization filter backend.
 
         Args:
             user: The user to filter executions for
@@ -61,11 +60,8 @@ def for_user(self, user) -> QuerySet:
             return self._org_scoped("org admin", user)
 
         # Defer to each resource's own ``for_user`` so execution visibility matches
-        # the resource list: memberships, group shares and ``shared_to_org``
-        # (UN-2651). Those managers org-scope themselves via ``UserContext``, so
-        # this is meaningful on request paths only. Off the request path they
-        # resolve to ``organization_id IS NULL`` rather than to nothing, which
-        # matches orphan rows — call ``for_user`` from a request, not a worker.
+        # the resource lists (UN-2651). Those managers scope by the org in
+        # context, so call this from a request, not a worker.
         workflow_filter = Q(workflow_id__in=Workflow.objects.for_user(user).values("pk"))
 
         # Filter for API deployments the user can access
@@ -87,10 +83,8 @@ def for_user(self, user) -> QuerySet:
     def _org_scoped(self, actor: str, user) -> QuerySet:
         """Every execution in the current organization, for the bypass roles.
 
-        Fails closed when there is no organization in context. This used to be
-        ``self.all()`` — every execution in every tenant — which is unreachable
-        from the three request-path callers but is a bad default for a manager
-        the view now trusts as its only tenant boundary.
+        Fails closed with no organization in context, rather than returning
+        every tenant's executions.
         """
         org = UserContext.get_organization()
         if org: