fix(auth): Detect SSO provider mismatch and fix 2FA redirect (#106041)

Fixes two issues in the SSO authentication flow:

**Provider mismatch detection**: When users authenticated with the wrong
SSO provider (e.g., Google when the org requires Okta),
`build_identity()` would fail with a confusing error. Now we detect the
mismatch before calling `build_identity()` by comparing the callback's
provider against the org's configured provider, and redirect users to
the correct SSO flow with a clear warning message.

**2FA redirect fix**: The 2FA flow was setting `after_2fa` to the SSO
callback URL instead of the user's original destination. Now it uses
`_next` from the session (validated with `is_valid_redirect()` to
prevent open redirects).

## Changes

- `src/sentry/auth/helper.py`: Add provider mismatch detection in
`finish_pipeline()`, fix `after_2fa` URL in `_login()`
- `src/sentry/auth/providers/oauth2.py`, `saml2/provider.py`: Store
`provider_key` in pipeline state for mismatch detection

Author: dcramer

src/sentry/auth/helper.py

@@ -133,10 +133,19 @@ def _login(self, user: Any) -> None:
             sample_rate=1.0,
             skip_internal=False,
         )
+
+        # Use the user's original destination (from _next) for 2FA redirect,
+        # falling back to current URL if not set or invalid
+        after_2fa_url = self.request.session.get("_next")
+        if not after_2fa_url or not auth.is_valid_redirect(
+            after_2fa_url, allowed_hosts=(self.request.get_host(),)
+        ):
+            after_2fa_url = self.request.build_absolute_uri()
+
         user_was_logged_in = auth.login(
             self.request,
             user,
-            after_2fa=self.request.build_absolute_uri(),
+            after_2fa=after_2fa_url,
             organization_id=self.organization.id,
         )
         if not user_was_logged_in:
@@ -780,6 +789,21 @@ def finish_pipeline(self) -> HttpResponseBase:
         if not data:
             return self.error(ERR_INVALID_IDENTITY)
 
+        # Check for provider mismatch - user authenticated with a different provider
+        # than what the organization requires. This can happen when a user has multiple
+        # SSO sessions in different tabs and completes the wrong one.
+        provider_key = data.get("provider_key")
+        if (
+            self.state.flow == FLOW_LOGIN
+            and self.provider_model
+            and provider_key
+            and provider_key != self.provider_model.provider
+        ):
+            return self._handle_provider_mismatch(
+                provider_key=provider_key,
+                expected_provider_key=self.provider_model.provider,
+            )
+
         try:
             identity = self.provider.build_identity(data)
         except IdentityNotValid as error:
@@ -796,6 +820,49 @@ def finish_pipeline(self) -> HttpResponseBase:
 
         return response
 
+    def _handle_provider_mismatch(
+        self,
+        provider_key: str,
+        expected_provider_key: str,
+    ) -> HttpResponseRedirect:
+        """
+        Handle when user authenticated with a different provider than required.
+
+        This happens when a user starts SSO for an org (e.g., "sentry" which requires Okta)
+        but authenticates with a different provider (e.g., Google). We redirect them back
+        to the org's login page to use the correct SSO provider.
+        """
+        logger.info(
+            "sso.provider-mismatch",
+            extra={
+                "organization_id": self.organization.id,
+                "expected_provider": expected_provider_key,
+                "actual_provider": provider_key,
+            },
+        )
+
+        metrics.incr(
+            "sso.provider_mismatch",
+            tags={
+                "expected_provider": expected_provider_key,
+                "actual_provider": provider_key,
+            },
+        )
+
+        # Clear the invalid pipeline state
+        self.clear_session()
+
+        expected_name = getattr(self.provider, "name", expected_provider_key)
+        messages.add_message(
+            self.request,
+            messages.WARNING,
+            f"This organization requires {expected_name} SSO. Please sign in with your organization's SSO provider.",
+        )
+
+        # Redirect to org-specific login to initiate correct SSO
+        redirect_uri = reverse("sentry-auth-organization", args=[self.organization.slug])
+        return HttpResponseRedirect(redirect_uri)
+
     def auth_handler(self, identity: Mapping[str, Any]) -> AuthIdentityHandler:
         assert self.provider_model is not None
         return AuthIdentityHandler(

src/sentry/auth/providers/oauth2.py

@@ -70,7 +70,11 @@ def dispatch(self, request: HttpRequest, pipeline: AuthHelper) -> HttpResponseBa
         if "code" in request.GET:
             return pipeline.next_step()
 
-        state = secrets.token_hex()
+        # Encode provider key in the state parameter so it survives the OAuth redirect.
+        # This allows detecting when a user completes an OAuth flow that was started
+        # for a different provider (e.g., multiple SSO tabs open).
+        nonce = secrets.token_hex()
+        state = f"{nonce}:{pipeline.provider.key}"
 
         params = self.get_authorize_params(state=state, redirect_uri=_get_redirect_url())
         authorization_url = f"{self.get_authorize_url()}?{urlencode(params)}"
@@ -151,6 +155,15 @@ def dispatch(self, request: HttpRequest, pipeline: AuthHelper) -> HttpResponseBa
         # hook here
         pipeline.bind_state("data", data)
 
+        # Extract the provider key from the OAuth state parameter.
+        # This was encoded when the OAuth flow started (in OAuth2Login) and survives
+        # the redirect through the IdP, allowing us to detect if the user completed
+        # an OAuth flow that was started for a different provider.
+        provider_key = None
+        if state and ":" in state:
+            provider_key = state.split(":", 1)[1]
+        pipeline.bind_state("provider_key", provider_key)
+
         return pipeline.next_step()
 
 

src/sentry/auth/providers/saml2/provider.py

@@ -78,7 +78,13 @@ def dispatch(self, request: HttpRequest, pipeline: AuthHelper) -> HttpResponseBa
         saml_config = build_saml_config(provider.config, pipeline.organization.slug)
         auth = build_auth(request, saml_config)
 
-        return HttpResponseRedirect(auth.login())
+        # Encode provider key in RelayState so it survives the SAML redirect.
+        # This allows detecting when a user completes a SAML flow that was started
+        # for a different provider (e.g., multiple SSO tabs open).
+        # Format: "provider_key:{key}" or just return_to URL for backward compat
+        relay_state = f"provider_key:{pipeline.provider.key}"
+
+        return HttpResponseRedirect(auth.login(return_to=relay_state))
 
 
 # With SAML, the SSO request can be initiated by both the Service Provider
@@ -150,6 +156,16 @@ def dispatch(self, request: HttpRequest, pipeline: AuthHelper) -> HttpResponseBa
 
         pipeline.bind_state("auth_attributes", auth.get_attributes())
 
+        # Extract the provider key from the RelayState parameter.
+        # This was encoded when the SAML flow started (in SAML2LoginView) and survives
+        # the redirect through the IdP, allowing us to detect if the user completed
+        # a SAML flow that was started for a different provider.
+        provider_key = None
+        relay_state = request.POST.get("RelayState") or request.GET.get("RelayState")
+        if relay_state and relay_state.startswith("provider_key:"):
+            provider_key = relay_state.split(":", 1)[1]
+        pipeline.bind_state("provider_key", provider_key)
+
         return pipeline.next_step()
 
 

src/sentry/utils/auth.py

@@ -145,8 +145,8 @@ def initiate_login(
     request: HttpRequest, next_url: str | None = None, referrer: str | None = None
 ) -> None:
     """
-    initiate_login simply clears session cache
-    if provided a `next_url` will append to the session after clearing previous keys
+    Clears existing login state and initializes a new login flow.
+    Optionally sets the post-login redirect destination and referrer.
     """
     for key in ("_next", "_after_2fa", "_pending_2fa", "_referrer"):
         try:

tests/sentry/auth/test_helper.py

@@ -632,3 +632,106 @@ def test_has_verified_account_fail_user_id(self) -> None:
         wrong_user = self.create_user()
         self.create_useremail(email=self.email, user=wrong_user)
         assert self.handler.has_verified_account(self.verification_value) is False
+
+
+@control_silo_test
+class ProviderMismatchTest(TestCase):
+    """Tests for provider mismatch detection when user auths with wrong SSO provider."""
+
+    def setUp(self) -> None:
+        self.provider = "dummy"
+        self.auth_provider_inst = AuthProvider.objects.create(
+            organization_id=self.organization.id, provider=self.provider
+        )
+
+        self.auth_key = "test_auth_key"
+        self.request = _set_up_request()
+        self.request.session["auth_key"] = self.auth_key
+
+    def _create_helper_with_state(self, provider_key=None):
+        """Create an AuthHelper with initial state and optional provider key mismatch."""
+        initial_state = {
+            "org_id": self.organization.id,
+            "flow": FLOW_LOGIN,
+            "provider_model_id": self.auth_provider_inst.id,
+            "provider_key": self.provider,
+            "referrer": None,
+            "step_index": 1,
+            "signature": None,
+            "config": {},
+            "data": {"provider_key": provider_key} if provider_key else {},
+        }
+        local_client = clusters.get("default").get_local_client_for_key(self.auth_key)
+        local_client.set(self.auth_key, json.dumps(initial_state))
+
+        helper = AuthHelper.get_for_request(self.request)
+        assert helper is not None
+        return helper
+
+    @mock.patch("sentry.auth.helper.messages")
+    @mock.patch("sentry.auth.helper.metrics")
+    def test_provider_mismatch_redirects_to_correct_sso(
+        self, mock_metrics: mock.MagicMock, mock_messages: mock.MagicMock
+    ) -> None:
+        """Test that authenticating with wrong provider redirects to correct SSO."""
+        helper = self._create_helper_with_state(provider_key="google")
+
+        # Mock the provider to have a build_identity that would fail
+        with mock.patch.object(helper.provider, "build_identity") as mock_build:
+            mock_build.side_effect = Exception("Should not be called")
+
+            response = helper.finish_pipeline()
+
+        # Should redirect to org SSO page
+        assert response.status_code == 302
+        assert f"/auth/login/{self.organization.slug}/" in response.url
+
+        # Should show warning message
+        mock_messages.add_message.assert_called_once()
+        call_args = mock_messages.add_message.call_args
+        assert call_args[0][1] == mock_messages.WARNING
+
+        # Should log metric
+        mock_metrics.incr.assert_called_with(
+            "sso.provider_mismatch",
+            tags={
+                "expected_provider": self.provider,
+                "actual_provider": "google",
+            },
+        )
+
+    @mock.patch("sentry.auth.helper.messages")
+    def test_provider_match_continues_normally(self, mock_messages: mock.MagicMock) -> None:
+        """Test that matching provider continues with normal flow."""
+        helper = self._create_helper_with_state(provider_key=self.provider)
+
+        # Mock build_identity to return a valid identity
+        with mock.patch.object(
+            helper.provider,
+            "build_identity",
+            return_value={"id": "123", "email": "[email protected]", "name": "Test"},
+        ):
+            # The flow will continue and eventually redirect
+            helper.finish_pipeline()
+
+        # Should not have shown provider mismatch warning
+        for call in mock_messages.add_message.call_args_list:
+            assert "SSO" not in str(call)
+
+    @mock.patch("sentry.auth.helper.messages")
+    def test_no_provider_key_continues_normally(self, mock_messages: mock.MagicMock) -> None:
+        """Test that missing provider_key doesn't trigger mismatch (backward compat)."""
+        helper = self._create_helper_with_state(provider_key=None)
+
+        # Mock build_identity to return a valid identity
+        with mock.patch.object(
+            helper.provider,
+            "build_identity",
+            return_value={"id": "123", "email": "[email protected]", "name": "Test"},
+        ):
+            helper.finish_pipeline()
+
+        # Should not have shown provider mismatch warning
+        for call in mock_messages.add_message.call_args_list:
+            if call[0][1] == mock_messages.WARNING:
+                assert "SSO" not in str(call)

tests/sentry/web/frontend/test_auth_oauth2.py

@@ -229,3 +229,37 @@ def test_response_errors(self) -> None:
         assert len(messages) == 1
         assert str(messages[0]).startswith("Authentication error")
         assert auth_resp.context["user"] != self.user
+
+    def test_state_contains_provider_key(self) -> None:
+        """Test that OAuth state parameter contains the provider key for mismatch detection."""
+        state = self.initiate_oauth_flow()
+
+        # State format should be "{nonce}:{provider_key}"
+        assert ":" in state
+        nonce, provider_key = state.split(":", 1)
+        assert len(nonce) > 0
+        assert provider_key == self.provider_name
+
+    @mock.patch("sentry.auth.providers.oauth2.safe_urlopen")
+    def test_provider_mismatch_detected(self, urlopen: mock.MagicMock) -> None:
+        """Test that authenticating with wrong provider in state triggers mismatch detection."""
+        # Start a normal OAuth flow to get the session set up
+        state = self.initiate_oauth_flow()
+
+        # Modify the state to have a different provider key (simulating multi-tab scenario
+        # where user started auth with a different provider)
+        nonce = state.split(":")[0]
+        wrong_state = f"{nonce}:wrong_provider"
+
+        # The state validation should fail because the full state doesn't match
+        headers = {"Content-Type": "application/json"}
+        auth_data = {"id": "oauth_external_id_1234", "email": self.user.email}
+        urlopen.return_value = MockResponse(headers, json.dumps(auth_data))
+
+        query = urlencode({"code": "1234", "state": wrong_state})
+        auth_resp = self.client.get(f"{self.sso_path}?{query}", follow=True)
+
+        # Should fail with state mismatch error
+        messages = list(auth_resp.context["messages"])
+        assert len(messages) == 1
+        assert str(messages[0]).startswith("Authentication error")

tests/sentry/web/frontend/test_auth_saml2.py

@@ -308,3 +308,26 @@ def test_verify_email(self, follow=False, **kwargs) -> None:
 
         # expect no linking before verification
         assert AuthIdentity.objects.filter(user_id=self.user.id).count() == 0
+
+    def test_relay_state_contains_provider_key(self) -> None:
+        """Test that SAML RelayState contains the provider key for mismatch detection."""
+        resp = self.client.post(self.login_path, {"init": True})
+
+        assert resp.status_code == 302
+        redirect = urlparse(resp.get("Location", ""))
+        query = parse_qs(redirect.query)
+
+        # RelayState should contain the provider key
+        assert "RelayState" in query
+        relay_state = query["RelayState"][0]
+        assert relay_state == f"provider_key:{self.provider_name}"
+
+    def test_idp_initiated_without_relay_state_continues(self) -> None:
+        """Test that IdP-initiated SAML without RelayState continues normally (backward compat)."""
+        # IdP-initiated auth doesn't have RelayState from our side
+        # This should still work - the provider_key will be None and the check will be skipped
+        auth = self.accept_auth()
+
+        # Should continue to identity confirmation
+        assert auth.status_code == 200
+        assert auth.context["existing_user"] == self.user