fix(aci): Forbid unsafe legacy interactions with shared Workflows/Detectors

src/sentry/api/bases/rule.py

@@ -1,5 +1,7 @@
 from typing import Any
 
+from rest_framework import status
+from rest_framework.exceptions import APIException
 from rest_framework.request import Request
 
 from sentry import features
@@ -14,6 +16,16 @@
 from sentry.workflow_engine.utils.legacy_metric_tracking import report_used_legacy_models
 
 
+class SharedWorkflowError(APIException):
+    status_code = status.HTTP_400_BAD_REQUEST
+
+    def __init__(self, workflow_id: int) -> None:
+        detail = (
+            f"Workflow {workflow_id} is shared with other rules. Use the workflow API to manage it."
+        )
+        super().__init__(detail=detail)
+
+
 class RuleEndpoint(ProjectEndpoint):
     owner = ApiOwner.ISSUES
     permission_classes = (ProjectAlertRulePermission,)
@@ -66,6 +78,16 @@ def convert_args(
                     workflow__organization=project.organization,
                     workflow__status=ObjectStatus.ACTIVE,
                 )
+                # For mutating requests via a legacy Rule ID, reject if the
+                # Workflow is shared with other Rules or AlertRules.
+                if request.method in ("PUT", "DELETE"):
+                    has_other_links = (
+                        AlertRuleWorkflow.objects.filter(workflow_id=arw.workflow_id)
+                        .exclude(id=arw.id)
+                        .exists()
+                    )
+                    if has_other_links:
+                        raise SharedWorkflowError(arw.workflow_id)
                 kwargs["rule"] = arw.workflow
             except AlertRuleWorkflow.DoesNotExist:
                 # XXX: this means the workflow was single written and has no ARW or related Rule object

src/sentry/incidents/endpoints/bases.py

@@ -1,6 +1,7 @@
 from typing import Any
 
-from rest_framework.exceptions import PermissionDenied
+from rest_framework import status
+from rest_framework.exceptions import APIException, PermissionDenied
 from rest_framework.request import Request
 
 from sentry import features
@@ -16,6 +17,28 @@
 from sentry.workflow_engine.models.detector import Detector
 
 
+class SharedDetectorError(APIException):
+    status_code = status.HTTP_400_BAD_REQUEST
+
+    def __init__(self, detector_id: int) -> None:
+        detail = (
+            f"Detector {detector_id} is shared with other rules. Use the detector API to manage it."
+        )
+        super().__init__(detail=detail)
+
+
+def _check_shared_detector(request: Request, ard: AlertRuleDetector) -> None:
+    """Reject PUT/DELETE on a Detector that is shared with other rules."""
+    if request.method in ("PUT", "DELETE"):
+        has_other_links = (
+            AlertRuleDetector.objects.filter(detector_id=ard.detector_id)
+            .exclude(id=ard.id)
+            .exists()
+        )
+        if has_other_links:
+            raise SharedDetectorError(ard.detector_id)
+
+
 class OrganizationAlertRuleBaseEndpoint(OrganizationEndpoint):
     """
     Base endpoint for organization-scoped alert rule creation.
@@ -131,6 +154,7 @@ def convert_args(
                     alert_rule_id=validated_alert_rule_id,
                     detector__project=project,
                 )
+                _check_shared_detector(request, ard)
                 kwargs["alert_rule"] = ard.detector
             except AlertRuleDetector.DoesNotExist:
                 # XXX: this means the detector was single written and has no ARD or related AlertRule object
@@ -179,6 +203,7 @@ def convert_args(
                     alert_rule_id=validated_alert_rule_id,
                     detector__project__organization=organization,
                 )
+                _check_shared_detector(request, ard)
                 kwargs["alert_rule"] = ard.detector
             except AlertRuleDetector.DoesNotExist:
                 # XXX: this means the detector was single written and has no ARD or related AlertRule object

tests/sentry/api/endpoints/test_project_rule_details.py

@@ -1473,6 +1473,27 @@ def test_delete_does_not_cascade_to_cross_org_rule(self) -> None:
         other_rule.refresh_from_db()
         assert other_rule.status == ObjectStatus.ACTIVE
 
+    def test_delete_shared_workflow_returns_400(self) -> None:
+        rule = self.create_project_rule(self.project)
+        arw = AlertRuleWorkflow.objects.get(rule_id=rule.id)
+        workflow = arw.workflow
+
+        # Create another Rule linked to the same Workflow.
+        other_rule = self.create_project_rule(self.project)
+        other_arw = AlertRuleWorkflow.objects.get(rule_id=other_rule.id)
+        other_arw.update(workflow_id=workflow.id)
+
+        response = self.get_error_response(
+            self.organization.slug, self.project.slug, rule.id, status_code=400
+        )
+        assert "Workflow" in response.data["detail"]
+        assert str(workflow.id) in response.data["detail"]
+
+        # Neither rule nor workflow were deleted.
+        rule.refresh_from_db()
+        assert rule.status == ObjectStatus.ACTIVE
+        assert Workflow.objects.filter(id=workflow.id).exists()
+
 
 class GetProjectRuleDetailsDeltaTest(ProjectRuleDetailsBaseTestCase):
     def test_dual_written_rule_parity(self) -> None:

tests/sentry/incidents/endpoints/test_organization_alert_rule_details.py

@@ -2757,3 +2757,24 @@ def test_dual_delete_detector_id_passed(self) -> None:
         assert not AlertRule.objects_with_snapshots.filter(name=self.alert_rule.name).exists()
         assert not AlertRule.objects_with_snapshots.filter(id=self.alert_rule.id).exists()
         assert not Detector.objects.filter(id=ard.detector_id).exists()
+
+    @with_feature("organizations:workflow-engine-rule-serializers")
+    def test_delete_shared_detector_returns_400(self) -> None:
+        self.create_member(
+            user=self.user, organization=self.organization, role="owner", teams=[self.team]
+        )
+        self.login_as(self.user)
+
+        ard = AlertRuleDetector.objects.get(alert_rule_id=self.alert_rule.id)
+        # Create another Rule linked to the same Detector.
+        AlertRuleDetector.objects.create(rule_id=999999, detector_id=ard.detector_id)
+
+        with self.feature("organizations:incidents"):
+            resp = self.get_error_response(
+                self.organization.slug, self.alert_rule.id, status_code=400
+            )
+
+        assert "Detector" in resp.data["detail"]
+        assert str(ard.detector_id) in resp.data["detail"]
+        # AlertRule was not deleted.
+        assert AlertRule.objects.filter(id=self.alert_rule.id).exists()