fix(api): Unwrap org context in alert mutations

Normalize RpcUserOrganizationContext to the underlying organization before resolving target projects for alert-rule and detector mutation permissions. Without that, the new project-scoped alerts:write path breaks during organization object-permission checks because those hooks run before convert_args swaps in the concrete organization.

Add team-admin session regressions that force the project-scoped alerts:write authorization path for alert rule and detector updates and deletes, including the workflow-engine fake-detector alert-rule route.

Co-Authored-By: Codex <noreply@openai.com>

Author: dcramer

src/sentry/incidents/endpoints/bases.py

@@ -14,11 +14,21 @@
 from sentry.incidents.models.alert_rule import AlertRule
 from sentry.models.organization import Organization
 from sentry.models.project import Project
+from sentry.organizations.services.organization import RpcOrganization, RpcUserOrganizationContext
 from sentry.workflow_engine.endpoints.utils.ids import to_valid_int_id
 from sentry.workflow_engine.models.alertrule_detector import AlertRuleDetector
 from sentry.workflow_engine.models.detector import Detector
 
 
+def _get_organization_id(
+    organization: Organization | RpcOrganization | RpcUserOrganizationContext,
+) -> int:
+    if isinstance(organization, RpcUserOrganizationContext):
+        return organization.organization.id
+
+    return organization.id
+
+
 class OrganizationAlertRuleBaseEndpoint(OrganizationEndpoint):
     """
     Base endpoint for organization-scoped alert rule creation.
@@ -89,7 +99,9 @@ class OrganizationAlertRuleEndpoint(OrganizationEndpoint):
     permission_classes = (OrganizationAlertRulePermission,)
 
     def get_alert_mutation_projects(
-        self, request: Request, organization: Organization
+        self,
+        request: Request,
+        organization: Organization | RpcOrganization | RpcUserOrganizationContext,
     ) -> Sequence[Project] | None:
         if request.method not in {"PUT", "DELETE"}:
             return None
@@ -104,8 +116,9 @@ def get_alert_mutation_projects(
             return None
 
         try:
+            organization_id = _get_organization_id(organization)
             alert_rule = AlertRule.objects.prefetch_related("projects").get(
-                organization=organization, id=validated_alert_rule_id
+                organization_id=organization_id, id=validated_alert_rule_id
             )
             return [alert_rule.projects.get()]
         except (
@@ -192,7 +205,9 @@ class WorkflowEngineOrganizationAlertRuleEndpoint(OrganizationAlertRuleEndpoint)
     workflow_engine_method_flags: dict[str, str] = {}
 
     def get_alert_mutation_projects(
-        self, request: Request, organization: Organization
+        self,
+        request: Request,
+        organization: Organization | RpcOrganization | RpcUserOrganizationContext,
     ) -> Sequence[Project] | None:
         projects = super().get_alert_mutation_projects(request, organization)
         if projects is not None:
@@ -210,11 +225,12 @@ def get_alert_mutation_projects(
         except RestFrameworkValidationError:
             return None
 
+        organization_id = _get_organization_id(organization)
         ard = (
             AlertRuleDetector.objects.select_related("detector__project")
             .filter(
                 alert_rule_id=validated_alert_rule_id,
-                detector__project__organization=organization,
+                detector__project__organization_id=organization_id,
             )
             .first()
         )
@@ -230,7 +246,7 @@ def get_alert_mutation_projects(
             Detector.objects.select_related("project")
             .filter(
                 id=detector_id,
-                project__organization=organization,
+                project__organization_id=organization_id,
             )
             .first()
         )

src/sentry/workflow_engine/endpoints/organization_detector_details.py

@@ -27,6 +27,7 @@
 from sentry.issues import grouptype
 from sentry.models.organization import Organization
 from sentry.models.project import Project
+from sentry.organizations.services.organization import RpcOrganization, RpcUserOrganizationContext
 from sentry.utils.audit import create_audit_entry
 from sentry.workflow_engine.endpoints.serializers.detector_serializer import DetectorSerializer
 from sentry.workflow_engine.endpoints.utils.ids import to_valid_int_id
@@ -39,6 +40,15 @@
 from sentry.workflow_engine.models import Detector
 
 
+def _get_organization_id(
+    organization: Organization | RpcOrganization | RpcUserOrganizationContext,
+) -> int:
+    if isinstance(organization, RpcUserOrganizationContext):
+        return organization.organization.id
+
+    return organization.id
+
+
 def remove_detector(request: Request, organization: Organization, detector: Detector) -> Response:
     """
     Delete a given detector. This method is used by the OrganizationAlertRuleDetailsEndpoint DELETE method
@@ -96,7 +106,9 @@ def get_detector_validator(
 @extend_schema(tags=["Monitors"])
 class OrganizationDetectorDetailsEndpoint(OrganizationEndpoint):
     def get_alert_mutation_projects(
-        self, request: Request, organization: Organization
+        self,
+        request: Request,
+        organization: Organization | RpcOrganization | RpcUserOrganizationContext,
     ) -> list[Project] | None:
         if request.method not in {"PUT", "DELETE"}:
             return None
@@ -110,11 +122,12 @@ def get_alert_mutation_projects(
         except ValidationError:
             return None
 
+        organization_id = _get_organization_id(organization)
         detector = (
             Detector.objects.select_related("project")
             .filter(
                 id=validated_detector_id,
-                project__organization_id=organization.id,
+                project__organization_id=organization_id,
             )
             .first()
         )

tests/sentry/incidents/endpoints/test_organization_alert_rule_details.py

@@ -1006,6 +1006,28 @@ def test_get_shows_count_when_stored_as_upsampled_count(
 class AlertRuleDetailsPutEndpointTest(AlertRuleDetailsBase):
     method = "put"
 
+    def test_team_admin_can_update_with_project_scoped_alerts_write(self) -> None:
+        team_admin_user = self.create_user()
+        self.create_member(
+            user=team_admin_user,
+            organization=self.organization,
+            role="member",
+            team_roles=[(self.team, "admin")],
+        )
+        self.organization.update_option("sentry:alerts_member_write", False)
+        self.login_as(team_admin_user)
+
+        with self.feature("organizations:incidents"):
+            response = self.client.put(
+                reverse(self.endpoint, args=[self.organization.slug, self.alert_rule.id]),
+                data=self.valid_params,
+                format="json",
+            )
+
+        assert response.status_code == 200
+        self.alert_rule.refresh_from_db()
+        assert self.alert_rule.name == self.valid_params["name"]
+
     def test_update_requires_alerts_write_scope_for_tokens(self) -> None:
         self.create_member(
             user=self.user, organization=self.organization, role="owner", teams=[self.team]
@@ -2647,6 +2669,27 @@ def test_error_response_from_sentry_app(self) -> None:
 class AlertRuleDetailsDeleteEndpointTest(AlertRuleDetailsBase):
     method = "delete"
 
+    def test_team_admin_can_delete_workflow_engine_detector_with_project_scoped_alerts_write(
+        self,
+    ) -> None:
+        team_admin_user = self.create_user()
+        self.create_member(
+            user=team_admin_user,
+            organization=self.organization,
+            role="member",
+            team_roles=[(self.team, "admin")],
+        )
+        self.organization.update_option("sentry:alerts_member_write", False)
+        self.login_as(team_admin_user)
+
+        detector = self.create_detector(project=self.project, type=MetricIssue.slug)
+        data_source = self.create_data_source()
+        self.create_data_source_detector(data_source, detector)
+        fake_detector_id = get_fake_id_from_object_id(detector.id)
+
+        with self.feature({"organizations:workflow-engine-rule-serializers": True}):
+            self.get_success_response(self.organization.slug, fake_detector_id, status_code=204)
+
     def test_delete_denies_alerts_write_scope_for_other_team_projects(self) -> None:
         team_admin_user = self.create_user()
         self.create_member(

tests/sentry/workflow_engine/endpoints/test_organization_detector_details.py

@@ -504,6 +504,28 @@ def test_update_requires_alerts_write_scope_for_tokens(self) -> None:
 
         assert response.status_code == 403
 
+    def test_team_admin_can_update_with_project_scoped_alerts_write(self) -> None:
+        team_admin_user = self.create_user(is_superuser=False)
+        self.create_member(
+            user=team_admin_user,
+            organization=self.organization,
+            role="member",
+            team_roles=[(self.team, "admin")],
+        )
+        self.organization.update_option("sentry:alerts_member_write", False)
+        self.login_as(team_admin_user)
+
+        with self.tasks():
+            response = self.client.put(
+                reverse(self.endpoint, args=[self.organization.slug, self.detector.id]),
+                data={"enabled": False},
+                format="json",
+            )
+
+        assert response.status_code == 200
+        self.detector.refresh_from_db()
+        assert self.detector.enabled is False
+
     def test_update_allows_alerts_write_scope_for_tokens(self) -> None:
         token = self._create_token("alerts:write")
 
@@ -1029,6 +1051,24 @@ def test_update_data_source_marks_user_updated_when_snapshot_exists(
 class OrganizationDetectorDetailsDeleteTest(OrganizationDetectorDetailsBaseTest):
     method = "DELETE"
 
+    def test_team_admin_can_delete_with_project_scoped_alerts_write(self) -> None:
+        team_admin_user = self.create_user(is_superuser=False)
+        self.create_member(
+            user=team_admin_user,
+            organization=self.organization,
+            role="member",
+            team_roles=[(self.team, "admin")],
+        )
+        self.organization.update_option("sentry:alerts_member_write", False)
+        self.login_as(team_admin_user)
+
+        with outbox_runner():
+            response = self.client.delete(
+                reverse(self.endpoint, args=[self.organization.slug, self.detector.id])
+            )
+
+        assert response.status_code == 204
+
     def test_delete_denies_alerts_write_scope_for_other_team_projects(self) -> None:
         team_admin_user = self.create_user(is_superuser=False)
         self.create_member(