fix(api): Tighten alert mutation write scopes

src/sentry/api/bases/organization.py

@@ -2,7 +2,7 @@
 
 from collections.abc import Sequence
 from datetime import datetime
-from typing import Any, Literal, NotRequired, TypedDict, overload
+from typing import Any, Literal, NotRequired, TypeAlias, TypedDict, overload
 
 import sentry_sdk
 from django.core.cache import cache
@@ -29,6 +29,7 @@
 from sentry.models.project import Project
 from sentry.models.release import Release
 from sentry.models.releases.release_project import ReleaseProject
+from sentry.models.team import Team
 from sentry.organizations.services.organization import (
     RpcOrganization,
     RpcUserOrganizationContext,
@@ -45,6 +46,15 @@ class NoProjects(Exception):
     pass
 
 
+def _get_organization_id(
+    organization: Organization | RpcOrganization | RpcUserOrganizationContext,
+) -> int:
+    if isinstance(organization, RpcUserOrganizationContext):
+        return organization.organization.id
+
+    return organization.id
+
+
 class OrganizationPermission(DemoSafePermission):
     scope_map = {
         "GET": ["org:read", "org:write", "org:admin"],
@@ -212,30 +222,88 @@ class OrganizationSearchPermission(OrganizationPermission):
 class OrganizationDataExportPermission(OrganizationPermission):
     scope_map = {
         "GET": ["event:read", "event:write", "event:admin"],
-        "POST": ["event:read", "event:write", "event:admin"],
+        "POST": ["event:write", "event:admin"],
     }
 
 
+def _has_any_team_scope(request: Request, scope: str) -> bool:
+    if not request.access.team_ids_with_membership:
+        return False
+
+    teams = Team.objects.filter(id__in=request.access.team_ids_with_membership)
+    return any(request.access.has_team_scope(team, scope) for team in teams)
+
+
+ALERT_MUTATION_SCOPES = frozenset({"org:write", "alerts:write"})
+
+
+class _AlertMutationLookupMiss:
+    pass
+
+
+ALERT_MUTATION_LOOKUP_MISS = _AlertMutationLookupMiss()
+AlertMutationProjectLookup: TypeAlias = Sequence[Project] | None | _AlertMutationLookupMiss
+
+
+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 | _AlertMutationLookupMiss:
+    get_projects = getattr(view, "get_alert_mutation_projects", None)
+    if not callable(get_projects):
+        return None
+
+    projects = get_projects(request, organization)
+    if projects is ALERT_MUTATION_LOOKUP_MISS:
+        return ALERT_MUTATION_LOOKUP_MISS
+    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"],
-        # grant org:read permission, but raise permission denied if the members aren't allowed
-        # to create alerts and the user isn't a team admin
-        "POST": ["org:read", "org:write", "org:admin", "alerts:write"],
+        "POST": ["org:write", "org:admin", "alerts:write"],
         "PUT": ["org:write", "org:admin", "alerts:write"],
         "DELETE": ["org:write", "org:admin", "alerts:write"],
     }
 
+    def has_object_permission(
+        self,
+        request: Request,
+        view: APIView,
+        organization: Organization | RpcOrganization | RpcUserOrganizationContext,
+    ) -> bool:
+        if super().has_object_permission(request, view, organization):
+            return True
 
-class OrganizationDetectorPermission(OrganizationPermission):
-    scope_map = {
-        "GET": ["org:read", "org:write", "org:admin", "alerts:read"],
-        # grant org:read permission, but raise permission denied if the members aren't allowed
-        # to create alerts and the user isn't a team admin
-        "POST": ["org:read", "org:write", "org:admin", "alerts:write"],
-        "PUT": ["org:read", "org:write", "org:admin", "alerts:write"],
-        "DELETE": ["org:read", "org:write", "org:admin", "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 ALERT_MUTATION_LOOKUP_MISS:
+            if _has_any_team_scope(request, "alerts:write"):
+                raise ResourceDoesNotExist
+            return False
+        if isinstance(project_scoped_access, bool):
+            return project_scoped_access
+
+        return bool(getattr(view, "allow_any_team_alert_write_fallback", False)) and (
+            _has_any_team_scope(request, "alerts:write")
+        )
+
+
+class OrganizationDetectorPermission(OrganizationAlertRulePermission):
+    pass
 
 
 class OrgAuthTokenPermission(OrganizationPermission):

src/sentry/incidents/endpoints/bases.py

@@ -1,16 +1,25 @@
 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
 from sentry.api.api_owners import ApiOwner
-from sentry.api.bases.organization import OrganizationAlertRulePermission, OrganizationEndpoint
+from sentry.api.bases.organization import (
+    ALERT_MUTATION_LOOKUP_MISS,
+    AlertMutationProjectLookup,
+    OrganizationAlertRulePermission,
+    OrganizationEndpoint,
+    _get_organization_id,
+)
 from sentry.api.bases.project import ProjectAlertRulePermission, ProjectEndpoint
 from sentry.api.exceptions import ResourceDoesNotExist
 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.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
@@ -24,6 +33,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 +94,37 @@ def convert_args(
 class OrganizationAlertRuleEndpoint(OrganizationEndpoint):
     permission_classes = (OrganizationAlertRulePermission,)
 
+    def get_alert_mutation_projects(
+        self,
+        request: Request,
+        organization: Organization | RpcOrganization | RpcUserOrganizationContext,
+    ) -> AlertMutationProjectLookup:
+        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 ALERT_MUTATION_LOOKUP_MISS
+
+        try:
+            organization_id = _get_organization_id(organization)
+            alert_rule = AlertRule.objects.prefetch_related("projects").get(
+                organization_id=organization_id, id=validated_alert_rule_id
+            )
+            return [alert_rule.projects.get()]
+        except (
+            AlertRule.DoesNotExist,
+            AlertRule.MultipleObjectsReturned,
+            Project.DoesNotExist,
+            Project.MultipleObjectsReturned,
+        ):
+            return ALERT_MUTATION_LOOKUP_MISS
+
     def convert_args(
         self, request: Request, alert_rule_id: int, *args: Any, **kwargs: Any
     ) -> tuple[tuple[Any, ...], dict[str, Any]]:
@@ -165,6 +207,57 @@ 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 | RpcOrganization | RpcUserOrganizationContext,
+    ) -> AlertMutationProjectLookup:
+        projects = super().get_alert_mutation_projects(request, organization)
+        if projects is not None and projects is not ALERT_MUTATION_LOOKUP_MISS:
+            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 ALERT_MUTATION_LOOKUP_MISS
+
+        organization_id = _get_organization_id(organization)
+        ard = (
+            AlertRuleDetector.objects.select_related("detector__project")
+            .filter(
+                alert_rule_id=validated_alert_rule_id,
+                detector__project__organization_id=organization_id,
+            )
+            .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 ALERT_MUTATION_LOOKUP_MISS
+
+        detector = (
+            Detector.objects.select_related("project")
+            .filter(
+                id=detector_id,
+                project__organization_id=organization_id,
+            )
+            .first()
+        )
+        if detector is None:
+            return ALERT_MUTATION_LOOKUP_MISS
+
+        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
@@ -323,6 +324,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/incidents/serializers/alert_rule.py

@@ -51,6 +51,8 @@
 
 logger = logging.getLogger(__name__)
 
+ALERT_RULE_PROJECT_SCOPES = ("project:read", "org:write", "org:admin", "alerts:write")
+
 
 class AlertRuleSerializer(SnubaQueryValidator, CamelSnakeModelSerializer[AlertRule]):
     """
@@ -62,7 +64,7 @@ class AlertRuleSerializer(SnubaQueryValidator, CamelSnakeModelSerializer[AlertRu
 
     environment = serializers.CharField(required=False, allow_null=True)
     projects = serializers.ListField(
-        child=ProjectField(scope="project:read"),
+        child=ProjectField(scope=ALERT_RULE_PROJECT_SCOPES),
         required=False,
         max_length=1,
     )

src/sentry/issues/endpoints/project_user_issue.py

@@ -180,7 +180,7 @@ class WebVitalsIssueDataSerializer(ProjectUserIssueRequestSerializer):
 class ProjectUserIssuePermission(ProjectPermission):
     scope_map = {
         "GET": [],
-        "POST": ["event:read", "event:write", "event:admin"],
+        "POST": ["event:write", "event:admin"],
         "PUT": [],
         "DELETE": [],
     }

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
@@ -40,6 +40,7 @@
 from sentry.incidents.endpoints.bases import OrganizationAlertRuleBaseEndpoint
 from sentry.models.environment import Environment
 from sentry.models.organization import Organization
+from sentry.models.project import Project
 from sentry.monitors.models import (
     DEFAULT_STATUS_ORDER,
     MONITOR_ENVIRONMENT_ORDERING,
@@ -326,11 +327,23 @@ def put(self, request: AuthenticatedHttpRequest, organization) -> Response:
 
         result = dict(validator.validated_data)
 
-        projects = self.get_projects(request, organization, include_all_accessible=True)
-        project_ids = [project.id for project in projects]
-
         monitor_guids = result.pop("ids", [])
-        monitors = list(Monitor.objects.filter(guid__in=monitor_guids, project_id__in=project_ids))
+        monitors = list(
+            Monitor.objects.filter(guid__in=monitor_guids, organization_id=organization.id)
+        )
+        projects_by_id = {
+            project.id: project
+            for project in Project.objects.filter(
+                id__in={monitor.project_id for monitor in monitors},
+                organization_id=organization.id,
+            )
+        }
+        if not all(
+            (project := projects_by_id.get(monitor.project_id))
+            and request.access.has_any_project_scope(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/monitors/validators.py

@@ -74,6 +74,7 @@
 INTERVAL_NAMES = ("year", "month", "week", "day", "hour", "minute")
 
 CRONTAB_WHITESPACE = re.compile(r"\s+")
+MONITOR_PROJECT_SCOPES = ("project:read", "org:write", "org:admin", "alerts:write")
 
 # XXX(dcramer): @reboot is not supported (as it cannot be)
 NONSTANDARD_CRONTAB_SCHEDULES = {
@@ -312,7 +313,7 @@ def validate(self, attrs):
 @extend_schema_serializer(exclude_fields=["alert_rule"])
 class MonitorValidator(CamelSnakeSerializer):
     project = ProjectField(
-        scope="project:read",
+        scope=MONITOR_PROJECT_SCOPES,
         required=True,
         help_text="The project slug to associate the monitor to.",
     )

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": ["project:read", "project:write", "project:admin"],
+        "DELETE": ["event:admin"],
     }
 
 

src/sentry/replays/endpoints/project_replay_summary.py

@@ -52,6 +52,13 @@ class ProjectReplaySummaryEndpoint(ProjectReplayEndpoint):
         "GET": ApiPublishStatus.EXPERIMENTAL,
         "POST": ApiPublishStatus.EXPERIMENTAL,
     }
+    readonly_mutation_scope_exceptions = {
+        "POST": (
+            "POST starts replay-summary generation but only derives summary data from existing "
+            "replay/event data. It intentionally follows replay read access instead of requiring "
+            "a separate write capability."
+        )
+    }
     permission_classes = (ReplaySummaryPermission,)
 
     def __init__(self, **kw) -> None: