AAP-56394 Fix SAML and azuread authentication user lookup (ansible#872)

bhavenst · Bryan Havenstein · Claude · web-flow · commit 30ad472e569d · 2025-10-22T11:30:27.000-06:00
More comprehensive fix for AuthenticatorUser lookup.  Fixes SAML and AzureAD and all authenticator types tested via manual and ATF testing

---------

Co-authored-by: Bryan Havenstein &lt;bhavenst@redhat.com&gt;
Co-authored-by: Claude &lt;claude@anthropic.com&gt;
diff --git a/ansible_base/authentication/utils/authentication.py b/ansible_base/authentication/utils/authentication.py
@@ -1,6 +1,6 @@
 import importlib
 import logging
-from typing import Optional, Tuple, Union
+from typing import List, Optional, Tuple, Union
 
 from django.conf import settings
 from django.contrib.auth import get_user_model
@@ -137,16 +137,16 @@ def determine_username_from_uid_social(**kwargs) -> dict:
     if authenticator.setting('USERNAME_IS_FULL_EMAIL', False):
         selected_username = kwargs.get('details', {}).get('email', None)
     else:
-        if authenticator.type == 'SAML':
-            selected_username = kwargs.get('uid', None)
-        else:
-            selected_username = kwargs.get('details', {}).get('username', None)
+        selected_username = kwargs.get('details', {}).get('username', None)
     if not selected_username:
         raise_auth_exception(
             _('Unable to get associated username from details, expected entry "username". Full user details: %(details)s')
             % {'details': kwargs.get("details", None)},
             backend=authenticator,
         )
+    # Different authenticators use different fields for AuthenticatorUser lookup
+    # Create a filter list
+    uid_filter = [selected_username, kwargs.get('uid')] if kwargs.get('uid', None) else [selected_username]
 
     alt_uid = authenticator.get_alternative_uid(**kwargs)
     email = kwargs.get('details', {}).get('email', None)
@@ -156,7 +156,7 @@ def determine_username_from_uid_social(**kwargs) -> dict:
     ):
         username = migrated_username
     else:
-        username = determine_username_from_uid(selected_username, email, authenticator.database_instance)
+        username = determine_username_from_uid(uid=selected_username, uid_filter=uid_filter, email=email, authenticator=authenticator.database_instance)
 
     return {"username": username}
 
@@ -238,7 +238,9 @@ def _handle_email_fallback_strategy(uid: str, email: Union[str, list[str], None]
 #         return new_username
 
 
-def determine_username_from_uid(uid: str = None, email: Union[str, list[str], None] = None, authenticator: Authenticator = None) -> str:
+def determine_username_from_uid(
+    uid: str, uid_filter: Union[List[str], None] = None, email: Union[str, list[str], None] = None, authenticator: Authenticator = None
+) -> str:
     """
     Determine what the username for the User object will be from the given uid and authenticator
     This will take uid like "bella" and search for an AuthenticatorUser and return:
@@ -259,7 +261,11 @@ def determine_username_from_uid(uid: str = None, email: Union[str, list[str], No
         raise
 
     # If we have an AuthenticatorUser with the exact uid and provider than we have a match
-    if (exact_match := AuthenticatorUser.objects.filter(uid=uid, provider=authenticator)).exists():
+    if uid_filter:
+        exact_match = AuthenticatorUser.objects.filter(uid__in=uid_filter, provider=authenticator)
+    else:
+        exact_match = AuthenticatorUser.objects.filter(uid=uid, provider=authenticator)
+    if exact_match.exists():
         new_username = exact_match.first().user.username
         logger.info(f"Authenticator {authenticator.name} already authenticated {uid} as {new_username}")
         return new_username
@@ -319,7 +325,7 @@ def get_or_create_authenticator_user(
     if migrated_username := migrate_from_existing_authenticator(uid=uid, alt_uid=None, authenticator=authenticator, preferred_username=uid):
         username = migrated_username
     else:
-        username = determine_username_from_uid(uid, email, authenticator)
+        username = determine_username_from_uid(uid=uid, uid_filter=[uid], email=email, authenticator=authenticator)
 
     # Step 2: Now that the username is finalized, try to find the AuthenticatorUser.
     auth_user = None
diff --git a/test_app/tests/authentication/test_merge_strategy_email_fallback.py b/test_app/tests/authentication/test_merge_strategy_email_fallback.py
@@ -179,3 +179,110 @@ def test_email_fallback_complex_scenario(self, github_authenticator, ldap_authen
 
         # Should also merge with first user
         assert ldap_username == 'complex_user'
+
+
+@pytest.mark.django_db
+class TestUidFilterParameter:
+    """Test the uid_filter parameter functionality in determine_username_from_uid."""
+
+    def test_uid_filter_single_matching_uid(self, existing_user_with_github_auth, github_authenticator):
+        """Test uid_filter with single UID that matches existing AuthenticatorUser."""
+        # existing_user_with_github_auth has uid 'john' with GitHub authenticator
+
+        username = determine_username_from_uid(uid='different_uid', uid_filter=['john'], email='test@example.com', authenticator=github_authenticator)
+
+        # Should match existing user even though uid parameter is different
+        assert username == 'john'
+
+    def test_uid_filter_multiple_uids_one_matches(self, existing_user_with_github_auth, github_authenticator):
+        """Test uid_filter with multiple UIDs where one matches existing AuthenticatorUser."""
+        # existing_user_with_github_auth has uid 'john' with GitHub authenticator
+
+        username = determine_username_from_uid(
+            uid='different_uid', uid_filter=['nonexistent1', 'john', 'nonexistent2'], email='test@example.com', authenticator=github_authenticator
+        )
+
+        # Should match existing user with uid 'john'
+        assert username == 'john'
+
+    def test_uid_filter_multiple_uids_multiple_matches(self, github_authenticator):
+        """Test uid_filter with multiple UIDs that all match existing AuthenticatorUsers."""
+        # Create two users with different UIDs for same authenticator
+        user1 = User.objects.create(username='user1', email='user1@example.com')
+        user2 = User.objects.create(username='user2', email='user2@example.com')
+
+        AuthenticatorUser.objects.create(user=user1, uid='uid1', provider=github_authenticator, extra_data={})
+        AuthenticatorUser.objects.create(user=user2, uid='uid2', provider=github_authenticator, extra_data={})
+
+        username = determine_username_from_uid(uid='different_uid', uid_filter=['uid1', 'uid2'], email='test@example.com', authenticator=github_authenticator)
+
+        # Should return the first match found
+        assert username in ['user1', 'user2']
+
+    def test_uid_filter_no_matches(self, github_authenticator):
+        """Test uid_filter with UIDs that don't match any existing AuthenticatorUser."""
+        username = determine_username_from_uid(
+            uid='new_user', uid_filter=['nonexistent1', 'nonexistent2'], email='newuser@example.com', authenticator=github_authenticator
+        )
+
+        # Should fall back to email fallback strategy since no exact matches
+        assert username == 'new_user'
+
+    def test_uid_filter_empty_list(self, github_authenticator):
+        """Test uid_filter as empty list."""
+        username = determine_username_from_uid(uid='new_user', uid_filter=[], email='newuser@example.com', authenticator=github_authenticator)
+
+        # Should fall back to default behavior (check uid parameter directly)
+        assert username == 'new_user'
+
+    def test_uid_filter_none_uses_default_behavior(self, existing_user_with_github_auth, github_authenticator):
+        """Test uid_filter as None uses default behavior."""
+        # existing_user_with_github_auth has uid 'john' with GitHub authenticator
+
+        username = determine_username_from_uid(uid='john', uid_filter=None, email='test@example.com', authenticator=github_authenticator)
+
+        # Should match existing user using uid parameter
+        assert username == 'john'
+
+    def test_uid_filter_vs_uid_parameter_difference(self, existing_user_with_github_auth, github_authenticator):
+        """Test when uid parameter differs from UIDs in uid_filter."""
+        # existing_user_with_github_auth has uid 'john' with GitHub authenticator
+
+        # Case 1: uid_filter matches, uid parameter doesn't
+        username = determine_username_from_uid(uid='different_uid', uid_filter=['john'], email='test@example.com', authenticator=github_authenticator)
+        assert username == 'john'  # Should use uid_filter match
+
+        # Case 2: uid parameter matches, uid_filter doesn't
+        username = determine_username_from_uid(uid='john', uid_filter=['nonexistent'], email='test@example.com', authenticator=github_authenticator)
+        # Should fall back to email strategy since uid_filter didn't match
+        # Uses dupe user strategy and suffixes with <hash>
+        assert username.startswith('john')
+
+    def test_uid_filter_with_different_authenticator(self, existing_user_with_github_auth, ldap_authenticator):
+        """Test uid_filter matching UID from different authenticator."""
+        # existing_user_with_github_auth has uid 'john' with GitHub authenticator
+
+        username = determine_username_from_uid(
+            uid='different_uid',
+            uid_filter=['john'],  # This UID exists but for GitHub, not LDAP
+            email='john@example.com',
+            authenticator=ldap_authenticator,  # Different authenticator
+        )
+
+        # Should not match because authenticator is different, but should merge via email
+        assert username == 'john'
+
+    def test_uid_filter_cross_authenticator_no_email_merge(self, existing_user_with_github_auth, ldap_authenticator):
+        """Test uid_filter with different authenticator and no email merge."""
+        # existing_user_with_github_auth has uid 'john' with GitHub authenticator
+
+        username = determine_username_from_uid(
+            uid='different_uid',
+            uid_filter=['john'],  # This UID exists but for GitHub, not LDAP
+            email='different@example.com',  # Different email, won't merge
+            authenticator=ldap_authenticator,
+        )
+
+        # Should create new unique username since no match and different email
+        assert username != 'john'
+        assert 'different_uid' in username or username == 'different_uid'
diff --git a/test_app/tests/authentication/utils/test_authentication.py b/test_app/tests/authentication/utils/test_authentication.py
@@ -85,7 +85,7 @@ def test_determine_username_from_uid(
         # Use different log levels based on the scenario
         log_level = 'warning' if related_authenticator == 'ldap' else 'info'
         with expected_log(self.logger, log_level, info_message):
-            new_username = authentication.determine_username_from_uid(uid, "", local_authenticator)
+            new_username = authentication.determine_username_from_uid(uid=uid, email="", authenticator=local_authenticator)
             if expected_username:
                 assert new_username == random_user.username
             else:
@@ -173,7 +173,7 @@ def test_external_system_user_login(self, request, auth_fixture):
         uid = settings.SYSTEM_USERNAME
         authenticator = request.getfixturevalue(auth_fixture)
         with pytest.raises(AuthException):
-            authentication.determine_username_from_uid(uid, "", authenticator)
+            authentication.determine_username_from_uid(uid=uid, email="", authenticator=authenticator)
         with pytest.raises(AuthException):
             authentication.get_or_create_authenticator_user(uid=uid, email="", authenticator=authenticator, user_details={}, extra_data={})
 
@@ -249,27 +249,6 @@ def test_determine_username_from_uid_social_happy_path(self, ldap_authenticator)
         )
         assert response == {'username': 'Bob'}
 
-    @pytest.mark.parametrize(
-        "auth_fixture, expected_username, uid_param, details_username",
-        [
-            # SAML authenticator uses uid parameter (line 141)
-            ("saml_authenticator", "SamlUser", "SamlUser", "should_not_be_used"),
-            # Non-SAML authenticators use details['username'] (line 143)
-            ("ldap_authenticator", "LdapUser", "should_not_be_used", "LdapUser"),
-            ("oidc_authenticator", "OidcUser", "should_not_be_used", "OidcUser"),
-            ("keycloak_authenticator", "KeycloakUser", "should_not_be_used", "KeycloakUser"),
-        ],
-    )
-    def test_determine_username_from_uid_social_authenticator_types(self, request, auth_fixture, expected_username, uid_param, details_username):
-        """Test username selection logic for different authenticator types (lines 140-143)"""
-        authenticator = request.getfixturevalue(auth_fixture)
-        response = authentication.determine_username_from_uid_social(
-            details={'username': details_username},
-            backend=get_authenticator_class(authenticator.type)(database_instance=authenticator),
-            uid=uid_param,
-        )
-        assert response == {'username': expected_username}
-
     def test_raise_auth_exception(self):
         try:
             authentication.raise_auth_exception('testing')
@@ -288,3 +267,113 @@ def test_raise_auth_exception_in_logs(self, local_authenticator, expected_log):
                 )
             except AuthException:
                 pass
+
+    @pytest.mark.parametrize(
+        "uid_param, expected_uid_filter",
+        [
+            ("user123", ["testuser", "user123"]),  # When uid is provided, should include both selected_username and uid
+            (None, ["testuser"]),  # When uid is None, should only include selected_username
+            ("", ["testuser"]),  # When uid is empty string, should only include selected_username
+        ],
+    )
+    def test_determine_username_from_uid_social_uid_filter_construction(self, ldap_authenticator, uid_param, expected_uid_filter):
+        """Test that uid_filter is constructed correctly in determine_username_from_uid_social."""
+        from unittest.mock import patch
+
+        authenticator_class = get_authenticator_class(ldap_authenticator.type)(database_instance=ldap_authenticator)
+
+        # Mock migrate_from_existing_authenticator to return None (no migration)
+        # Mock determine_username_from_uid to capture the uid_filter parameter
+        with patch('ansible_base.authentication.utils.authentication.migrate_from_existing_authenticator') as mock_migrate:
+            mock_migrate.return_value = None
+
+            with patch('ansible_base.authentication.utils.authentication.determine_username_from_uid') as mock_determine_uid:
+                mock_determine_uid.return_value = 'testuser'
+
+                kwargs = {
+                    'details': {'username': 'testuser', 'email': 'test@example.com'},
+                    'backend': authenticator_class,
+                }
+                if uid_param is not None:
+                    kwargs['uid'] = uid_param
+
+                response = authentication.determine_username_from_uid_social(**kwargs)
+
+                # Verify determine_username_from_uid was called with correct uid_filter
+                mock_determine_uid.assert_called_once_with(
+                    uid='testuser', uid_filter=expected_uid_filter, email='test@example.com', authenticator=ldap_authenticator
+                )
+                assert response == {'username': 'testuser'}
+
+    def test_determine_username_from_uid_social_uid_filter_with_email_username(self, ldap_authenticator):
+        """Test uid_filter construction when USERNAME_IS_FULL_EMAIL is True."""
+        from unittest.mock import patch
+
+        authenticator_class = get_authenticator_class(ldap_authenticator.type)(database_instance=ldap_authenticator)
+
+        # Mock migrate_from_existing_authenticator to return None (no migration)
+        # Mock the setting method to return True for USERNAME_IS_FULL_EMAIL
+        with patch('ansible_base.authentication.utils.authentication.migrate_from_existing_authenticator') as mock_migrate:
+            mock_migrate.return_value = None
+
+            with patch.object(authenticator_class, 'setting') as mock_setting:
+                mock_setting.return_value = True
+
+                with patch('ansible_base.authentication.utils.authentication.determine_username_from_uid') as mock_determine_uid:
+                    mock_determine_uid.return_value = 'user@example.com'
+
+                    response = authentication.determine_username_from_uid_social(
+                        details={'username': 'testuser', 'email': 'user@example.com'}, backend=authenticator_class, uid='uid123'
+                    )
+
+                    # When USERNAME_IS_FULL_EMAIL is True, selected_username should be the email
+                    mock_determine_uid.assert_called_once_with(
+                        uid='user@example.com', uid_filter=['user@example.com', 'uid123'], email='user@example.com', authenticator=ldap_authenticator
+                    )
+                    assert response == {'username': 'user@example.com'}
+
+    def test_determine_username_from_uid_social_uid_filter_no_migration(self, ldap_authenticator):
+        """Test that uid_filter is passed to determine_username_from_uid when migration doesn't occur."""
+        from unittest.mock import patch
+
+        authenticator_class = get_authenticator_class(ldap_authenticator.type)(database_instance=ldap_authenticator)
+
+        with patch('ansible_base.authentication.utils.authentication.migrate_from_existing_authenticator') as mock_migrate:
+            mock_migrate.return_value = None  # No migration
+
+            with patch('ansible_base.authentication.utils.authentication.determine_username_from_uid') as mock_determine_uid:
+                mock_determine_uid.return_value = 'finaluser'
+
+                response = authentication.determine_username_from_uid_social(
+                    details={'username': 'originaluser', 'email': 'test@example.com'}, backend=authenticator_class, uid='uid456'
+                )
+
+                # Verify migration was attempted first
+                mock_migrate.assert_called_once_with(uid='uid456', alt_uid=None, authenticator=ldap_authenticator, preferred_username='originaluser')
+
+                # Verify determine_username_from_uid was called with correct uid_filter
+                mock_determine_uid.assert_called_once_with(
+                    uid='originaluser', uid_filter=['originaluser', 'uid456'], email='test@example.com', authenticator=ldap_authenticator
+                )
+                assert response == {'username': 'finaluser'}
+
+    def test_determine_username_from_uid_social_uid_filter_with_migration(self, ldap_authenticator):
+        """Test that determine_username_from_uid is not called when migration succeeds."""
+        from unittest.mock import patch
+
+        authenticator_class = get_authenticator_class(ldap_authenticator.type)(database_instance=ldap_authenticator)
+
+        with patch('ansible_base.authentication.utils.authentication.migrate_from_existing_authenticator') as mock_migrate:
+            mock_migrate.return_value = 'migrated_user'  # Migration succeeded
+
+            with patch('ansible_base.authentication.utils.authentication.determine_username_from_uid') as mock_determine_uid:
+                response = authentication.determine_username_from_uid_social(
+                    details={'username': 'originaluser', 'email': 'test@example.com'}, backend=authenticator_class, uid='uid789'
+                )
+
+                # Verify migration was attempted
+                mock_migrate.assert_called_once_with(uid='uid789', alt_uid=None, authenticator=ldap_authenticator, preferred_username='originaluser')
+
+                # determine_username_from_uid should NOT be called when migration succeeds
+                mock_determine_uid.assert_not_called()
+                assert response == {'username': 'migrated_user'}
