feat(integrations): Add API driven pipeline for Slack

Implement `get_pipeline_api_steps()` on `SlackIntegrationProvider` using a
single OAuth2 step with Slack's `user_scope` parameter. Refactors
`OAuth2Provider.get_pipeline_api_steps()` to expose `make_oauth_api_step()`
so providers can customize step parameters (like extra_authorize_params).
Updates `build_integration()` to handle both legacy `state["identity"]["data"]`
and new `state["oauth_data"]` paths.

Ref [VDY-42](https://linear.app/getsentry/issue/VDY-42/slack-api-driven-integration-setup)

Author: evanpurkhiser

src/sentry/identity/oauth2.py

@@ -147,22 +147,24 @@ def get_pipeline_views(self) -> list[PipelineView[IdentityPipeline]]:
             ),
         ]
 
-    def get_pipeline_api_steps(self) -> list[OAuth2ApiStep]:
+    def make_oauth_api_step(self, **kwargs: Any) -> OAuth2ApiStep:
         redirect_url = self.config.get(
             "redirect_url",
             reverse("sentry-extension-setup", kwargs={"provider_id": "default"}),
         )
-        return [
-            OAuth2ApiStep(
-                authorize_url=self.get_oauth_authorize_url(),
-                client_id=self.get_oauth_client_id(),
-                client_secret=self.get_oauth_client_secret(),
-                access_token_url=self.get_oauth_access_token_url(),
-                scope=" ".join(self.get_oauth_scopes()),
-                redirect_url=redirect_url,
-                verify_ssl=self.config.get("verify_ssl", True),
-            ),
-        ]
+        return OAuth2ApiStep(
+            authorize_url=self.get_oauth_authorize_url(),
+            client_id=self.get_oauth_client_id(),
+            client_secret=self.get_oauth_client_secret(),
+            access_token_url=self.get_oauth_access_token_url(),
+            scope=" ".join(self.get_oauth_scopes()),
+            redirect_url=redirect_url,
+            verify_ssl=self.config.get("verify_ssl", True),
+            **kwargs,
+        )
+
+    def get_pipeline_api_steps(self) -> list[OAuth2ApiStep]:
+        return [self.make_oauth_api_step()]
 
     def get_refresh_token_params(
         self, refresh_token: str, identity: Identity | RpcIdentity, **kwargs: Any

src/sentry/integrations/slack/integration.py

@@ -10,6 +10,7 @@
 from slack_sdk.errors import SlackApiError
 
 from sentry.identity.pipeline import IdentityPipeline
+from sentry.identity.slack.provider import SlackIdentityProvider
 from sentry.integrations.base import (
     FeatureDescription,
     IntegrationData,
@@ -36,7 +37,7 @@
 )
 from sentry.notifications.platform.target import IntegrationNotificationTarget
 from sentry.organizations.services.organization.model import RpcOrganization
-from sentry.pipeline.views.base import PipelineView
+from sentry.pipeline.views.base import ApiPipelineSteps, PipelineView
 from sentry.pipeline.views.nested import NestedPipelineView
 from sentry.shared_integrations.exceptions import IntegrationError
 from sentry.utils.http import absolute_uri
@@ -322,6 +323,7 @@ def _get_oauth_scopes(self) -> frozenset[str]:
         return self.identity_oauth_scopes
 
     setup_dialog_config = {"width": 600, "height": 900}
+    setup_url_path = "/extensions/slack/setup/"
 
     def _identity_pipeline_view(self) -> PipelineView[IntegrationPipeline]:
         return NestedPipelineView(
@@ -331,13 +333,28 @@ def _identity_pipeline_view(self) -> PipelineView[IntegrationPipeline]:
             config={
                 "oauth_scopes": self._get_oauth_scopes(),
                 "user_scopes": self.user_scopes,
-                "redirect_url": absolute_uri("/extensions/slack/setup/"),
+                "redirect_url": absolute_uri(self.setup_url_path),
             },
         )
 
     def get_pipeline_views(self) -> Sequence[PipelineView[IntegrationPipeline]]:
         return [self._identity_pipeline_view()]
 
+    def _make_identity_provider(self) -> SlackIdentityProvider:
+        return SlackIdentityProvider(
+            oauth_scopes=self._get_oauth_scopes(),
+            redirect_url=absolute_uri(self.setup_url_path),
+        )
+
+    def get_pipeline_api_steps(self) -> ApiPipelineSteps[IntegrationPipeline]:
+        provider = self._make_identity_provider()
+        return [
+            provider.make_oauth_api_step(
+                bind_key="oauth_data",
+                extra_authorize_params={"user_scope": " ".join(self.user_scopes)},
+            ),
+        ]
+
     def _get_team_info(self, access_token: str) -> Any:
         # Manually add authorization since this method is part of slack installation
 
@@ -352,7 +369,13 @@ def _get_team_info(self, access_token: str) -> Any:
             raise IntegrationError("Could not retrieve Slack team information.")
 
     def build_integration(self, state: Mapping[str, Any]) -> IntegrationData:
-        data = state["identity"]["data"]
+        # TODO: legacy views write token data to state["identity"]["data"] via
+        # NestedPipelineView. API steps write directly to state["oauth_data"].
+        # Remove the legacy path once the old views are retired.
+        if "oauth_data" in state:
+            data = state["oauth_data"]
+        else:
+            data = state["identity"]["data"]
         assert data["ok"]
 
         access_token = data["access_token"]

src/sentry/integrations/slack/staging/integration.py

@@ -5,6 +5,7 @@
 from typing import Any
 
 from sentry.identity.pipeline import IdentityPipeline
+from sentry.identity.slack.provider import SlackStagingIdentityProvider
 from sentry.integrations.base import (
     IntegrationData,
 )
@@ -22,10 +23,17 @@ class SlackStagingIntegrationProvider(SlackIntegrationProvider):
     key = IntegrationProviderSlug.SLACK_STAGING.value
     name = "Slack (Staging)"
     requires_feature_flag = True
+    setup_url_path = "/extensions/slack-staging/setup/"
 
     def _get_oauth_scopes(self) -> frozenset[str]:
         return self.identity_oauth_scopes | self.extended_oauth_scopes
 
+    def _make_identity_provider(self) -> SlackStagingIdentityProvider:
+        return SlackStagingIdentityProvider(
+            oauth_scopes=self._get_oauth_scopes(),
+            redirect_url=absolute_uri(self.setup_url_path),
+        )
+
     def _identity_pipeline_view(self) -> PipelineView[IntegrationPipeline]:
         return NestedPipelineView(
             bind_key="identity",
@@ -34,7 +42,7 @@ def _identity_pipeline_view(self) -> PipelineView[IntegrationPipeline]:
             config={
                 "oauth_scopes": self._get_oauth_scopes(),
                 "user_scopes": self.user_scopes,
-                "redirect_url": absolute_uri("/extensions/slack-staging/setup/"),
+                "redirect_url": absolute_uri(self.setup_url_path),
             },
         )
 

tests/sentry/integrations/slack/test_integration.py

@@ -1,16 +1,21 @@
+from __future__ import annotations
+
+from typing import Any
 from unittest.mock import MagicMock, patch
 from urllib.parse import parse_qs, urlencode, urlparse
 
 import orjson
 import pytest
 import responses
+from django.urls import reverse
 from responses.matchers import query_string_matcher
 from slack_sdk.errors import SlackApiError
 from slack_sdk.web import SlackResponse
 
 from sentry import audit_log
 from sentry.integrations.models.integration import Integration
 from sentry.integrations.models.organization_integration import OrganizationIntegration
+from sentry.integrations.pipeline import IntegrationPipeline
 from sentry.integrations.slack import SlackIntegration, SlackIntegrationProvider
 from sentry.integrations.slack.utils.constants import SlackScope
 from sentry.integrations.slack.utils.users import SLACK_GET_USERS_PAGE_SIZE
@@ -653,3 +658,114 @@ def test_get_thread_history_error_returns_empty_list(
             channel_id=self.channel_id, thread_ts=self.thread_ts
         )
         assert result == []
+
+
+@control_silo_test
+class SlackApiPipelineTest(APITestCase):
+    endpoint = "sentry-api-0-organization-pipeline"
+    method = "post"
+
+    def setUp(self) -> None:
+        super().setUp()
+        self.login_as(self.user)
+
+    def tearDown(self) -> None:
+        responses.reset()
+        super().tearDown()
+
+    def _get_pipeline_url(self) -> str:
+        return reverse(
+            self.endpoint,
+            args=[self.organization.slug, IntegrationPipeline.pipeline_name],
+        )
+
+    def _initialize_pipeline(self) -> Any:
+        return self.client.post(
+            self._get_pipeline_url(),
+            data={"action": "initialize", "provider": "slack"},
+            format="json",
+        )
+
+    def _advance_step(self, data: dict[str, Any]) -> Any:
+        return self.client.post(self._get_pipeline_url(), data=data, format="json")
+
+    def _get_pipeline_signature(self, resp: Any) -> str:
+        return resp.data["data"]["oauthUrl"].split("state=")[1].split("&")[0]
+
+    @responses.activate
+    def test_initialize_pipeline(self) -> None:
+        resp = self._initialize_pipeline()
+        assert resp.status_code == 200
+        assert resp.data["step"] == "oauth_login"
+        assert resp.data["stepIndex"] == 0
+        assert resp.data["totalSteps"] == 1
+        assert resp.data["provider"] == "slack"
+        oauth_url = resp.data["data"]["oauthUrl"]
+        assert "slack.com/oauth/v2/authorize" in oauth_url
+        assert "user_scope=" in oauth_url
+
+        # Verify the OAuth URL requests the correct bot scopes, not the
+        # identity provider's default scopes
+        parsed = urlparse(oauth_url)
+        params = parse_qs(parsed.query)
+        requested_scopes = set(params["scope"][0].split(" "))
+        assert requested_scopes == SlackIntegrationProvider.identity_oauth_scopes
+
+    @responses.activate
+    def test_oauth_step_missing_code(self) -> None:
+        self._initialize_pipeline()
+        resp = self._advance_step({})
+        assert resp.status_code == 400
+
+    @responses.activate
+    @patch("sentry.integrations.slack.integration.WebClient")
+    def test_full_pipeline_flow(self, mock_web_client_cls: MagicMock) -> None:
+        mock_client = MagicMock()
+        mock_web_client_cls.return_value = mock_client
+        mock_client.team_info.return_value = SlackResponse(
+            client=mock_client,
+            http_verb="GET",
+            api_url="https://slack.com/api/team.info",
+            req_args={},
+            data={
+                "ok": True,
+                "team": {
+                    "name": "Test Team",
+                    "id": "T1234",
+                    "domain": "test-team",
+                    "icon": {"image_132": "https://example.com/icon.png"},
+                },
+            },
+            headers={},
+            status_code=200,
+        )
+
+        responses.add(
+            responses.POST,
+            "https://slack.com/api/oauth.v2.access",
+            json={
+                "ok": True,
+                "access_token": "xoxb-test-token",
+                "scope": "channels:read,chat:write",
+                "team": {"name": "Test Team", "id": "T1234"},
+                "authed_user": {"id": "U1234"},
+            },
+        )
+
+        resp = self._initialize_pipeline()
+        assert resp.data["step"] == "oauth_login"
+        pipeline_signature = self._get_pipeline_signature(resp)
+
+        resp = self._advance_step({"code": "slack-auth-code", "state": pipeline_signature})
+        assert resp.status_code == 200
+        assert resp.data["status"] == "complete"
+        assert "data" in resp.data
+
+        integration = Integration.objects.get(provider="slack")
+        assert integration.external_id == "T1234"
+        assert integration.name == "Test Team"
+
+        assert OrganizationIntegration.objects.filter(
+            organization_id=self.organization.id,
+            integration=integration,
+        ).exists()