fix(api): tighten published mutation scopes

src/sentry/api/bases/organization.py

@@ -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,
@@ -212,31 +213,100 @@ 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"})
+
+
+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"],
-        # 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
+
+        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")
+        )
+
 
 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"],
+        "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
+
+        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")
+        )
+
 
 class OrgAuthTokenPermission(OrganizationPermission):
     scope_map = {

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,11 +13,22 @@
 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
 
 
+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.
@@ -24,6 +37,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 +98,37 @@ def convert_args(
 class OrganizationAlertRuleEndpoint(OrganizationEndpoint):
     permission_classes = (OrganizationAlertRulePermission,)
 
+    def get_alert_mutation_projects(
+        self,
+        request: Request,
+        organization: Organization | RpcOrganization | RpcUserOrganizationContext,
+    ) -> 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:
+            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 None
+
     def convert_args(
         self, request: Request, alert_rule_id: int, *args: Any, **kwargs: Any
     ) -> tuple[tuple[Any, ...], dict[str, Any]]:
@@ -158,6 +204,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,
+    ) -> 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
+
+        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 None
+
+        detector = (
+            Detector.objects.select_related("project")
+            .filter(
+                id=detector_id,
+                project__organization_id=organization_id,
+            )
+            .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/incidents/serializers/alert_rule.py

@@ -45,6 +45,8 @@
 
 logger = logging.getLogger(__name__)
 
+ALERT_RULE_PROJECT_SCOPES = ("project:read", "org:write", "org:admin", "alerts:write")
+
 
 class AlertRuleSerializer(SnubaQueryValidator, CamelSnakeModelSerializer[AlertRule]):
     """
@@ -56,7 +58,7 @@ class AlertRuleSerializer(SnubaQueryValidator, CamelSnakeModelSerializer[AlertRu
 
     environment = EnvironmentField(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
@@ -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/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: