feat(replay): guard endpoints by granular access permissions

Author: shellmayr

src/sentry/replays/endpoints/organization_replay_count.py

@@ -21,6 +21,7 @@
 from sentry.models.organization import Organization
 from sentry.models.project import Project
 from sentry.ratelimits.config import RateLimitConfig
+from sentry.replays.permissions import has_replay_permission
 from sentry.replays.usecases.replay_counts import get_replay_counts
 from sentry.snuba.dataset import Dataset
 from sentry.types.ratelimit import RateLimit, RateLimitCategory
@@ -84,6 +85,8 @@ def get(self, request: Request, organization: Organization) -> Response:
         """
         if not features.has("organizations:session-replay", organization, actor=request.user):
             return Response(status=404)
+        if not has_replay_permission(organization, request.user):
+            return Response(status=403)
 
         try:
             snuba_params = self.get_snuba_params(request, organization)

src/sentry/replays/endpoints/organization_replay_details.py

@@ -10,13 +10,14 @@
 from sentry.api.api_owners import ApiOwner
 from sentry.api.api_publish_status import ApiPublishStatus
 from sentry.api.base import region_silo_endpoint
-from sentry.api.bases.organization import NoProjects, OrganizationEndpoint
+from sentry.api.bases.organization import NoProjects
 from sentry.apidocs.constants import RESPONSE_BAD_REQUEST, RESPONSE_FORBIDDEN, RESPONSE_NOT_FOUND
 from sentry.apidocs.examples.replay_examples import ReplayExamples
 from sentry.apidocs.parameters import GlobalParams, ReplayParams
 from sentry.apidocs.utils import inline_sentry_response_serializer
 from sentry.constants import ALL_ACCESS_PROJECTS
 from sentry.models.organization import Organization
+from sentry.replays.endpoints.organization_replay_endpoint import OrganizationReplayEndpoint
 from sentry.replays.lib.eap import read as eap_read
 from sentry.replays.lib.eap.snuba_transpiler import RequestMeta, Settings
 from sentry.replays.post_process import ReplayDetailsResponse, process_raw_response
@@ -144,7 +145,7 @@ def query_replay_instance_eap(
 
 @region_silo_endpoint
 @extend_schema(tags=["Replays"])
-class OrganizationReplayDetailsEndpoint(OrganizationEndpoint):
+class OrganizationReplayDetailsEndpoint(OrganizationReplayEndpoint):
     """
     The same data as ProjectReplayDetails, except no project is required.
     This works as we'll query for this replay_id across all projects in the
@@ -171,8 +172,8 @@ def get(self, request: Request, organization: Organization, replay_id: str) -> R
         """
         Return details on an individual replay.
         """
-        if not features.has("organizations:session-replay", organization, actor=request.user):
-            return Response(status=404)
+        if response := self.check_replay_access(request, organization):
+            return response
 
         try:
             filter_params = self.get_filter_params(

src/sentry/replays/endpoints/organization_replay_endpoint

@@ -0,0 +1,32 @@
+from rest_framework.request import Request
+from rest_framework.response import Response
+
+from sentry import features
+from sentry.api.bases.organization import OrganizationEndpoint
+from sentry.models.organization import Organization
+from sentry.replays.permissions import has_replay_permission
+
+
+class OrganizationReplayEndpoint(OrganizationEndpoint):
+    """
+    Base endpoint for replay-related organizationendpoints.
+    Provides centralized feature and permission checks for session replay access.
+    Added to ensure that all replay endpoints are consistent and follow the same pattern
+    for allowing granular user-based replay access control, in addition to the existing
+    role-based access control and feature flag-based access control.
+    """
+
+    def check_replay_access(self, request: Request, organization: Organization) -> Response | None:
+        """
+        Check if the session replay feature is enabled and user has replay permissions.
+        Returns a Response object if access should be denied, None if access is granted.
+        """
+        if not features.has(
+            "organizations:session-replay", organization, actor=request.user
+        ):
+            return Response(status=404)
+
+        if not has_replay_permission(organization, request.user):
+            return Response(status=403)
+
+        return None

src/sentry/replays/endpoints/organization_replay_events_meta.py

@@ -12,6 +12,7 @@
 from sentry.api.paginator import GenericOffsetPaginator
 from sentry.api.utils import reformat_timestamp_ms_to_isoformat
 from sentry.models.organization import Organization
+from sentry.replays.permissions import has_replay_permission
 
 
 @region_silo_endpoint
@@ -53,6 +54,9 @@ def get(self, request: Request, organization: Organization) -> Response:
         if not features.has("organizations:session-replay", organization, actor=request.user):
             return Response(status=404)
 
+        if not has_replay_permission(organization, request.user):
+            return Response(status=403)
+
         try:
             snuba_params = self.get_snuba_params(request, organization)
         except NoProjects:

src/sentry/replays/endpoints/organization_replay_index.py

@@ -6,18 +6,18 @@
 from rest_framework.request import Request
 from rest_framework.response import Response
 
-from sentry import features
 from sentry.api.api_owners import ApiOwner
 from sentry.api.api_publish_status import ApiPublishStatus
 from sentry.api.base import region_silo_endpoint
-from sentry.api.bases.organization import NoProjects, OrganizationEndpoint
+from sentry.api.bases.organization import NoProjects
 from sentry.api.event_search import parse_search_query
 from sentry.apidocs.constants import RESPONSE_BAD_REQUEST, RESPONSE_FORBIDDEN
 from sentry.apidocs.examples.replay_examples import ReplayExamples
 from sentry.apidocs.parameters import GlobalParams
 from sentry.apidocs.utils import inline_sentry_response_serializer
 from sentry.exceptions import InvalidSearchQuery
 from sentry.models.organization import Organization
+from sentry.replays.endpoints.organization_replay_endpoint import OrganizationReplayEndpoint
 from sentry.replays.post_process import ReplayDetailsResponse, process_raw_response
 from sentry.replays.query import query_replays_collection_paginated, replay_url_parser_config
 from sentry.replays.usecases.errors import handled_snuba_exceptions
@@ -28,7 +28,7 @@
 
 @region_silo_endpoint
 @extend_schema(tags=["Replays"])
-class OrganizationReplayIndexEndpoint(OrganizationEndpoint):
+class OrganizationReplayIndexEndpoint(OrganizationReplayEndpoint):
     owner = ApiOwner.REPLAY
     publish_status = {
         "GET": ApiPublishStatus.PUBLIC,
@@ -50,8 +50,9 @@ def get(self, request: Request, organization: Organization) -> Response:
         Return a list of replays belonging to an organization.
         """
 
-        if not features.has("organizations:session-replay", organization, actor=request.user):
-            return Response(status=404)
+        if response := self.check_replay_access(request, organization):
+            return response
+
         try:
             filter_params = self.get_filter_params(request, organization)
         except NoProjects:

src/sentry/replays/endpoints/organization_replay_selector_index.py

@@ -23,11 +23,10 @@
 )
 from snuba_sdk import Request as SnubaRequest
 
-from sentry import features
 from sentry.api.api_owners import ApiOwner
 from sentry.api.api_publish_status import ApiPublishStatus
 from sentry.api.base import region_silo_endpoint
-from sentry.api.bases.organization import NoProjects, OrganizationEndpoint
+from sentry.api.bases.organization import NoProjects
 from sentry.api.event_search import QueryToken, parse_search_query
 from sentry.api.paginator import GenericOffsetPaginator
 from sentry.apidocs.constants import RESPONSE_BAD_REQUEST, RESPONSE_FORBIDDEN
@@ -36,6 +35,7 @@
 from sentry.apidocs.utils import inline_sentry_response_serializer
 from sentry.exceptions import InvalidSearchQuery
 from sentry.models.organization import Organization
+from sentry.replays.endpoints.organization_replay_endpoint import OrganizationReplayEndpoint
 from sentry.replays.lib.new_query.conditions import IntegerScalar
 from sentry.replays.lib.new_query.fields import FieldProtocol, IntegerColumnField
 from sentry.replays.lib.new_query.parsers import parse_int
@@ -75,7 +75,7 @@ class ReplaySelectorResponse(TypedDict):
 
 @region_silo_endpoint
 @extend_schema(tags=["Replays"])
-class OrganizationReplaySelectorIndexEndpoint(OrganizationEndpoint):
+class OrganizationReplaySelectorIndexEndpoint(OrganizationReplayEndpoint):
     owner = ApiOwner.REPLAY
     publish_status = {
         "GET": ApiPublishStatus.PUBLIC,
@@ -106,8 +106,9 @@ def get_replay_filter_params(self, request, organization):
     )
     def get(self, request: Request, organization: Organization) -> Response:
         """Return a list of selectors for a given organization."""
-        if not features.has("organizations:session-replay", organization, actor=request.user):
-            return Response(status=404)
+        if response := self.check_replay_access(request, organization):
+            return response
+
         try:
             filter_params = self.get_replay_filter_params(request, organization)
         except NoProjects:

src/sentry/replays/endpoints/project_replay_clicks_index.py

@@ -24,11 +24,9 @@
 )
 from snuba_sdk.orderby import Direction
 
-from sentry import features
 from sentry.api.api_owners import ApiOwner
 from sentry.api.api_publish_status import ApiPublishStatus
 from sentry.api.base import region_silo_endpoint
-from sentry.api.bases.project import ProjectEndpoint
 from sentry.api.event_search import ParenExpression, QueryToken, SearchFilter, parse_search_query
 from sentry.api.paginator import GenericOffsetPaginator
 from sentry.apidocs.constants import RESPONSE_BAD_REQUEST, RESPONSE_FORBIDDEN, RESPONSE_NOT_FOUND
@@ -37,6 +35,7 @@
 from sentry.apidocs.utils import inline_sentry_response_serializer
 from sentry.exceptions import InvalidSearchQuery
 from sentry.models.project import Project
+from sentry.replays.endpoints.project_replay_endpoint import ProjectReplayEndpoint
 from sentry.replays.lib.new_query.errors import CouldNotParseValue, OperatorNotSupported
 from sentry.replays.lib.new_query.fields import FieldProtocol
 from sentry.replays.lib.query import attempt_compressed_condition
@@ -58,7 +57,7 @@ class ReplayClickResponse(TypedDict):
 
 @region_silo_endpoint
 @extend_schema(tags=["Replays"])
-class ProjectReplayClicksIndexEndpoint(ProjectEndpoint):
+class ProjectReplayClicksIndexEndpoint(ProjectReplayEndpoint):
     owner = ApiOwner.REPLAY
     publish_status = {
         "GET": ApiPublishStatus.PUBLIC,
@@ -85,10 +84,8 @@ class ProjectReplayClicksIndexEndpoint(ProjectEndpoint):
     )
     def get(self, request: Request, project: Project, replay_id: str) -> Response:
         """Retrieve a collection of RRWeb DOM node-ids and the timestamp they were clicked."""
-        if not features.has(
-            "organizations:session-replay", project.organization, actor=request.user
-        ):
-            return Response(status=404)
+        if response := self.check_replay_access(request, project):
+            return response
 
         filter_params = self.get_filter_params(request, project)
 

src/sentry/replays/endpoints/project_replay_details.py

@@ -8,10 +8,11 @@
 from sentry.api.api_owners import ApiOwner
 from sentry.api.api_publish_status import ApiPublishStatus
 from sentry.api.base import region_silo_endpoint
-from sentry.api.bases.project import ProjectEndpoint, ProjectPermission
+from sentry.api.bases.project import ProjectPermission
 from sentry.apidocs.constants import RESPONSE_NO_CONTENT, RESPONSE_NOT_FOUND
 from sentry.apidocs.parameters import GlobalParams, ReplayParams
 from sentry.models.project import Project
+from sentry.replays.endpoints.project_replay_endpoint import ProjectReplayEndpoint
 from sentry.replays.post_process import process_raw_response
 from sentry.replays.query import query_replay_instance
 from sentry.replays.tasks import delete_replay
@@ -29,7 +30,7 @@ class ReplayDetailsPermission(ProjectPermission):
 
 @region_silo_endpoint
 @extend_schema(tags=["Replays"])
-class ProjectReplayDetailsEndpoint(ProjectEndpoint):
+class ProjectReplayDetailsEndpoint(ProjectReplayEndpoint):
     owner = ApiOwner.REPLAY
     publish_status = {
         "DELETE": ApiPublishStatus.PUBLIC,
@@ -39,10 +40,8 @@ class ProjectReplayDetailsEndpoint(ProjectEndpoint):
     permission_classes = (ReplayDetailsPermission,)
 
     def get(self, request: Request, project: Project, replay_id: str) -> Response:
-        if not features.has(
-            "organizations:session-replay", project.organization, actor=request.user
-        ):
-            return Response(status=404)
+        if response := self.check_replay_access(request, project):
+            return response
 
         filter_params = self.get_filter_params(request, project)
 
@@ -87,11 +86,8 @@ def delete(self, request: Request, project: Project, replay_id: str) -> Response
         """
         Delete a replay.
         """
-
-        if not features.has(
-            "organizations:session-replay", project.organization, actor=request.user
-        ):
-            return Response(status=404)
+        if response := self.check_replay_access(request, project):
+            return response
 
         if has_archived_segment(project.id, replay_id):
             return Response(status=404)

src/sentry/replays/endpoints/project_replay_endpoint.py

@@ -0,0 +1,32 @@
+from rest_framework.request import Request
+from rest_framework.response import Response
+
+from sentry import features
+from sentry.api.bases.project import ProjectEndpoint
+from sentry.models.project import Project
+from sentry.replays.permissions import has_replay_permission
+
+
+class ProjectReplayEndpoint(ProjectEndpoint):
+    """
+    Base endpoint for replay-related endpoints.
+    Provides centralized feature and permission checks for session replay access.
+    Added to ensure that all replay endpoints are consistent and follow the same pattern
+    for allowing granular user-based replay access control, in addition to the existing
+    role-based access control and feature flag-based access control.
+    """
+
+    def check_replay_access(self, request: Request, project: Project) -> Response | None:
+        """
+        Check if the session replay feature is enabled and user has replay permissions.
+        Returns a Response object if access should be denied, None if access is granted.
+        """
+        if not features.has(
+            "organizations:session-replay", project.organization, actor=request.user
+        ):
+            return Response(status=404)
+
+        if not has_replay_permission(project.organization, request.user):
+            return Response(status=403)
+
+        return None

src/sentry/replays/endpoints/project_replay_jobs_delete.py

@@ -10,7 +10,9 @@
 from sentry.api.exceptions import ResourceDoesNotExist
 from sentry.api.paginator import OffsetPaginator
 from sentry.api.serializers import Serializer, serialize
+from sentry.replays.endpoints.project_replay_endpoint import ProjectReplayEndpoint
 from sentry.replays.models import ReplayDeletionJobModel
+from sentry.replays.permissions import has_replay_permission
 from sentry.replays.tasks import run_bulk_replay_delete_job
 
 
@@ -67,6 +69,9 @@ def get(self, request: Request, project) -> Response:
         """
         Retrieve a collection of replay delete jobs.
         """
+        if not has_replay_permission(project.organization, request.user):
+            return Response(status=403)
+
         queryset = ReplayDeletionJobModel.objects.filter(
             organization_id=project.organization_id, project_id=project.id
         )
@@ -85,6 +90,9 @@ def post(self, request: Request, project) -> Response:
         """
         Create a new replay deletion job.
         """
+        if not has_replay_permission(project.organization, request.user):
+            return Response(status=403)
+
         serializer = ReplayDeletionJobCreateSerializer(data=request.data)
         if not serializer.is_valid():
             return Response(serializer.errors, status=400)
@@ -124,7 +132,7 @@ def post(self, request: Request, project) -> Response:
 
 
 @region_silo_endpoint
-class ProjectReplayDeletionJobDetailEndpoint(ProjectEndpoint):
+class ProjectReplayDeletionJobDetailEndpoint(ProjectReplayEndpoint):
     owner = ApiOwner.REPLAY
     publish_status = {
         "GET": ApiPublishStatus.PRIVATE,
@@ -135,6 +143,9 @@ def get(self, request: Request, project, job_id: int) -> Response:
         """
         Fetch a replay delete job instance.
         """
+        if response := self.check_replay_access(request, project):
+            return response
+
         try:
             job = ReplayDeletionJobModel.objects.get(
                 id=job_id, organization_id=project.organization_id, project_id=project.id