fix(auth): Preserve redirect state across SSO failures and detect provider mismatch

Two issues in the authentication flow are addressed:

1. Redirect state loss: The original `next` URL was cleared by
   `initiate_login()` when SSO failed and users retried. Added a
   persistent `_original_next` session key that survives multiple
   login attempts, with a 1-hour TTL for security.

2. Provider mismatch: When users authenticated with the wrong SSO
   provider (e.g., Google when Okta required), `build_identity()`
   would fail silently. Now we detect the mismatch before calling
   `build_identity()` and redirect users to the correct SSO flow
   with a clear warning message.

Also fixes the 2FA flow setting `after_2fa` to the SSO callback URL
instead of the user's original destination. URLs are validated with
`is_valid_redirect()` to prevent open redirect attacks.

Author: dcramer

src/sentry/auth/helper.py

@@ -133,10 +133,25 @@ def _login(self, user: Any) -> None:
             sample_rate=1.0,
             skip_internal=False,
         )
+
+        # Get the original intended destination for after_2fa, not the current SSO URL
+        after_2fa_url = self.request.session.get("_next")
+        if not after_2fa_url:
+            after_2fa_url = auth.get_persistent_next(self.request)
+
+        # Validate the URL to prevent open redirect attacks
+        if after_2fa_url and not auth.is_valid_redirect(
+            after_2fa_url, allowed_hosts=(self.request.get_host(),)
+        ):
+            after_2fa_url = None
+
+        if not after_2fa_url:
+            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 +795,16 @@ 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
+        actual_provider_key = data.get("actual_provider_key")
+        expected_provider_key = self.provider.key
+        if actual_provider_key and actual_provider_key != expected_provider_key:
+            return self._handle_provider_mismatch(
+                actual_provider_key=actual_provider_key,
+                expected_provider_key=expected_provider_key,
+            )
+
         try:
             identity = self.provider.build_identity(data)
         except IdentityNotValid as error:
@@ -796,6 +821,49 @@ def finish_pipeline(self) -> HttpResponseBase:
 
         return response
 
+    def _handle_provider_mismatch(
+        self,
+        actual_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": actual_provider_key,
+            },
+        )
+
+        metrics.incr(
+            "sso.provider_mismatch",
+            tags={
+                "expected_provider": expected_provider_key,
+                "actual_provider": actual_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

@@ -151,6 +151,9 @@ def dispatch(self, request: HttpRequest, pipeline: AuthHelper) -> HttpResponseBa
         # hook here
         pipeline.bind_state("data", data)
 
+        # Store the actual provider key for mismatch detection
+        pipeline.bind_state("actual_provider_key", pipeline.provider.key)
+
         return pipeline.next_step()
 
 

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

@@ -150,6 +150,9 @@ def dispatch(self, request: HttpRequest, pipeline: AuthHelper) -> HttpResponseBa
 
         pipeline.bind_state("auth_attributes", auth.get_attributes())
 
+        # Store the actual provider key for mismatch detection
+        pipeline.bind_state("actual_provider_key", pipeline.provider.key)
+
         return pipeline.next_step()
 
 

src/sentry/utils/auth.py

@@ -141,12 +141,57 @@ def get_login_url(reset: bool = False) -> str:
     return _LOGIN_URL
 
 
+# Persistent redirect URL storage - survives across multiple login attempts
+PERSISTENT_NEXT_KEY = "_original_next"
+PERSISTENT_NEXT_MAX_AGE = 60 * 60  # 1 hour
+
+
+def set_persistent_next(request: HttpRequest, next_url: str) -> None:
+    """
+    Store the original next URL persistently in the session.
+    Only sets if not already set, preserving the original destination
+    across multiple login attempts and SSO failures.
+    """
+    if not next_url:
+        return
+    if PERSISTENT_NEXT_KEY not in request.session:
+        request.session[PERSISTENT_NEXT_KEY] = {
+            "url": next_url,
+            "timestamp": time(),
+        }
+
+
+def get_persistent_next(request: HttpRequest, max_age: int | None = None) -> str | None:
+    """
+    Retrieve the persistent next URL if it exists and is not stale.
+    Returns None if the URL has expired or doesn't exist.
+    """
+    data = request.session.get(PERSISTENT_NEXT_KEY)
+    if not data:
+        return None
+
+    if max_age is None:
+        max_age = PERSISTENT_NEXT_MAX_AGE
+
+    if time() - data.get("timestamp", 0) > max_age:
+        clear_persistent_next(request)
+        return None
+
+    return data.get("url")
+
+
+def clear_persistent_next(request: HttpRequest) -> None:
+    """Clear the persistent next URL from session."""
+    request.session.pop(PERSISTENT_NEXT_KEY, None)
+
+
 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
+    initiate_login clears transient session state for authentication.
+    If provided a `next_url`, stores it both as the immediate redirect target
+    and as a persistent fallback that survives across multiple login attempts.
     """
     for key in ("_next", "_after_2fa", "_pending_2fa", "_referrer"):
         try:
@@ -156,6 +201,8 @@ def initiate_login(
 
     if next_url:
         request.session["_next"] = next_url
+        # Also store as persistent fallback (only sets if not already present)
+        set_persistent_next(request, next_url)
     if referrer:
         request.session["_referrer"] = referrer
 
@@ -186,7 +233,13 @@ def _get_login_redirect(request: HttpRequest, default: str | None = None) -> str
     if after_2fa is not None:
         return after_2fa
 
+    # Try the transient _next first
     login_url = request.session.pop("_next", None)
+
+    # Fall back to persistent next URL if transient is not set
+    if not login_url:
+        login_url = get_persistent_next(request)
+
     if not login_url:
         return default
 
@@ -361,6 +414,9 @@ def login(
     with outbox_context(flush=False):
         _login(request, user)
 
+    # Clear the persistent next URL now that login is complete
+    clear_persistent_next(request)
+
     log_auth_success(request, user.username, organization_id, source)
     return True
 

tests/sentry/auth/test_helper.py

@@ -590,3 +590,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, actual_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": {"actual_provider_key": actual_provider_key} if actual_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(actual_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(actual_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
+            response = 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_actual_provider_key_continues_normally(self, mock_messages: mock.MagicMock) -> None:
+        """Test that missing actual_provider_key doesn't trigger mismatch (backward compat)."""
+        helper = self._create_helper_with_state(actual_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"},
+        ):
+            response = 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)