fix(dynamic-sampling): add project-level permission check before adding custom rule

src/sentry/api/endpoints/custom_rules.py

@@ -5,14 +5,14 @@
 import sentry_sdk
 from django.db import DatabaseError
 from rest_framework import serializers
-from rest_framework.permissions import BasePermission
 from rest_framework.request import Request
 from rest_framework.response import Response
 
 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 import OrganizationEndpoint
+from sentry.api.bases.organization import OrganizationPermission
 from sentry.exceptions import InvalidSearchQuery
 from sentry.models.dynamicsampling import (
     CUSTOM_RULE_DATE_FORMAT,
@@ -94,7 +94,7 @@ def validate(self, data):
         return data
 
 
-class CustomRulePermission(BasePermission):
+class CustomRulePermission(OrganizationPermission):
     scope_map = {
         "GET": [
             "org:read",
@@ -130,7 +130,10 @@ def post(self, request: Request, organization: Organization) -> Response:
             return Response(serializer.errors, status=400)
 
         query = serializer.validated_data["query"]
-        projects = serializer.validated_data.get("projects")
+        project_ids = serializer.validated_data.get("projects")
+
+        # project-level permission check
+        self.get_projects(request, organization, project_ids=set(project_ids))
 
         try:
             condition = get_rule_condition(query)
@@ -145,7 +148,7 @@ def post(self, request: Request, organization: Organization) -> Response:
                 condition=condition,
                 start=start,
                 end=end,
-                project_ids=projects,
+                project_ids=project_ids,
                 organization_id=organization.id,
                 num_samples=NUM_SAMPLES_PER_CUSTOM_RULE,
                 sample_rate=1.0,
@@ -154,7 +157,7 @@ def post(self, request: Request, organization: Organization) -> Response:
             )
 
             # schedule update for affected project configs
-            _schedule_invalidate_project_configs(organization, projects)
+            _schedule_invalidate_project_configs(organization, project_ids)
 
             return _rule_to_response(rule)
         except UnsupportedSearchQuery as e:
@@ -189,6 +192,9 @@ def get(self, request: Request, organization: Organization) -> Response:
         except ValueError:
             return Response({"projects": ["Invalid project id"]}, status=400)
 
+        # project-level permission check
+        self.get_projects(request, organization, project_ids=set(requested_projects_ids))
+
         if requested_projects_ids:
             org_rule = False
             invalid_projects = []

tests/sentry/api/endpoints/test_custom_rules.py

@@ -161,6 +161,28 @@ def test_does_not_find_rule_when_project_doesnt_match(self):
         )
         assert resp.status_code == 204
 
+    def test_disallow_when_no_project_access(self):
+        # disable Open Membership
+        self.organization.flags.allow_joinleave = False
+        self.organization.save()
+
+        # user has no access to the first project
+        user_no_team = self.create_user(is_superuser=False)
+        self.create_member(
+            user=user_no_team, organization=self.organization, role="member", teams=[]
+        )
+        self.login_as(user_no_team)
+
+        response = self.get_response(
+            self.organization.slug,
+            qs_params={
+                "query": "event.type:transaction environment:prod transaction:/hello",
+                "project": [self.project.id],
+            },
+        )
+        assert response.status_code == 403, response.data
+        assert response.data == {"detail": "You do not have permission to perform this action."}
+
 
 class CustomRulesEndpoint(APITestCase):
     """
@@ -202,6 +224,27 @@ def test_create(self):
         rule = rules[0]
         assert rule.external_rule_id == rule_id
 
+    def test_disallow_when_no_project_access(self):
+        # disable Open Membership
+        self.organization.flags.allow_joinleave = False
+        self.organization.save()
+
+        # user has no access to the first project
+        user_no_team = self.create_user(is_superuser=False)
+        self.create_member(
+            user=user_no_team, organization=self.organization, role="member", teams=[]
+        )
+        self.login_as(user_no_team)
+
+        request_data = {
+            "query": "event.type:transaction http.method:POST",
+            "projects": [self.project.id],
+            "period": "1h",
+        }
+        response = self.get_response(self.organization.slug, raw_data=request_data)
+        assert response.status_code == 403, response.data
+        assert response.data == {"detail": "You do not have permission to perform this action."}
+
     def test_updates_existing(self):
         """
         Test that the endpoint updates an existing rule if the same rule condition and projects is given