fix: enhance get_user_if_exists pipeline function with debugging capabilities

CHANGELOG.rst

@@ -13,6 +13,18 @@ Unreleased
 ----------
 
 
+[4.6.1] - 2025-08-26
+--------------------
+
+Fixed
+~~~~~
+
+* 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.
+
+  * 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.
+
 [4.6.0] - 2025-06-18
 --------------------
 

auth_backends/__init__.py

@@ -3,4 +3,4 @@
  These package is designed to be used primarily with Open edX Django projects, but should be compatible with non-edX
  projects as well.
 """
-__version__ = '4.6.0'  # pragma: no cover
+__version__ = '4.6.1'  # pragma: no cover

auth_backends/backends.py

@@ -31,7 +31,6 @@ def _to_language(locale):
     return locale.replace('_', '-').lower()
 
 
-# pylint: disable=abstract-method
 class EdXOAuth2(BaseOAuth2):
     """
     IMPORTANT: The oauth2 application must have access to the ``user_id`` scope in order

auth_backends/pipeline.py

@@ -22,27 +22,67 @@
 # .. toggle_target_removal_date: 2025-08-18
 SKIP_UPDATE_EMAIL_ON_USERNAME_MISMATCH = SettingToggle("SKIP_UPDATE_EMAIL_ON_USERNAME_MISMATCH", default=False)
 
+# .. toggle_name: IGNORE_LOGGED_IN_USER_ON_MISMATCH
+# .. toggle_implementation: SettingToggle
+# .. toggle_default: True
+# .. toggle_description: Controls behavior when there's a username mismatch between the logged-in user
+#    and social auth details. When enabled (True), ignores the logged-in user and proceeds with
+#    user lookup from social auth details. When disabled (False), proceeds with the logged-in user
+#    despite the mismatch. This toggle is for temporary rollout only to ensure we don't create bugs.
+# .. toggle_use_cases: temporary
+# .. toggle_creation_date: 2025-07-25
+# .. toggle_target_removal_date: 2025-09-25
+IGNORE_LOGGED_IN_USER_ON_MISMATCH = SettingToggle("IGNORE_LOGGED_IN_USER_ON_MISMATCH", default=True)
+
 
 # pylint: disable=unused-argument
 # The function parameters must be named exactly as they are below.
 # Do not change them to appease Pylint.
 def get_user_if_exists(strategy, details, user=None, *args, **kwargs):  # pylint: disable=keyword-arg-before-vararg
-    """Return a User with the given username iff the User exists."""
+    """
+    Return a User with the given username iff the User exists.
+    """
+    # .. custom_attribute_name: get_user_if_exists.ignore_toggle_enabled
+    # .. custom_attribute_description: Tracks whether the IGNORE_LOGGED_IN_USER_ON_MISMATCH
+    #    toggle is enabled during this pipeline execution.
+    set_custom_attribute('get_user_if_exists.ignore_toggle_enabled', IGNORE_LOGGED_IN_USER_ON_MISMATCH.is_enabled())
+
     if user:
-        return {'is_new': False}
+        # Check for username mismatch and toggle behavior
+        details_username = details.get('username')
+        user_username = getattr(user, 'username', None)
+
+        # .. custom_attribute_name: get_user_if_exists.has_details_username
+        # .. custom_attribute_description: Tracks whether the details contain a username field.
+        #    True if username is present, False if missing.
+        set_custom_attribute('get_user_if_exists.has_details_username', bool(details_username))
+
+        username_mismatch = details_username != user_username
+
+        # .. custom_attribute_name: get_user_if_exists.username_mismatch
+        # .. custom_attribute_description: Tracks whether there's a mismatch between
+        #    the username in the social details and the user's actual username.
+        #    True if usernames don't match, False if they match.
+        set_custom_attribute('get_user_if_exists.username_mismatch', username_mismatch)
+
+        if username_mismatch and IGNORE_LOGGED_IN_USER_ON_MISMATCH.is_enabled():
+            logger.info(
+                "Username mismatch detected. Username from Details: %s, Username from User: %s.",
+                details_username, user_username
+            )
+        else:
+            return {'is_new': False}
+
     try:
         username = details.get('username')
 
-        # Return the user if it exists
         return {
             'is_new': False,
             'user': User.objects.get(username=username)
         }
     except User.DoesNotExist:
-        # Fall to the default return value
         pass
 
-    # Nothing to return since we don't have a user
     return {}
 
 

auth_backends/tests/test_backends.py

@@ -4,6 +4,7 @@
 from calendar import timegm
 
 import jwt
+import responses
 import six
 from Cryptodome.PublicKey import RSA
 from django.core.cache import cache
@@ -37,10 +38,13 @@ def set_social_auth_setting(self, setting_name, value):
         # does not rely on Django settings.
         self.strategy.set_settings({f'SOCIAL_AUTH_{backend_name}_{setting_name}': value})
 
-    def access_token_body(self, request, _url, headers):
+    def access_token_body(self, request):
         """ Generates a response from the provider's access token endpoint. """
         # The backend should always request JWT access tokens, not Bearer.
-        body = six.moves.urllib.parse.parse_qs(request.body.decode('utf8'))
+        body_content = request.body
+        if isinstance(body_content, bytes):
+            body_content = body_content.decode('utf8')
+        body = six.moves.urllib.parse.parse_qs(body_content)
         self.assertEqual(body['token_type'], ['jwt'])
 
         expires_in = 3600
@@ -51,7 +55,16 @@ def access_token_body(self, request, _url, headers):
             'expires_in': expires_in,
             'access_token': access_token
         })
-        return 200, headers, body
+        return (200, {}, body)
+
+    def pre_complete_callback(self, start_url):
+        """ Override to properly set up the access token response with callback. """
+        responses.add_callback(
+            responses.POST,
+            url=self.backend.access_token_url(),
+            callback=self.access_token_body,
+            content_type="application/json",
+        )
 
     def create_jwt_access_token(self, expires_in=3600, issuer=None, key=None, alg='RS512'):
         """

auth_backends/tests/test_pipeline.py

@@ -1,39 +1,129 @@
 """ Tests for pipelines. """
 
-from unittest.mock import patch
+from unittest.mock import patch, call
+import ddt
 from django.contrib.auth import get_user_model
-from django.test import TestCase
+from django.test import TestCase, override_settings
 from social_django.utils import load_strategy
 
 from auth_backends.pipeline import get_user_if_exists, update_email
 
 User = get_user_model()
 
 
+@ddt.ddt
 class GetUserIfExistsPipelineTests(TestCase):
-    """ Tests for the get_user_if_exists pipeline function. """
+    """
+    Tests for the get_user_if_exists pipeline function.
+    """
 
     def setUp(self):
         super().setUp()
-        self.username = 'edx'
-        self.details = {'username': self.username}
-
-    def test_no_user_exists(self):
-        """ Verify an empty dict is returned if no user exists. """
-        actual = get_user_if_exists(None, self.details)
-        self.assertDictEqual(actual, {})
-
-    def test_existing_user(self):
-        """ Verify a dict with the user and extra details is returned if the user exists. """
-        user = User.objects.create(username=self.username)
-        actual = get_user_if_exists(None, self.details)
-        self.assertDictEqual(actual, {'is_new': False, 'user': user})
-
-    def test_get_user_if_exists(self):
-        """ Verify only the details are returned if a user is passed to the function. """
-        user = User.objects.create(username=self.username)
-        actual = get_user_if_exists(None, self.details, user=user)
-        self.assertDictEqual(actual, {'is_new': False})
+        self.details_for_existing_user = {'username': 'existing_user'}
+        self.details_for_non_existing_user = {'username': 'non_existing_user'}
+        self.details_for_different_user = {'username': 'different_user'}
+        self.existing_user = User.objects.create(**self.details_for_existing_user)
+
+    @ddt.data(True, False)  # test IGNORE_LOGGED_IN_USER_ON_MISMATCH toggle enabled/disabled
+    def test_no_user_exists(self, toggle_enabled):
+        """Returns empty dict if no user exists regardless of setting."""
+        with override_settings(IGNORE_LOGGED_IN_USER_ON_MISMATCH=toggle_enabled):
+            actual = get_user_if_exists(None, self.details_for_non_existing_user)
+            expected = {}
+            self.assertDictEqual(actual, expected)
+
+    @ddt.data(True, False)  # test IGNORE_LOGGED_IN_USER_ON_MISMATCH toggle enabled/disabled
+    @patch('auth_backends.pipeline.logger')
+    @patch('auth_backends.pipeline.set_custom_attribute')
+    def test_get_user_if_exists_no_current_user(self, toggle_enabled, mock_set_attribute, mock_logger):
+        """Returns details user when it can be found and there is no current user, regardless of toggle setting"""
+        with override_settings(IGNORE_LOGGED_IN_USER_ON_MISMATCH=toggle_enabled):
+            existing_user = self.existing_user
+            actual = get_user_if_exists(None, self.details_for_existing_user, user=None)
+
+            expected = {'is_new': False, 'user': existing_user}
+            self.assertDictEqual(actual, expected)
+            mock_set_attribute.assert_any_call('get_user_if_exists.ignore_toggle_enabled', toggle_enabled)
+            mock_logger.info.assert_not_called()
+
+    @ddt.data(True, False)  # test IGNORE_LOGGED_IN_USER_ON_MISMATCH toggle enabled/disabled
+    @patch('auth_backends.pipeline.logger')
+    @patch('auth_backends.pipeline.set_custom_attribute')
+    def test_get_user_if_exists_username_match(self, toggle_enabled, mock_set_attribute, mock_logger):
+        """Returns dict without user element when current user matches details username, regardless of toggle."""
+        with override_settings(IGNORE_LOGGED_IN_USER_ON_MISMATCH=toggle_enabled):
+            existing_user = self.existing_user
+            actual = get_user_if_exists(None, self.details_for_existing_user, user=existing_user)
+            expected = {'is_new': False}
+
+            self.assertDictEqual(actual, expected)
+            mock_set_attribute.assert_has_calls([
+                call('get_user_if_exists.ignore_toggle_enabled', toggle_enabled),
+                call('get_user_if_exists.has_details_username', True),
+                call('get_user_if_exists.username_mismatch', False)
+            ], any_order=True)
+            mock_logger.info.assert_not_called()
+
+    @ddt.data(True, False)  # test IGNORE_LOGGED_IN_USER_ON_MISMATCH toggle enabled/disabled
+    @patch('auth_backends.pipeline.logger')
+    @patch('auth_backends.pipeline.set_custom_attribute')
+    def test_get_user_if_exists_username_mismatch_and_details_user_found(
+        self, toggle_enabled, mock_set_attribute, mock_logger
+    ):
+        """Toggle enabled: return found details user. Toggle disabled: return dict without user element."""
+        with override_settings(IGNORE_LOGGED_IN_USER_ON_MISMATCH=toggle_enabled):
+            existing_user = self.existing_user
+            different_user = User.objects.create(**self.details_for_different_user)
+
+            actual = get_user_if_exists(None, self.details_for_different_user, user=existing_user)
+
+            if toggle_enabled:
+                expected = {'is_new': False, 'user': different_user}
+                mock_logger.info.assert_called_with(
+                    "Username mismatch detected. Username from Details: %s, Username from User: %s.",
+                    'different_user',
+                    'existing_user'
+                )
+            else:
+                expected = {'is_new': False}
+                mock_logger.info.assert_not_called()
+
+            self.assertDictEqual(actual, expected)
+            mock_set_attribute.assert_has_calls([
+                call('get_user_if_exists.ignore_toggle_enabled', toggle_enabled),
+                call('get_user_if_exists.has_details_username', True),
+                call('get_user_if_exists.username_mismatch', True)
+            ], any_order=True)
+
+    @ddt.data(True, False)  # test IGNORE_LOGGED_IN_USER_ON_MISMATCH toggle enabled/disabled
+    @patch('auth_backends.pipeline.logger')
+    @patch('auth_backends.pipeline.set_custom_attribute')
+    def test_get_user_if_exists_username_mismatch_details_user_not_found(
+        self, toggle_enabled, mock_set_attribute, mock_logger
+    ):
+        """Toggle enabled: return empty dict. Toggle disabled: return dict without user element."""
+        with override_settings(IGNORE_LOGGED_IN_USER_ON_MISMATCH=toggle_enabled):
+            existing_user = self.existing_user
+
+            actual = get_user_if_exists(None, self.details_for_non_existing_user, user=existing_user)
+
+            if toggle_enabled:
+                expected = {}
+                mock_logger.info.assert_called_with(
+                    "Username mismatch detected. Username from Details: %s, Username from User: %s.",
+                    'non_existing_user',
+                    'existing_user'
+                )
+            else:
+                expected = {'is_new': False}
+                mock_logger.info.assert_not_called()
+
+            self.assertDictEqual(actual, expected)
+            mock_set_attribute.assert_has_calls([
+                call('get_user_if_exists.ignore_toggle_enabled', toggle_enabled),
+                call('get_user_if_exists.has_details_username', True),
+                call('get_user_if_exists.username_mismatch', True)
+            ], any_order=True)
 
 
 class UpdateEmailPipelineTests(TestCase):

requirements/test.in

@@ -4,12 +4,15 @@
 -r base.txt                # Core dependencies
 
 coverage
+ddt                        # for running multiple test cases with multiple input
 edx-lint
 httpretty
 pycodestyle
-pycryptodomex                   # used for crypto tests
+pycryptodomex              # used for crypto tests
 pytest-cov
 pytest-django
+responses                  # required by ddt
 tox
+typing_extensions          # required by ddt
 unittest2
-edx-django-release-util   # Contains the reserved keyword check
+edx-django-release-util    # Contains the reserved keyword check