fix: added logout user before sso pipeline starts

CHANGELOG.rst

@@ -13,17 +13,15 @@ Unreleased
 ----------
 
 
-[4.6.1] - 2025-08-26
+[4.6.1] - 2025-09-23
 --------------------
 
-Fixed
+Added
 ~~~~~
 
-* Fixed authentication issues where the social auth details username is not being respected as the user to be logged in in certain cases.
-  When the already logged-in user does not match the social auth details user, the pipeline should log in the new user, rather than leaving some other user logged in.
+* Enhanced OAuth2 authentication backend to logout any old session before initiating a new OAuth2 flow. This prevents user association conflicts with the previously logged-in user.
 
-  * Added temporary rollout toggle IGNORE_LOGGED_IN_USER_ON_MISMATCH to ensure the bug fix doesn't introduce any issues.
-  * Enhanced get_user_if_exists function with monitoring capabilities for debugging authentication conflicts.
+  * Added temporary rollout toggle ENABLE_OAUTH_SESSION_CLEANUP to control session cleanup during OAuth start process.
 
 [4.6.0] - 2025-06-18
 --------------------

auth_backends/backends.py

@@ -2,9 +2,27 @@
 
 For more information visit https://docs.djangoproject.com/en/dev/topics/auth/customizing/.
 """
+import logging
 import jwt
+from django.contrib.auth import logout
 from django.dispatch import Signal
 from social_core.backends.oauth import BaseOAuth2
+from edx_toggles.toggles import SettingToggle
+from edx_django_utils.monitoring import set_custom_attribute
+
+logger = logging.getLogger(__name__)
+
+# .. toggle_name: ENABLE_OAUTH_SESSION_CLEANUP
+# .. toggle_implementation: SettingToggle
+# .. toggle_default: True
+# .. toggle_description: Controls whether to perform session cleanup during OAuth start.
+#    When enabled (True), existing user sessions are cleared before OAuth authentication
+#    to prevent user association conflicts. When disabled (False), session cleanup is skipped.
+#    This toggle allows for gradual rollout and quick rollback if issues arise.
+# .. toggle_use_cases: temporary
+# .. toggle_creation_date: 2025-09-15
+# .. toggle_target_removal_date: 2025-11-15
+ENABLE_OAUTH_SESSION_CLEANUP = SettingToggle("ENABLE_OAUTH_SESSION_CLEANUP", default=True)
 
 PROFILE_CLAIMS_TO_DETAILS_KEY_MAP = {
     'preferred_username': 'username',
@@ -69,6 +87,56 @@ def logout_url(self):
         else:
             return self.end_session_url()
 
+    def start(self):
+        """Initialize OAuth authentication with optional session cleanup."""
+
+        # .. custom_attribute_name: session_cleanup.toggle_enabled
+        # .. custom_attribute_description: Tracks whether the ENABLE_OAUTH_SESSION_CLEANUP
+        #    toggle is enabled during OAuth start.
+        set_custom_attribute('session_cleanup.toggle_enabled', ENABLE_OAUTH_SESSION_CLEANUP.is_enabled())
+
+        request = self.strategy.request if hasattr(self.strategy, 'request') else None
+
+        # .. custom_attribute_name: session_cleanup.has_request
+        # .. custom_attribute_description: Tracks whether a request object is available
+        #    during OAuth start. True if request exists, False if missing.
+        set_custom_attribute('session_cleanup.has_request', request is not None)
+
+        user_authenticated = (
+            request is not None and
+            hasattr(request, 'user') and
+            request.user.is_authenticated
+        )
+
+        # .. custom_attribute_name: session_cleanup.logout_required
+        # .. custom_attribute_description: Tracks whether a user was authenticated
+        #    before session cleanup. True if user was logged in, False otherwise.
+        set_custom_attribute('session_cleanup.logout_required', user_authenticated)
+
+        if user_authenticated and ENABLE_OAUTH_SESSION_CLEANUP.is_enabled():
+            existing_username = getattr(request.user, 'username', 'unknown')
+
+            # .. custom_attribute_name: session_cleanup.logged_out_username
+            # .. custom_attribute_description: Records the username that was logged out
+            #    during session cleanup for tracking and debugging purposes.
+            set_custom_attribute('session_cleanup.logged_out_username', existing_username)
+
+            logger.info(
+                "OAuth start: Performing session cleanup for user '%s'",
+                existing_username
+            )
+
+            logout(request)
+
+            # .. custom_attribute_name: session_cleanup.logout_performed
+            # .. custom_attribute_description: Indicates that session cleanup was
+            #    actually performed during OAuth start.
+            set_custom_attribute('session_cleanup.logout_performed', True)
+        else:
+            set_custom_attribute('session_cleanup.logout_performed', False)
+
+        return super().start()
+
     def authorization_url(self):
         url_root = self.get_public_or_internal_url_root()
         return f'{url_root}/oauth2/authorize'

auth_backends/tests/test_backends.py

@@ -2,15 +2,25 @@
 import datetime
 import json
 from calendar import timegm
+from unittest.mock import patch, call
 
+import ddt
 import jwt
+import pytest
 import responses
 import six
 from Cryptodome.PublicKey import RSA
+from django.contrib.auth import get_user_model
+from django.contrib.sessions.middleware import SessionMiddleware
 from django.core.cache import cache
+from django.test import RequestFactory
+from django.test.utils import override_settings
 from social_core.tests.backends.oauth import OAuth2Test
 
+User = get_user_model()
 
+
+@ddt.ddt
 class EdXOAuth2Tests(OAuth2Test):
     """ Tests for the EdXOAuth2 backend. """
 
@@ -116,6 +126,56 @@ def extra_settings(self):
     def test_login(self):
         self.do_login()
 
+    @pytest.mark.django_db
+    @ddt.data(True, False)  # Test session cleanup with both toggle enabled and disabled
+    @patch('auth_backends.backends.set_custom_attribute')
+    @patch('auth_backends.backends.logger')
+    def test_start_with_session_cleanup(self, toggle_enabled, mock_logger, mock_set_attr):
+        """Test start method for session cleanup of existing user with toggle variation."""
+        with override_settings(ENABLE_OAUTH_SESSION_CLEANUP=toggle_enabled):
+            existing_user = User.objects.create_user(username='existing_user', email='existing@example.com')
+
+            request = RequestFactory().get('/auth/login/edx-oauth2/')
+            request.user = existing_user
+
+            middleware = SessionMiddleware(lambda req: None)
+            middleware.process_request(request)
+            request.session.save()
+
+            initial_session_key = request.session.session_key
+
+            self.backend.strategy.request = request
+
+            self.do_start()
+
+            if toggle_enabled:
+                self.assertNotEqual(request.session.session_key, initial_session_key)
+
+                self.assertTrue(request.user.is_anonymous)
+
+                mock_set_attr.assert_has_calls([
+                    call('session_cleanup.toggle_enabled', True),
+                    call('session_cleanup.logout_performed', True),
+                    call('session_cleanup.logged_out_username', 'existing_user')
+                ], any_order=True)
+
+                mock_logger.info.assert_called_with(
+                    "OAuth start: Performing session cleanup for user '%s'",
+                    'existing_user'
+                )
+            else:
+                self.assertEqual(request.session.session_key, initial_session_key)
+
+                self.assertEqual(request.user, existing_user)
+                self.assertFalse(request.user.is_anonymous)
+
+                mock_set_attr.assert_has_calls([
+                    call('session_cleanup.toggle_enabled', False),
+                    call('session_cleanup.logout_performed', False)
+                ], any_order=True)
+
+                mock_logger.info.assert_not_called()
+
     def test_partial_pipeline(self):
         self.do_partial_pipeline()