fix(workflows): Check workflow permissions before attaching detectors

src/sentry/workflow_engine/endpoints/organization_workflow_index.py

@@ -56,7 +56,6 @@
 from sentry.deletions.models.scheduleddeletion import CellScheduledDeletion
 from sentry.exceptions import InvalidSearchQuery
 from sentry.models.organization import Organization
-from sentry.models.project import Project
 from sentry.search.utils import parse_user_value
 from sentry.utils.audit import create_audit_entry
 from sentry.utils.dates import ensure_aware
@@ -75,7 +74,7 @@
     DetectorWorkflowMutationValidator,
 )
 from sentry.workflow_engine.endpoints.validators.utils import (
-    is_workflow_connected_to_all_projects_detector,
+    enforce_workflow_access,
     should_include_all_projects_detector_workflows,
     should_include_all_projects_detector_workflows_or_raise,
 )
@@ -130,27 +129,7 @@ def convert_args(
         except Workflow.DoesNotExist:
             raise ResourceDoesNotExist
 
-        # Check project access for workflows connected to detectors.
-        # User must have access to at least one connected project.
-        # Workflows with no detector connections are org-level and accessible
-        # to anyone with org-level workflow permissions.
-        workflow = kwargs["workflow"]
-        organization = kwargs["organization"]
-        if is_workflow_connected_to_all_projects_detector(workflow):
-            if not should_include_all_projects_detector_workflows_or_raise(request, organization):
-                raise PermissionDenied
-            return args, kwargs
-
-        connected_projects = Project.objects.filter(
-            detector__detectorworkflow__workflow=workflow
-        ).distinct()
-
-        if connected_projects.exists():
-            has_access = any(
-                request.access.has_project_access(project) for project in connected_projects
-            )
-            if not has_access:
-                raise PermissionDenied
+        enforce_workflow_access(kwargs["workflow"], kwargs["organization"], request)
 
         return args, kwargs
 

src/sentry/workflow_engine/endpoints/utils/permissions.py

@@ -1,48 +1,16 @@
-from collections.abc import Sequence
-
 from rest_framework.exceptions import PermissionDenied
 from rest_framework.request import Request
 
 from sentry.models.organization import Organization
 from sentry.workflow_engine.endpoints.validators.utils import (
-    can_edit_detector_workflow_connections,
+    ORGANIZATION_WORKFLOW_WRITE_SCOPES,
     validate_detectors_exist_and_have_permissions,
 )
-from sentry.workflow_engine.models import DetectorWorkflow, Workflow
+from sentry.workflow_engine.endpoints.validators.utils import (
+    can_edit_workflows as can_edit_workflows,
+)
 from sentry.workflow_engine.types import DetectorId
 
-ORGANIZATION_WORKFLOW_WRITE_SCOPES = ("org:write", "org:admin", "alerts:write")
-
-
-def can_edit_workflows(workflows: Sequence[Workflow], request: Request) -> bool:
-    """
-    Determine if the requesting user can edit every workflow in the sequence.
-
-    Organization alert writers can delete organization-level workflows. Otherwise,
-    every workflow must be connected to at least one detector and the user must be
-    able to edit every detector-workflow connection.
-    """
-    workflow_ids = {workflow.id for workflow in workflows}
-    if not workflow_ids:
-        return False
-
-    if any(request.access.has_scope(scope) for scope in ORGANIZATION_WORKFLOW_WRITE_SCOPES):
-        return True
-
-    detector_workflows = list(
-        DetectorWorkflow.objects.filter(workflow_id__in=workflow_ids).select_related(
-            "detector", "detector__project"
-        )
-    )
-    connected_workflow_ids = {
-        detector_workflow.workflow_id for detector_workflow in detector_workflows
-    }
-
-    return workflow_ids == connected_workflow_ids and all(
-        can_edit_detector_workflow_connections(detector_workflow.detector, request)
-        for detector_workflow in detector_workflows
-    )
-
 
 def enforce_workflow_creation_permissions(
     request: Request,

src/sentry/workflow_engine/endpoints/validators/base/detector.py

@@ -33,6 +33,7 @@
     get_unknown_detector_type_error,
     log_alerting_quota_hit,
     update_owner,
+    validate_workflow_connections,
 )
 from sentry.workflow_engine.models import (
     DataConditionGroup,
@@ -128,6 +129,15 @@ def validate_type(self, value: str) -> builtins.type[GroupType]:
         # org/user is allowed to add a detector
         return type
 
+    def validate_workflow_ids(self, value: list[int]) -> list[int]:
+        validate_workflow_connections(
+            value,
+            self.context["organization"],
+            self.context["request"],
+            self.instance.id if self.instance else None,
+        )
+        return value
+
     @property
     def data_conditions(self) -> BaseDataConditionValidator:
         raise NotImplementedError

src/sentry/workflow_engine/endpoints/validators/utils.py

@@ -30,6 +30,7 @@
 # Only those with organization write permissions can edit system-created detectors (e.g. error detectors).
 SYSTEM_CREATED_DETECTOR_REQUIRED_SCOPES = {"org:write"}
 USER_CREATED_DETECTOR_REQUIRED_SCOPES = {"org:write", "alerts:write"}
+ORGANIZATION_WORKFLOW_WRITE_SCOPES = ("org:write", "org:admin", "alerts:write")
 
 
 def is_system_created_detector(detector: Detector) -> bool:
@@ -167,6 +168,73 @@ def validate_workflows_exist(
     return workflows
 
 
+def enforce_workflow_access(
+    workflow: Workflow, organization: Organization, request: Request
+) -> None:
+    """Enforce workflow access using its existing detector connections."""
+    if is_workflow_connected_to_all_projects_detector(workflow):
+        if not should_include_all_projects_detector_workflows_or_raise(request, organization):
+            raise PermissionDenied
+        return
+
+    connected_projects = Project.objects.filter(
+        detector__detectorworkflow__workflow=workflow
+    ).distinct()
+    if connected_projects.exists() and not any(
+        request.access.has_project_access(project) for project in connected_projects
+    ):
+        raise PermissionDenied
+
+
+def can_edit_workflows(workflows: Sequence[Workflow], request: Request) -> bool:
+    """
+    Check write permissions after enforcing access to each workflow.
+
+    Organization alert writers can edit workflows. Otherwise, every workflow must
+    have connections and the caller must be able to edit every connection.
+    """
+    workflow_ids = {workflow.id for workflow in workflows}
+    if not workflow_ids:
+        return False
+
+    if any(request.access.has_scope(scope) for scope in ORGANIZATION_WORKFLOW_WRITE_SCOPES):
+        return True
+
+    detector_workflows = list(
+        DetectorWorkflow.objects.filter(workflow_id__in=workflow_ids).select_related(
+            "detector", "detector__project"
+        )
+    )
+    connected_workflow_ids = {
+        detector_workflow.workflow_id for detector_workflow in detector_workflows
+    }
+
+    return workflow_ids == connected_workflow_ids and all(
+        can_edit_detector_workflow_connections(detector_workflow.detector, request)
+        for detector_workflow in detector_workflows
+    )
+
+
+def validate_workflow_connections(
+    workflow_ids: list[int],
+    organization: Organization,
+    request: Request,
+    detector_id: DetectorId | None = None,
+) -> None:
+    workflows = validate_workflows_exist(workflow_ids, organization)
+    if detector_id is not None:
+        # Retaining or removing an existing connection only requires permission
+        # on the detector. Check workflow permissions for new connections.
+        workflows = workflows.exclude(detectorworkflow__detector_id=detector_id)
+
+    new_workflows = list(workflows)
+    for workflow in new_workflows:
+        enforce_workflow_access(workflow, organization, request)
+
+    if new_workflows and not can_edit_workflows(new_workflows, request):
+        raise PermissionDenied
+
+
 def connect_workflows_to_detectors(
     request: Request,
     organization: Organization,
@@ -228,7 +296,9 @@ def connect_detectors_to_workflows(
     update: bool = False,
 ) -> None:
     if workflow_ids is not None:
-        validate_workflows_exist(workflow_ids, organization)
+        validate_workflow_connections(
+            workflow_ids, organization, request, detector_id if update else None
+        )
 
         def get_detector_workflows_to_add(
             detector_id: DetectorId, workflow_ids: set[WorkflowId]

tests/sentry/workflow_engine/endpoints/test_organization_detector_details.py

@@ -15,6 +15,7 @@
 from sentry.incidents.utils.constants import INCIDENTS_SNUBA_SUBSCRIPTION_TYPE
 from sentry.incidents.utils.subscription_limits import METRIC_SUBSCRIPTION_FEATURE_FLAGS
 from sentry.models.auditlogentry import AuditLogEntry
+from sentry.monitors.grouptype import MonitorIncidentType
 from sentry.silo.base import SiloMode
 from sentry.snuba.dataset import Dataset
 from sentry.snuba.models import QuerySubscription, SnubaQuery, SnubaQueryEventType
@@ -41,10 +42,64 @@
 from sentry.workflow_engine.models.detector_workflow import DetectorWorkflow
 from sentry.workflow_engine.types import DetectorPriorityLevel
 from sentry.workflow_engine.typings.grouptype import IssueStreamGroupType
+from tests.sentry.workflow_engine.test_base import ProjectAccessTestMixin
 
 pytestmark = [pytest.mark.sentry_metrics, requires_snuba, requires_kafka]
 
 
+@cell_silo_test
+class OrganizationDetectorWorkflowAccessTest(APITestCase, ProjectAccessTestMixin):
+    endpoint = "sentry-api-0-organization-detector-details"
+    method = "PUT"
+
+    def setUp(self) -> None:
+        super().setUp()
+        self.setup_project_access_test_data()
+        self.organization.update_option("sentry:alerts_member_write", True)
+        self.login_as(self.limited_user)
+        self.detector = self.create_detector(
+            project=self.user_project, type=MonitorIncidentType.slug
+        )
+        self.connection = self.create_detector_workflow(
+            detector=self.detector, workflow=self.user_workflow
+        )
+
+    def test_workflow_attachment_permissions(self) -> None:
+        workflow_url = (
+            f"/api/0/organizations/{self.organization.slug}/workflows/{self.other_workflow.id}/"
+        )
+        assert self.client.get(workflow_url).status_code == 403
+        original_name = self.detector.name
+
+        self.get_error_response(
+            self.organization.slug,
+            self.detector.id,
+            name="Unauthorized change",
+            workflowIds=[self.unattached_workflow.id, self.other_workflow.id],
+            status_code=403,
+        )
+
+        self.detector.refresh_from_db()
+        assert self.detector.name == original_name
+        assert list(
+            DetectorWorkflow.objects.filter(detector=self.detector).values_list("id", flat=True)
+        ) == [self.connection.id]
+        assert self.client.get(workflow_url).status_code == 403
+
+        # An authorized replacement still succeeds after the rejected update.
+        self.get_success_response(
+            self.organization.slug,
+            self.detector.id,
+            workflowIds=[self.unattached_workflow.id],
+        )
+
+        assert list(
+            DetectorWorkflow.objects.filter(detector=self.detector).values_list(
+                "workflow_id", flat=True
+            )
+        ) == [self.unattached_workflow.id]
+
+
 @pytest.mark.snuba_ci
 @with_feature(METRIC_SUBSCRIPTION_FEATURE_FLAGS)
 class OrganizationDetectorDetailsBaseTest(APITestCase):

tests/sentry/workflow_engine/endpoints/test_organization_project_detector_index.py

@@ -36,6 +36,72 @@
 from sentry.workflow_engine.models.detector_workflow import DetectorWorkflow
 from sentry.workflow_engine.registry import data_source_type_registry
 from sentry.workflow_engine.types import DetectorPriorityLevel
+from tests.sentry.workflow_engine.test_base import ProjectAccessTestMixin
+
+
+@cell_silo_test
+class OrganizationProjectDetectorWorkflowAccessTest(APITestCase, ProjectAccessTestMixin):
+    endpoint = "sentry-api-0-organization-project-detector-index"
+    method = "POST"
+
+    def setUp(self) -> None:
+        super().setUp()
+        self.setup_project_access_test_data()
+        self.organization.update_option("sentry:alerts_member_write", True)
+        self.login_as(self.limited_user)
+        self.data = {
+            "type": MonitorIncidentType.slug,
+            "name": "Test Monitor Detector",
+            "dataSources": [
+                {
+                    "name": "Test Monitor",
+                    "config": {"schedule": "0 * * * *", "scheduleType": "crontab"},
+                }
+            ],
+        }
+
+    def test_workflow_attachment_permissions(self) -> None:
+        workflow_url = (
+            f"/api/0/organizations/{self.organization.slug}/workflows/{self.other_workflow.id}/"
+        )
+        assert self.client.get(workflow_url).status_code == 403
+        initial_detectors = Detector.objects.filter(project=self.user_project).count()
+        initial_data_sources = DataSource.objects.filter(organization=self.organization).count()
+        initial_monitors = Monitor.objects.filter(project_id=self.user_project.id).count()
+
+        self.get_error_response(
+            self.organization.slug,
+            self.user_project.slug,
+            **self.data,
+            workflowIds=[self.user_workflow.id, self.other_workflow.id],
+            status_code=403,
+        )
+
+        assert Detector.objects.filter(project=self.user_project).count() == initial_detectors
+        assert (
+            DataSource.objects.filter(organization=self.organization).count()
+            == initial_data_sources
+        )
+        assert Monitor.objects.filter(project_id=self.user_project.id).count() == initial_monitors
+        assert not DetectorWorkflow.objects.filter(
+            detector__project=self.user_project, workflow=self.other_workflow
+        ).exists()
+        assert self.client.get(workflow_url).status_code == 403
+
+        # The same member can create a detector with workflows they can edit.
+        response = self.get_success_response(
+            self.organization.slug,
+            self.user_project.slug,
+            **self.data,
+            workflowIds=[self.user_workflow.id, self.unattached_workflow.id],
+            status_code=201,
+        )
+
+        assert set(
+            DetectorWorkflow.objects.filter(detector_id=response.data["id"]).values_list(
+                "workflow_id", flat=True
+            )
+        ) == {self.user_workflow.id, self.unattached_workflow.id}
 
 
 class OrganizationProjectDetectorIndexBaseTest(APITestCase):