fix(api): Tighten alert mutations and replay delete scope

Limit org-scoped alert and detector mutations to either the target project's alerts:write access or an explicit endpoint opt-in for the older team-scoped fallback. This closes the cross-team mutation path on alert rule and detector details, and adds project-scoped checks for the other shared consumers that still relied on the broad organization permission.

Align replay deletion with the rest of the event delete surface by requiring event:admin instead of event:write. Add regression coverage for the scope changes and the cross-project authorization cases.

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

Author: dcramer

src/sentry/api/bases/organization.py

@@ -225,6 +225,31 @@ def _has_any_team_scope(request: Request, scope: str) -> bool:
     return any(request.access.has_team_scope(team, scope) for team in teams)
 
 
+ALERT_MUTATION_SCOPES = frozenset({"org:write", "alerts:write"})
+
+
+def _has_project_alert_write_access(request: Request, projects: Sequence[Project]) -> bool:
+    return bool(projects) and all(
+        request.access.has_any_project_scope(project, ALERT_MUTATION_SCOPES) for project in projects
+    )
+
+
+def _has_view_project_scoped_alert_write(
+    request: Request,
+    view: APIView,
+    organization: Organization | RpcOrganization | RpcUserOrganizationContext,
+) -> bool | None:
+    get_projects = getattr(view, "get_alert_mutation_projects", None)
+    if not callable(get_projects):
+        return None
+
+    projects = get_projects(request, organization)
+    if projects is None:
+        return None
+
+    return _has_project_alert_write_access(request, projects)
+
+
 class OrganizationAlertRulePermission(OrganizationPermission):
     scope_map = {
         "GET": ["org:read", "org:write", "org:admin", "alerts:read"],
@@ -242,8 +267,15 @@ def has_object_permission(
         if super().has_object_permission(request, view, organization):
             return True
 
-        return request.method in {"POST", "PUT", "DELETE"} and _has_any_team_scope(
-            request, "alerts:write"
+        if request.method not in {"POST", "PUT", "DELETE"}:
+            return False
+
+        project_scoped_access = _has_view_project_scoped_alert_write(request, view, organization)
+        if project_scoped_access is not None:
+            return project_scoped_access
+
+        return bool(getattr(view, "allow_any_team_alert_write_fallback", False)) and (
+            _has_any_team_scope(request, "alerts:write")
         )
 
 
@@ -264,8 +296,15 @@ def has_object_permission(
         if super().has_object_permission(request, view, organization):
             return True
 
-        return request.method in {"POST", "PUT", "DELETE"} and _has_any_team_scope(
-            request, "alerts:write"
+        if request.method not in {"POST", "PUT", "DELETE"}:
+            return False
+
+        project_scoped_access = _has_view_project_scoped_alert_write(request, view, organization)
+        if project_scoped_access is not None:
+            return project_scoped_access
+
+        return bool(getattr(view, "allow_any_team_alert_write_fallback", False)) and (
+            _has_any_team_scope(request, "alerts:write")
         )
 
 

src/sentry/incidents/endpoints/bases.py

@@ -1,6 +1,8 @@
+from collections.abc import Sequence
 from typing import Any
 
 from rest_framework.exceptions import PermissionDenied
+from rest_framework.exceptions import ValidationError as RestFrameworkValidationError
 from rest_framework.request import Request
 
 from sentry import features
@@ -11,6 +13,7 @@
 from sentry.incidents.endpoints.serializers.utils import get_object_id_from_fake_id
 from sentry.incidents.models.alert_rule import AlertRule
 from sentry.models.organization import Organization
+from sentry.models.project import Project
 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
@@ -24,6 +27,8 @@ class OrganizationAlertRuleBaseEndpoint(OrganizationEndpoint):
     org-level permissions and team admin project-scoped permissions.
     """
 
+    allow_any_team_alert_write_fallback = True
+
     def check_can_create_alert(self, request: Request, organization: Organization) -> None:
         """
         Determine if the requesting user has access to alert creation. If the request does not have the "alerts:write"
@@ -83,6 +88,34 @@ def convert_args(
 class OrganizationAlertRuleEndpoint(OrganizationEndpoint):
     permission_classes = (OrganizationAlertRulePermission,)
 
+    def get_alert_mutation_projects(
+        self, request: Request, organization: Organization
+    ) -> Sequence[Project] | None:
+        if request.method not in {"PUT", "DELETE"}:
+            return None
+
+        raw_alert_rule_id = self.kwargs.get("alert_rule_id")
+        if raw_alert_rule_id is None:
+            return None
+
+        try:
+            validated_alert_rule_id = to_valid_int_id("alert_rule_id", raw_alert_rule_id)
+        except RestFrameworkValidationError:
+            return None
+
+        try:
+            alert_rule = AlertRule.objects.prefetch_related("projects").get(
+                organization=organization, id=validated_alert_rule_id
+            )
+            return [alert_rule.projects.get()]
+        except (
+            AlertRule.DoesNotExist,
+            AlertRule.MultipleObjectsReturned,
+            Project.DoesNotExist,
+            Project.MultipleObjectsReturned,
+        ):
+            return None
+
     def convert_args(
         self, request: Request, alert_rule_id: int, *args: Any, **kwargs: Any
     ) -> tuple[tuple[Any, ...], dict[str, Any]]:
@@ -158,6 +191,54 @@ class WorkflowEngineOrganizationAlertRuleEndpoint(OrganizationAlertRuleEndpoint)
     # with the broad workflow-engine-rule-serializers flag.
     workflow_engine_method_flags: dict[str, str] = {}
 
+    def get_alert_mutation_projects(
+        self, request: Request, organization: Organization
+    ) -> Sequence[Project] | None:
+        projects = super().get_alert_mutation_projects(request, organization)
+        if projects is not None:
+            return projects
+
+        if request.method not in {"PUT", "DELETE"}:
+            return None
+
+        raw_alert_rule_id = self.kwargs.get("alert_rule_id")
+        if raw_alert_rule_id is None:
+            return None
+
+        try:
+            validated_alert_rule_id = to_valid_int_id("alert_rule_id", raw_alert_rule_id)
+        except RestFrameworkValidationError:
+            return None
+
+        ard = (
+            AlertRuleDetector.objects.select_related("detector__project")
+            .filter(
+                alert_rule_id=validated_alert_rule_id,
+                detector__project__organization=organization,
+            )
+            .first()
+        )
+        if ard is not None:
+            return [ard.detector.project]
+
+        try:
+            detector_id = get_object_id_from_fake_id(validated_alert_rule_id)
+        except ValueError:
+            return None
+
+        detector = (
+            Detector.objects.select_related("project")
+            .filter(
+                id=detector_id,
+                project__organization=organization,
+            )
+            .first()
+        )
+        if detector is None:
+            return None
+
+        return [detector.project]
+
     def convert_args(
         self, request: Request, alert_rule_id: int, *args: Any, **kwargs: Any
     ) -> tuple[tuple[Any, ...], dict[str, Any]]:

src/sentry/incidents/endpoints/organization_alert_rule_details.py

@@ -13,6 +13,7 @@
 from sentry.api.api_owners import ApiOwner
 from sentry.api.api_publish_status import ApiPublishStatus
 from sentry.api.base import cell_silo_endpoint
+from sentry.api.bases.organization import ALERT_MUTATION_SCOPES
 from sentry.api.fields.actor import OwnerActorField
 from sentry.api.serializers import serialize
 from sentry.api.serializers.rest_framework.project import ProjectField
@@ -325,6 +326,11 @@ def wrapper(
         if not request.access.has_project_access(project):
             return Response(status=status.HTTP_403_FORBIDDEN)
 
+        if request.method in {"PUT", "DELETE"} and not request.access.has_any_project_scope(
+            project, ALERT_MUTATION_SCOPES
+        ):
+            return Response(status=status.HTTP_403_FORBIDDEN)
+
         return func(self, request, organization, alert_rule)
 
     return wrapper

src/sentry/monitors/endpoints/organization_monitor_index.py

@@ -18,7 +18,7 @@
 from sentry.api.api_publish_status import ApiPublishStatus
 from sentry.api.base import cell_silo_endpoint
 from sentry.api.bases import NoProjects
-from sentry.api.bases.organization import OrganizationAlertRulePermission
+from sentry.api.bases.organization import ALERT_MUTATION_SCOPES, OrganizationAlertRulePermission
 from sentry.api.helpers.teams import get_teams
 from sentry.api.paginator import OffsetPaginator
 from sentry.api.serializers import serialize
@@ -331,6 +331,11 @@ def put(self, request: AuthenticatedHttpRequest, organization) -> Response:
 
         monitor_guids = result.pop("ids", [])
         monitors = list(Monitor.objects.filter(guid__in=monitor_guids, project_id__in=project_ids))
+        if not all(
+            request.access.has_any_project_scope(monitor.project, ALERT_MUTATION_SCOPES)
+            for monitor in monitors
+        ):
+            return self.respond(status=403)
 
         status = result.get("status")
         # If enabling monitors, ensure we can assign all before moving forward

src/sentry/replays/endpoints/project_replay_details.py

@@ -23,7 +23,7 @@ class ReplayDetailsPermission(ProjectPermission):
         "GET": ["project:read", "project:write", "project:admin"],
         "POST": ["project:write", "project:admin"],
         "PUT": ["project:write", "project:admin"],
-        "DELETE": ["event:write", "event:admin"],
+        "DELETE": ["event:admin"],
     }
 
 

src/sentry/seer/endpoints/organization_events_anomalies.py

@@ -6,7 +6,7 @@
 from sentry.api.api_owners import ApiOwner
 from sentry.api.api_publish_status import ApiPublishStatus
 from sentry.api.base import cell_silo_endpoint
-from sentry.api.bases.organization import OrganizationAlertRulePermission
+from sentry.api.bases.organization import ALERT_MUTATION_SCOPES, OrganizationAlertRulePermission
 from sentry.api.bases.organization_events import OrganizationEventsEndpointBase
 from sentry.api.exceptions import ResourceDoesNotExist
 from sentry.api.serializers.base import serialize
@@ -33,10 +33,22 @@ class OrganizationEventsAnomaliesEndpoint(OrganizationEventsEndpointBase):
     publish_status = {
         "POST": ApiPublishStatus.EXPERIMENTAL,
     }
+    allow_any_team_alert_write_fallback = True
     # This POST previews anomaly-detection config used while authoring metric
     # alerts/detectors, so it intentionally follows alert-write permissions.
     permission_classes = (OrganizationAlertRulePermission,)
 
+    def get_alert_mutation_projects(self, request: Request, organization: Organization):
+        raw_project_id = request.data.get("project_id")
+        if raw_project_id is None:
+            return None
+
+        try:
+            project_id = to_valid_int_id("project_id", raw_project_id)
+            return self.get_projects(request, organization, project_ids={project_id})
+        except Exception:
+            return None
+
     @extend_schema(
         operation_id="Identify anomalies in historical data",
         parameters=[GlobalParams.ORG_ID_OR_SLUG],
@@ -94,6 +106,11 @@ def post(self, request: Request, organization: Organization) -> Response:
         projects = self.get_projects(request, organization, project_ids={project_id})
         if not projects:
             return Response({"detail": "Invalid project"}, status=400)
+        if not all(
+            request.access.has_any_project_scope(project, ALERT_MUTATION_SCOPES)
+            for project in projects
+        ):
+            return Response(status=403)
 
         anomalies = get_historical_anomaly_data_from_seer_preview(
             current_data=current_data,

src/sentry/uptime/endpoints/organization_uptime_alert_preview_check.py

@@ -30,6 +30,7 @@
 @cell_silo_endpoint
 class OrganizationUptimeAlertPreviewCheckEndpoint(OrganizationEndpoint):
     owner = ApiOwner.CRONS
+    allow_any_team_alert_write_fallback = True
     # This POST previews monitor creation and validation, so it intentionally
     # uses the same permission surface as creating the alert itself.
     permission_classes = (OrganizationAlertRulePermission,)

src/sentry/uptime/endpoints/organization_uptime_assertion_suggestions.py

@@ -50,6 +50,7 @@ class OrganizationUptimeAssertionSuggestionsEndpoint(OrganizationEndpoint):
     """
 
     owner = ApiOwner.CRONS
+    allow_any_team_alert_write_fallback = True
     # This POST is part of the uptime monitor authoring flow, so it should
     # track the same alert-write permission as the monitor it helps create.
     permission_classes = (OrganizationAlertRulePermission,)

src/sentry/workflow_engine/endpoints/organization_detector_details.py

@@ -95,6 +95,34 @@ def get_detector_validator(
 @cell_silo_endpoint
 @extend_schema(tags=["Monitors"])
 class OrganizationDetectorDetailsEndpoint(OrganizationEndpoint):
+    def get_alert_mutation_projects(
+        self, request: Request, organization: Organization
+    ) -> list[Project] | None:
+        if request.method not in {"PUT", "DELETE"}:
+            return None
+
+        raw_detector_id = self.kwargs.get("detector_id")
+        if raw_detector_id is None:
+            return None
+
+        try:
+            validated_detector_id = to_valid_int_id("detector_id", raw_detector_id)
+        except ValidationError:
+            return None
+
+        detector = (
+            Detector.objects.select_related("project")
+            .filter(
+                id=validated_detector_id,
+                project__organization_id=organization.id,
+            )
+            .first()
+        )
+        if detector is None:
+            return None
+
+        return [detector.project]
+
     def convert_args(
         self, request: Request, detector_id: str, *args: Any, **kwargs: Any
     ) -> tuple[tuple[Any, ...], dict[str, Organization | Detector]]:

src/sentry/workflow_engine/endpoints/organization_detector_index.py

@@ -150,6 +150,7 @@ class OrganizationDetectorIndexEndpoint(OrganizationEndpoint):
         "DELETE": ApiPublishStatus.PUBLIC,
     }
     owner = ApiOwner.ISSUES
+    allow_any_team_alert_write_fallback = True
 
     permission_classes = (OrganizationDetectorPermission,)