fix: better username lookup in tpa pipeline (#33145)

common/djangoapps/third_party_auth/pipeline.py

@@ -1024,6 +1024,11 @@ def get_username(strategy, details, backend, user=None, *args, **kwargs):  # lin
             )
     else:
         final_username = storage.user.get_username(user)
+    logger.info(
+        '[THIRD_PARTY_AUTH] get_username complete: '
+        f'details={details}, '
+        f'final_username={final_username}'
+    )
     return {'username': final_username}
 
 

common/djangoapps/third_party_auth/tests/test_models.py

@@ -2,10 +2,10 @@
 Tests for third_party_auth/models.py.
 """
 import unittest
-from django.test import TestCase
+from django.test import TestCase, override_settings
 
 from .factories import SAMLProviderConfigFactory
-from ..models import SAMLProviderConfig
+from ..models import SAMLProviderConfig, clean_username
 
 
 class TestSamlProviderConfigModel(TestCase, unittest.TestCase):
@@ -39,3 +39,17 @@ class TestSamlProviderConfigModel(TestCase, unittest.TestCase):
             bad_config = SAMLProviderConfig(entity_id=self.saml_provider_config.entity_id)
             bad_config.save()
         assert ctx.records[0].msg == f'Entity ID: {self.saml_provider_config.entity_id} already in use'
+
+    @override_settings(FEATURES={'ENABLE_UNICODE_USERNAME': False})
+    def test_clean_username_unicode_disabled(self):
+        """
+        Test the username cleaner function with unicode disabled
+        """
+        assert clean_username('ItJüstWòrks™') == 'ItJ_stW_rks'
+
+    @override_settings(FEATURES={'ENABLE_UNICODE_USERNAME': True})
+    def test_clean_username_unicode_enabled(self):
+        """
+        Test the username cleaner function with unicode enabled
+        """
+        assert clean_username('ItJüstWòrks™') == 'ItJüstWòrks'

common/djangoapps/third_party_auth/tests/test_pipeline.py

@@ -107,3 +107,14 @@ class PipelineOverridesTest(SamlIntegrationTestUtilities, IntegrationTestMixin,
                 mock_randint.side_effect = [1, 2, 4]
                 final_username = pipeline.get_username(strategy, details, self.provider.backend_class())
                 assert expected_username == final_username['username']
+
+    def test_get_username(self):
+        """
+        Test get_username method when the first username candidate is available
+        """
+        details = {
+            "username": 'invalid'
+        }
+        __, strategy = self.get_request_and_strategy()
+        final_username = pipeline.get_username(strategy, details, self.provider.backend_class())
+        assert final_username['username'] == 'invalid'

common/djangoapps/third_party_auth/utils.py

@@ -16,6 +16,10 @@ from lxml import etree
 from onelogin.saml2.utils import OneLogin_Saml2_Utils
 from requests import exceptions
 from social_core.pipeline.social_auth import associate_by_email
+from common.djangoapps.student.models import (
+    email_exists_or_retired,
+    username_exists_or_retired
+)
 
 from common.djangoapps.third_party_auth.models import OAuth2ProviderConfig, SAMLProviderData
 from openedx.core.djangolib.markup import Text
@@ -139,16 +143,13 @@ def user_exists(details):
     Returns:
         (bool): True if user with given details exists, `False` otherwise.
     """
-    user_queryset_filter = {}
     email = details.get('email')
     username = details.get('username')
     if email:
-        user_queryset_filter['email'] = email
+        return email_exists_or_retired(email)
     elif username:
-        user_queryset_filter['username__iexact'] = username
-
-    if user_queryset_filter:
-        return User.objects.filter(**user_queryset_filter).exists()
+        # username__iexact preserves the original case insensitivity
+        return User.objects.filter(username__iexact=username).exists() or username_exists_or_retired(username)
 
     return False