fix(sentry-apps): gate error webhook subscriptions correctly

src/sentry/sentry_apps/api/endpoints/sentry_app_details.py

@@ -44,6 +44,7 @@
 from sentry.sentry_apps.logic import SentryAppUpdater
 from sentry.sentry_apps.models.sentry_app import SentryApp
 from sentry.sentry_apps.models.sentry_app_installation import SentryAppInstallation
+from sentry.sentry_apps.utils.webhooks import has_error_events
 from sentry.users.models.user import User
 from sentry.users.services.user.model import RpcUser
 from sentry.utils.audit import create_audit_entry
@@ -139,7 +140,7 @@ def put(
         )
         if (
             owner_context
-            and self._has_hook_events(request)
+            and has_error_events(request.data.get("events"))
             and not features.has(
                 "organizations:integrations-event-hooks",
                 owner_context.organization,
@@ -283,9 +284,3 @@ def delete(
             return Response(status=204)
 
         return Response({"detail": ["Published apps cannot be removed."]}, status=403)
-
-    def _has_hook_events(self, request: Request):
-        if not request.data.get("events"):
-            return False
-
-        return "error" in request.data["events"]

src/sentry/sentry_apps/api/endpoints/sentry_apps.py

@@ -26,6 +26,7 @@
 )
 from sentry.sentry_apps.logic import SentryAppCreator
 from sentry.sentry_apps.models.sentry_app import SentryApp
+from sentry.sentry_apps.utils.webhooks import has_error_events
 from sentry.users.models.user import User
 from sentry.users.services.user.model import RpcUser
 from sentry.users.services.user.service import user_service
@@ -97,7 +98,7 @@ def post(self, request: Request, organization) -> Response:
             ),
         }
 
-        if self._has_hook_events(request) and not features.has(
+        if has_error_events(request.data.get("events")) and not features.has(
             "organizations:integrations-event-hooks", organization, actor=request.user
         ):
             return Response(
@@ -182,9 +183,3 @@ def _filter_queryset_for_user(self, queryset: BaseQuerySet[SentryApp, SentryApp]
                 owner_ids.append(o.id)
 
         return queryset.filter(owner_id__in=owner_ids)
-
-    def _has_hook_events(self, request: Request):
-        if not request.data.get("events"):
-            return False
-
-        return "error" in request.data["events"]

src/sentry/sentry_apps/utils/webhooks.py

@@ -136,6 +136,14 @@
     return EVENT_TO_RESOURCE.get(event)
 
 
+def has_error_events(events: Collection[str] | None) -> bool:
+    """Whether any entry subscribes to error webhooks, as the whole resource or a single event."""
+    return any(
+        event == SentryAppResourceType.ERROR or resource_of(event) is SentryAppResourceType.ERROR
+        for event in events or ()
+    )
+
+
 def is_subscribed(stored_events: Collection[str], event: str) -> bool:
     """
     Whether a stored subscription covers a fired event.

tests/sentry/sentry_apps/api/endpoints/test_sentry_app_details.py

@@ -607,6 +607,16 @@ def test_can_add_error_created_hook_with_flag(self) -> None:
             status_code=200,
         )
 
+    @with_feature({"organizations:integrations-event-hooks": False})
+    @override_options({"staff.ga-rollout": True})
+    def test_cannot_add_granular_error_created_without_flag(self) -> None:
+        app = self.create_sentry_app(name="SampleApp", organization=self.organization)
+        self.get_error_response(
+            app.slug,
+            events=["error.created"],
+            status_code=403,
+        )
+
     @override_options({"staff.ga-rollout": True})
     def test_can_add_granular_events(self) -> None:
         app = self.create_sentry_app(name="SampleApp", organization=self.organization)

tests/sentry/sentry_apps/api/endpoints/test_sentry_apps.py

@@ -700,6 +700,17 @@ def test_cannot_create_with_error_created_hook_without_flag(self) -> None:
                 ]
             }
 
+    def test_cannot_create_with_granular_error_created_without_flag(self) -> None:
+        with Feature({"organizations:integrations-event-hooks": False}):
+            response = self.get_error_response(
+                **self.get_data(events=("error.created",)), status_code=403
+            )
+            assert response.data == {
+                "non_field_errors": [
+                    "Your organization does not have access to the 'error' resource subscription."
+                ]
+            }
+
     def test_can_create_with_granular_events(self) -> None:
         response = self.get_success_response(
             **self.get_data(events=("issue.resolved",)), status_code=201