fix: prevent silent errors during SSO (account settings)

common/djangoapps/third_party_auth/tests/test_middleware.py

@@ -5,12 +5,15 @@
 
 from unittest import mock
 
+import ddt
 from django.contrib.messages.middleware import MessageMiddleware
 from django.http import HttpResponse
 from django.test.client import RequestFactory
 from requests.exceptions import HTTPError
+from social_core import exceptions as social_exceptions
 
 from common.djangoapps.student.helpers import get_next_url_for_login_page
+from common.djangoapps.third_party_auth import pipeline
 from common.djangoapps.third_party_auth.middleware import ExceptionMiddleware
 from common.djangoapps.third_party_auth.tests.testutil import TestCase
 from openedx.core.djangolib.testing.utils import skip_unless_lms
@@ -43,3 +46,106 @@ def test_http_exception_redirection(self):
 
         assert response.status_code == 302
         assert target_url.endswith(login_url)
+
+
+@ddt.ddt
+class ExceptionMiddlewareAccountSettingsDispatchTestCase(TestCase):
+    """
+    Tests that ExceptionMiddleware.get_redirect_uri() dispatches to the URL
+    registered in AUTH_DISPATCH_URLS for the current auth_entry, and that it
+    no longer needs to duplicate the error message in a custom session key:
+    SocialAuthExceptionMiddleware.process_exception (the parent class)
+    already leaves it in the Django messages framework for any
+    SocialAuthBaseException, tagged 'social-auth <backend name>'. The
+    Account MFE reads that message via
+    openedx.core.djangoapps.user_api.views.ThirdPartyAuthErrorMessageView.
+    """
+
+    def _build_request(self, auth_entry=pipeline.AUTH_ENTRY_ACCOUNT_SETTINGS):
+        """Build a fake request with session and backend for testing TPA error handling."""
+        request = RequestFactory().get('/auth/login/tpa-saml/')
+        request.session = {}
+        request.session[pipeline.AUTH_ENTRY_KEY] = auth_entry
+
+        class FakeBackend:
+            name = 'tpa-saml'
+
+        request.backend = FakeBackend()
+        request.social_strategy = mock.MagicMock()
+        request.social_strategy.setting.return_value = None
+        return request
+
+    @ddt.data(
+        social_exceptions.AuthAlreadyAssociated,
+        social_exceptions.AuthCanceled,
+        social_exceptions.AuthFailed,
+        social_exceptions.AuthTokenError,
+        social_exceptions.AuthStateMissing,
+        social_exceptions.AuthStateForbidden,
+        social_exceptions.AuthTokenRevoked,
+        social_exceptions.AuthUnreachableProvider,
+        social_exceptions.InvalidEmail,
+    )
+    def test_dispatches_to_account_settings_url_for_any_tpa_exception(self, exception_class):
+        """The redirect target is /account/settings for the account_settings flow, regardless of exception type."""
+        request = self._build_request()
+
+        redirect_uri = ExceptionMiddleware(get_response=lambda r: None).get_redirect_uri(
+            request, exception_class('tpa-saml')
+        )
+
+        assert redirect_uri == pipeline.AUTH_DISPATCH_URLS[pipeline.AUTH_ENTRY_ACCOUNT_SETTINGS]
+
+    def test_dispatches_elsewhere_outside_account_settings_entry(self):
+        """AUTH_DISPATCH_URLS is keyed by auth_entry, not by exception type."""
+        request = self._build_request(auth_entry=pipeline.AUTH_ENTRY_LOGIN)
+
+        redirect_uri = ExceptionMiddleware(get_response=lambda r: None).get_redirect_uri(
+            request, social_exceptions.AuthAlreadyAssociated('tpa-saml')
+        )
+
+        assert redirect_uri == pipeline.AUTH_DISPATCH_URLS[pipeline.AUTH_ENTRY_LOGIN]
+
+    @skip_unless_lms
+    @ddt.data(
+        social_exceptions.AuthAlreadyAssociated,
+        social_exceptions.AuthCanceled,
+        social_exceptions.AuthFailed,
+        social_exceptions.AuthTokenError,
+        social_exceptions.AuthStateMissing,
+        social_exceptions.AuthStateForbidden,
+        social_exceptions.AuthTokenRevoked,
+        social_exceptions.AuthUnreachableProvider,
+    )
+    def test_process_exception_leaves_social_auth_tagged_message_for_mfe(self, exception_class):
+        """
+        End to end: process_exception (the parent implementation) must still
+        queue a Django message tagged with 'social-auth' for the Account MFE
+        to read, since we no longer save it ourselves. Covers every
+        recognized third-party-auth exception, not just AuthAlreadyAssociated.
+        """
+        request = self._build_request()
+        MessageMiddleware(get_response=lambda request: None).process_request(request)
+        exception = exception_class('tpa-saml')
+
+        ExceptionMiddleware(get_response=lambda r: None).process_exception(request, exception)
+
+        queued_messages = list(request._messages)  # pylint: disable=protected-access
+        assert len(queued_messages) == 1
+        assert queued_messages[0].extra_tags.split() == ['social-auth', 'tpa-saml']
+        assert str(queued_messages[0]) == str(exception)
+
+    @skip_unless_lms
+    def test_process_exception_does_not_queue_message_for_unrecognized_exception(self):
+        """
+        Non-SocialAuthBaseException errors are untouched by
+        SocialAuthExceptionMiddleware.process_exception -- confirms we're not
+        accidentally tagging unrelated exceptions as TPA errors.
+        """
+        request = self._build_request()
+        MessageMiddleware(get_response=lambda request: None).process_request(request)
+
+        result = ExceptionMiddleware(get_response=lambda r: None).process_exception(request, ValueError('boom'))
+
+        assert result is None
+        assert not list(request._messages)  # pylint: disable=protected-access

openedx/core/djangoapps/user_api/legacy_urls.py

@@ -2,8 +2,8 @@
 Defines the URL routes for this app.
 """
 from django.conf import settings
+from django.shortcuts import redirect
 from django.urls import include, path, re_path
-from django.views.generic import RedirectView
 from rest_framework import routers
 
 from openedx.core.djangoapps.site_configuration import helpers as configuration_helpers
@@ -15,16 +15,30 @@
 USER_API_ROUTER.register(r'users', user_api_views.UserViewSet)
 USER_API_ROUTER.register(r'user_prefs', user_api_views.UserPreferenceViewSet)
 
+def account_settings_redirect_view(request):
+    """
+    Backward-compatible redirect for /account and /account/settings to the
+    Account MFE.
+
+    Unlike a plain RedirectView, this re-evaluates ACCOUNT_MICROFRONTEND_URL
+    per request so site-aware configuration is honored.
+
+    Any third-party-auth error message from the pipeline is *not* forwarded
+    here as a query param: SocialAuthExceptionMiddleware already leaves it in
+    the Django messages framework, and the Account MFE reads it directly via
+    ThirdPartyAuthErrorMessageView (openedx/core/djangoapps/user_api/views.py).
+    """
+    account_mfe_url = configuration_helpers.get_value(
+        'ACCOUNT_MICROFRONTEND_URL',
+        settings.ACCOUNT_MICROFRONTEND_URL,
+    )
+    return redirect(account_mfe_url)
+
 urlpatterns = [
     # This redirect is needed for backward compatibility with the old URL structure for the authentication
     # workflows using third-party authentication providers until the authentication workflows fully support
     # the URL structure with MFEs.
-    re_path(r'^account(?:/settings)?/?$', RedirectView.as_view(
-        url=configuration_helpers.get_value(
-            'ACCOUNT_MICROFRONTEND_URL',
-            settings.ACCOUNT_MICROFRONTEND_URL,
-        )),
-    ),
+    re_path(r'^account(?:/settings)?/?$', account_settings_redirect_view),
     path('user_api/v1/', include(USER_API_ROUTER.urls)),
     re_path(
         fr'^user_api/v1/preferences/(?P<pref_key>{UserPreference.KEY_REGEX})/users/$',

openedx/core/djangoapps/user_api/tests/test_legacy_urls.py

@@ -0,0 +1,54 @@
+"""
+Tests for account_settings_redirect_view in legacy_urls.py.
+"""
+from unittest import mock
+
+from django.test import RequestFactory, TestCase, override_settings
+from django.urls import resolve
+
+from openedx.core.djangoapps.user_api.legacy_urls import account_settings_redirect_view
+from openedx.core.djangolib.testing.utils import skip_unless_lms
+
+
+@override_settings(ACCOUNT_MICROFRONTEND_URL='https://account.example.com')
+class AccountSettingsRedirectViewTests(TestCase):
+    """
+    Tests for the view that replaces the legacy /account/settings
+    RedirectView.
+
+    Any third-party-auth error message is intentionally *not* forwarded here:
+    it's read by the Account MFE from ThirdPartyAuthErrorMessageView instead
+    (see openedx/core/djangoapps/user_api/views.py), so this view is a plain,
+    site-aware redirect.
+    """
+
+    def test_redirects_to_account_mfe(self):
+        """Redirect to the configured Account MFE URL, with no query params."""
+        request = RequestFactory().get('/account/settings')
+
+        response = account_settings_redirect_view(request)
+
+        assert response.status_code == 302
+        assert response.url == 'https://account.example.com'
+
+    @override_settings(ACCOUNT_MICROFRONTEND_URL='https://fallback.example.com')
+    def test_uses_site_configuration_value_over_django_settings(self):
+        """The Account MFE URL is re-evaluated per request, honoring site configuration."""
+        request = RequestFactory().get('/account/settings')
+
+        with_site_override = 'openedx.core.djangoapps.site_configuration.helpers.get_value'
+        with mock.patch(with_site_override, return_value='https://site-specific.example.com'):
+            response = account_settings_redirect_view(request)
+
+        assert response.url == 'https://site-specific.example.com'
+
+    @skip_unless_lms
+    def test_account_and_account_settings_urls_route_here(self):
+        """
+        The legacy /account and /account/settings paths (with or without a
+        trailing slash) reach this view. LMS-only: legacy_urls.py is wired
+        into lms/urls.py, not cms/urls.py -- CMS has no account settings page.
+        """
+        for path in ('/account', '/account/', '/account/settings', '/account/settings/'):
+            match = resolve(path)
+            assert match.func == account_settings_redirect_view  # pylint: disable=comparison-with-callable

openedx/core/djangoapps/user_api/tests/test_views.py

@@ -2,6 +2,10 @@
 
 import ddt
 import pytest
+from django.contrib import messages as django_messages
+from django.contrib.messages.storage.session import SessionStorage
+from django.http import HttpResponse
+from django.test import RequestFactory
 from django.test.utils import override_settings
 from django.urls import reverse
 from opaque_keys.edx.keys import CourseKey
@@ -99,6 +103,117 @@ def test_get_list_empty(self):
         assert result['results'] == []
 
 
+@skip_unless_lms
+class ThirdPartyAuthErrorMessageViewTests(ApiTestCase):
+    """
+    Tests for the endpoint the Account MFE polls once on mount to read the
+    pending third-party-auth error message left in the session by
+    common.djangoapps.third_party_auth.middleware.ExceptionMiddleware (via
+    SocialAuthExceptionMiddleware.process_exception).
+    """
+
+    URL = '/api/user/v1/accounts/third_party_auth_error/'
+    TEST_PASSWORD = 'Password1234'
+
+    def setUp(self):
+        super().setUp()
+        self.user = UserFactory.create(password=self.TEST_PASSWORD)
+        self.client.login(username=self.user.username, password=self.TEST_PASSWORD)
+
+    def _queue_session_message(self, text, extra_tags=''):
+        """Simulate SocialAuthExceptionMiddleware queuing a Django message in the current session."""
+        session = self.client.session
+        request = RequestFactory().get('/')
+        request.session = session
+        storage = SessionStorage(request)
+        storage.add(django_messages.constants.ERROR, text, extra_tags=extra_tags)
+        storage.update(HttpResponse())
+        session.save()
+
+    def test_requires_authentication(self):
+        """
+        Anonymous requests are rejected. DRF returns 403 (not 401) here
+        because SessionAuthentication.authenticate_header() returns None, so
+        no WWW-Authenticate challenge is issued.
+        """
+        self.client.logout()
+
+        response = self.client.get(self.URL)
+
+        assert response.status_code == 403
+
+    def test_returns_null_when_no_pending_message(self):
+        """No 'social-auth'-tagged message in the session means nothing to show."""
+        response = self.client.get(self.URL)
+
+        assert response.status_code == 200
+        assert response.json() == {'user_message': None}
+
+    def test_returns_pending_social_auth_message(self):
+        """A message tagged 'social-auth ...' (as SocialAuthExceptionMiddleware tags it) is surfaced."""
+        self._queue_session_message('This account is already in use.', extra_tags='social-auth tpa-saml')
+
+        response = self.client.get(self.URL)
+
+        assert response.status_code == 200
+        assert response.json() == {'user_message': 'This account is already in use.'}
+
+    def test_message_is_consumed_on_read(self):
+        """Matches Django's messages flash semantics: read once, gone on the next read."""
+        self._queue_session_message('This account is already in use.', extra_tags='social-auth tpa-saml')
+
+        self.client.get(self.URL)
+        second_response = self.client.get(self.URL)
+
+        assert second_response.json() == {'user_message': None}
+
+    def test_ignores_messages_without_social_auth_tag(self):
+        """Unrelated Django messages (e.g. from other flows) are not leaked through this endpoint."""
+        self._queue_session_message('Unrelated message', extra_tags='some-other-tag')
+
+        response = self.client.get(self.URL)
+
+        assert response.json() == {'user_message': None}
+
+    def test_preserves_unrelated_message_for_a_later_read(self):
+        """
+        Reading the storage at all marks the whole thing consumed, so an
+        unrelated message queued in the same session must be explicitly
+        re-queued -- otherwise it would be silently dropped here instead of
+        being shown wherever it was actually meant to be displayed.
+        """
+        self._queue_session_message('Unrelated message', extra_tags='some-other-tag')
+
+        response = self.client.get(self.URL)
+        assert response.json() == {'user_message': None}
+
+        # Read it back through a fresh storage bound to the client's session,
+        # the same way _queue_session_message wrote it.
+        session = self.client.session
+        request = RequestFactory().get('/')
+        request.session = session
+        remaining = list(SessionStorage(request))
+        assert [str(m) for m in remaining] == ['Unrelated message']
+        assert remaining[0].extra_tags == 'some-other-tag'
+
+    def test_only_first_social_auth_message_is_returned_rest_are_preserved(self):
+        """If somehow two social-auth messages are queued, only the first is returned this call."""
+        self._queue_session_message('First error', extra_tags='social-auth tpa-saml')
+        session = self.client.session
+        request = RequestFactory().get('/')
+        request.session = session
+        storage = SessionStorage(request)
+        storage.add(django_messages.constants.ERROR, 'Second error', extra_tags='social-auth tpa-saml')
+        storage.update(HttpResponse())
+        session.save()
+
+        first_response = self.client.get(self.URL)
+        second_response = self.client.get(self.URL)
+
+        assert first_response.json() == {'user_message': 'First error'}
+        assert second_response.json() == {'user_message': 'Second error'}
+
+
 class UserApiTestCase(UserAPITestCase):
     """
     Generalized test case class for specific implementations below

openedx/core/djangoapps/user_api/urls.py

@@ -115,6 +115,11 @@
         ProfileImageView.as_view(),
         name='accounts_profile_image_api'
     ),
+    path(
+        'v1/accounts/third_party_auth_error/',
+        user_api_views.ThirdPartyAuthErrorMessageView.as_view(),
+        name='third_party_auth_error_message'
+    ),
     re_path(
         fr'^v1/accounts/{settings.USERNAME_PATTERN}/deactivate/$',
         AccountDeactivationView.as_view(),