diff --git a/backend/tenant_account_v2/organization_member_service.py b/backend/tenant_account_v2/organization_member_service.py index 7ba92d9dbd..241b7112a6 100644 --- a/backend/tenant_account_v2/organization_member_service.py +++ b/backend/tenant_account_v2/organization_member_service.py @@ -2,11 +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 for the admin predicate, set on the ``User`` instance as +# ``(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" + 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``, 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 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 and memo[0] == org_id: + return memo[1] try: member = OrganizationMember.objects.get(user=user.id) # type: ignore except OrganizationMember.DoesNotExist: @@ -42,7 +56,11 @@ 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. + setattr(user, _ADMIN_MEMO_ATTR, (org_id, is_admin)) + return is_admin @staticmethod def get_user_by_user_id(user_id: str) -> OrganizationMember | None: diff --git a/backend/workflow_manager/execution/access.py b/backend/workflow_manager/execution/access.py new file mode 100644 index 0000000000..fad7eb8302 --- /dev/null +++ b/backend/workflow_manager/execution/access.py @@ -0,0 +1,33 @@ +"""Shared access gate for the ``execution//...`` routes (UN-2651). + +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 + +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 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 + + 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 new file mode 100644 index 0000000000..1fb92ff2d3 --- /dev/null +++ b/backend/workflow_manager/execution/tests/test_shared_execution_access.py @@ -0,0 +1,369 @@ +"""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 +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, force_authenticate +from tenant_account_v2.models import ResourceGroupShare +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 + +_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. ``self.owner`` owns the org A fixtures. + """ + + 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. + 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, + ) + # Creator access flows through an OWNER row, not ``created_by`` (UN-2202). + 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=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, resource) -> None: + ResourceGroupShare.objects.create( + group=self.group, + content_type=ContentType.objects.get_for_model(type(resource)), + object_id=str(resource.pk), + 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``.""" + request = APIRequestFactory().get("/") + request.user = user + view = WorkflowExecutionLogViewSet() + view.request = request + 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()) + # 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)) + + 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)) + + 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_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_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_workflow_share_does_not_expose_unshared_deployment_runs(self) -> None: + """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) + + 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 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.""" + 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, + ) + + # 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)) + + # 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) + + # --- 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, 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))) + + 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) + + # --- 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: + """``/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) + + 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: + """The execution list is scoped the same way as the logs.""" + 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 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: + 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 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" + ) + 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..13c0f1b97d 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,10 @@ class FileCentricExecutionViewSet(viewsets.ReadOnlyModelViewSet): def get_queryset(self): execution_id = self.kwargs.get("pk") + # Same id and same data class as ``/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 # 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 2d8575080d..3ee690c64f 100644 --- a/backend/workflow_manager/workflow_v2/execution_log_view.py +++ b/backend/workflow_manager/workflow_v2/execution_log_view.py @@ -7,13 +7,13 @@ 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.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_log import ExecutionLog from workflow_manager.workflow_v2.serializers import WorkflowExecutionLogSerializer @@ -26,9 +26,11 @@ MAX_SYNC_EXPORT_ROWS = 50_000 -class WorkflowExecutionLogViewSet(viewsets.ModelViewSet): +class WorkflowExecutionLogViewSet(viewsets.ReadOnlyModelViewSet): + # 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, IsOwner] + permission_classes = [IsAuthenticated] serializer_class = WorkflowExecutionLogSerializer pagination_class = CustomPagination ordering_fields = ["event_time"] @@ -38,8 +40,12 @@ class WorkflowExecutionLogViewSet(viewsets.ModelViewSet): def get_queryset(self) -> QuerySet: execution_id = self.kwargs.get("pk") - # Query by execution_id for backward compatibility - # Remove filter after execution_id is removed + # 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. 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 47d99d93f5..80701105f3 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,11 @@ 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 + # ``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) + .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 dc63a19ac7..a0ec40ae64 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 @@ -44,7 +43,9 @@ 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. 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 @@ -53,34 +54,21 @@ 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() - - # 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) - ) + return self._org_scoped("org admin", user) + + # Defer to each resource's own ``for_user`` so execution visibility matches + # 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 - 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 @@ -92,6 +80,23 @@ 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 with no organization in context, rather than returning + every tenant's executions. + """ + 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.