AAP-50130: Fallback authentication only on migrated users (#778)

## Description
<!-- Mandatory: Provide a clear, concise description of the changes and
their purpose -->
- What is being changed? Modifies the controller fallback authentication
so that it only is triggered if the gateway user has flag
`use_controller_password=true`. Once the controller fallback
authentication is completed, this flag is set to false to disallow users
from then signing in again with their controller password.
- Why is this change needed? This change is needed to ensure passwords
can only be set from their controller passwords once. This provides more
security benefits so that the controller password cannot always be used
as a "fallback" password, limiting exposure if this password had ever
been leaked after it has been changed.
- How does this change address the issue? This change addresses the
issue by adding another safeguard check in the fallback controller
authentication logic to validate that this flag is set on the user, and
if it is, it will update it to false, after the password has been
modified.

## Type of Change
<!-- Mandatory: Check one or more boxes that apply -->
- [x] Bug fix (non-breaking change which fixes an issue)
- [ ] New feature (non-breaking change which adds functionality)
- [ ] Breaking change (fix or feature that would cause existing
functionality to not work as expected)
- [ ] Documentation update
- [ ] Test update
- [ ] Refactoring (no functional changes)
- [ ] Development environment change
- [ ] Configuration change

## Self-Review Checklist
<!-- These items help ensure quality - they complement our automated CI
checks -->
- [x] I have performed a self-review of my code
- [x] I have added relevant comments to complex code sections
- [x] I have updated documentation where needed
- [x] I have considered the security impact of these changes
- [x] I have considered performance implications
- [x] I have thought about error handling and edge cases
- [x] I have tested the changes in my local environment

## Testing Instructions
<!-- Optional for test-only changes. Mandatory for all other changes -->
<!-- Must be detailed enough for reviewers to reproduce -->
### Prerequisites
<!-- List any specific setup required -->

### Steps to Test

Testing instructions use aap-dev for validation

1. Deploy AAP Dev, with this branch, and also this [gateway
PR](ansible-automation-platform/aap-gateway#918)
2. Check `/gateway/v1/users/`, validate the administrator account has
`use_controller_password=false`
3. Attempt to sign in via gateway with administrator controller
password. This should fail. Only gateway password will work
4. Exec into controller-task container and run the following command to
create a user on controller only -
```bash
ANSIBLE_REVERSE_RESOURCE_SYNC=false awx-manage shell_plus --quiet-load -c "User.objects.create_user(username='dummy', email='[email protected]', password='test', first_name='dummy', last_name='user')"
```
5. Exec into gateway-api pod and run the migrate_service_data command -
```bash
aap-gateway-manage migrate_service_data --username=admin
```
6. Check `/gateway/v1/users`, validate dummy account exists, with
`use_controller_password=true`
7. Sign into gateway with the dummy user account, using the password
created in controller, sign in should succeed.
8. Check `/gateway/v1/users`, see that, with
`use_controller_password=false`
9. Change dummy user password in gateway
10. Attempt to sign in with dummy user again using original controller
password. Signin should fail. Only gateway password can be used now.

### Expected Results

Controller fallback authentication can only happen once for any account,
and only on those accounts which were migrated via migrate_service_data

## Additional Context
<!-- Optional but helpful information -->

### Required Actions
<!-- Check if changes require work in other areas -->
<!-- Remove section if no external actions needed -->
- [ ] Requires documentation updates
  <!-- API docs, feature docs, deployment guides -->
- [ ] Requires downstream repository changes
  <!-- Specify repos: django-ansible-base, eda-server, etc. -->
- [ ] Requires infrastructure/deployment changes
  <!-- CI/CD, installer updates, new services -->
- [ ] Requires coordination with other teams
  <!-- UI team, platform services, infrastructure -->
- [ ] Blocked by PR/MR: #XXX
  <!-- Reference blocking PRs/MRs with brief context -->

### Screenshots/Logs
<!-- Add if relevant to demonstrate the changes -->

Author: zkayyali812

ansible_base/authentication/authenticator_plugins/local.py

@@ -94,11 +94,17 @@ def _can_authenticate_from_controller(self, username, password):
         If the user is valid, update the gateway users credentials with the controller credentials.
         """
         try:
-            UserModel._default_manager.get_by_natural_key(username)
+            user = UserModel._default_manager.get_by_natural_key(username)
         except UserModel.DoesNotExist:
             logger.warning(f"User '{username}' does not exist in the database.")
             return False
 
+        # Skip controller authentication if user has use_controller_password field set to False
+        # Default to False when field doesn't exist (test environments)
+        if not getattr(user, 'use_controller_password', False):
+            logger.warning(f"User '{username}' password not in Controller.")
+            return False
+
         if controller_user := self._get_controller_user(username, password):
             # Validate controller_user has a ldap_dn field, if it is not None, then the user is a local user
             ldap_dn = controller_user.get("ldap_dn")
@@ -174,7 +180,14 @@ def update_gateway_user(self, username, password):
         """
         user = UserModel._default_manager.get_by_natural_key(username)
         user.set_password(password)
-        user.save(update_fields=['password'])
+
+        # Set use_controller_password to False if the field exists
+        update_fields = ['password']
+        if hasattr(user, 'use_controller_password'):
+            user.use_controller_password = False
+            update_fields.append('use_controller_password')
+
+        user.save(update_fields=update_fields)
         logger.info(f"Updated user {username} gateway account")
 
     def _convert_to_seconds(self, s):

ansible_base/authentication/utils/authentication.py

@@ -145,11 +145,7 @@ def _handle_no_merge_strategy(uid: str, authenticator: Authenticator) -> str:
     if AuthenticatorUser.objects.filter(Q(uid=uid) | Q(user__username=uid)).exists():
         # Some other provider is providing this username so we need to create our own username
         new_username = get_local_username({'username': uid})
-        logger.info(
-            f'Authenticator {authenticator.name} wants to authenticate {uid} but that'
-            f' username is already in use by another authenticator,'
-            f' the user from this authenticator will be {new_username}'
-        )
+        logger.warning(f"User {uid} is already associated with an existing user, creating a new user with username {new_username}")
         return new_username
     else:
         # We didn't have an exact match but no other provider is servicing this uid so lets return that for usage
@@ -188,11 +184,7 @@ def _handle_email_fallback_strategy(uid: str, email: Union[str, list[str], None]
     # PROPOSAL FLOW: Does an AAP User with this email exist? No
     else:
         new_username = get_local_username({'username': uid})
-        logger.warning(
-            f"Authenticator {authenticator.name} wants to authenticate {uid}/{email} "
-            "but that username is already in use by another authenticator, "
-            f"the user from this authenticator will be {new_username}"
-        )
+        logger.warning(f"User {uid} is already associated with an existing user, creating a new user with username {new_username}")
         return new_username
 
 
@@ -219,10 +211,8 @@ def _handle_email_fallback_strategy(uid: str, email: Union[str, list[str], None]
 #     else:
 #         # Some other provider is providing this username so we need to create our own username
 #         new_username = get_local_username({'username': uid})
-#         logger.info(
-#             f'Authenticator {authenticator.name} wants to authenticate {uid}/{email} but that'
-#             f' username is already in use by another authenticator,'
-#             f' the user from this authenticator will be {new_username}'
+#         logger.warning(
+#             f"User {uid} is already associated with an existing user, creating a new user with username {new_username}"
 #         )
 #         return new_username
 

test_app/tests/authentication/authenticator_plugins/test_local.py

@@ -22,6 +22,25 @@ def mock_get_setting(setting_name):
         return None
 
 
+@pytest.fixture
+def controller_auth_mocks(user):
+    """
+    Helper fixture that provides common mocks for controller authentication tests.
+    Returns a context manager that sets up the basic mocks needed for controller auth.
+    """
+    from contextlib import contextmanager
+
+    @contextmanager
+    def mock_controller_auth():
+        with (
+            mock.patch.object(user, 'use_controller_password', True, create=True),
+            mock.patch('ansible_base.authentication.authenticator_plugins.local.UserModel._default_manager.get_by_natural_key', return_value=user),
+        ):
+            yield
+
+    return mock_controller_auth
+
+
 @mock.patch("rest_framework.views.APIView.authentication_classes", [SessionAuthentication])
 def test_local_auth_successful(unauthenticated_api_client, local_authenticator, user):
     """
@@ -122,7 +141,7 @@ def test_can_authenticate_from_controller_nonexistent_user():
 
 
 @pytest.mark.django_db()
-def test_can_authenticate_from_controller_success(user):
+def test_can_authenticate_from_controller_success(user, controller_auth_mocks):
     """
     Test that _can_authenticate_from_controller returns True when all conditions are met.
     """
@@ -132,8 +151,8 @@ def test_can_authenticate_from_controller_success(user):
     user.password = "$encrypted$"
     user.save()
 
-    # Mock the controller user response - local user with encrypted password
-    with mock.patch.object(plugin, '_get_controller_user', return_value={"ldap_dn": "", "password": "$encrypted$"}):
+    # Use fixture for common mocks and add specific controller user response
+    with controller_auth_mocks(), mock.patch.object(plugin, '_get_controller_user', return_value={"ldap_dn": "", "password": "$encrypted$"}):
         result = plugin._can_authenticate_from_controller(user.username, "password")
         assert result is True
 
@@ -412,17 +431,20 @@ def test_can_authenticate_from_controller_logs_warning_invalid_format(user, expe
     """
     plugin = AuthenticatorPlugin()
 
-    # Mock the request to return invalid format (non-dict result)
-    with mock.patch('ansible_base.authentication.authenticator_plugins.local.get_setting', side_effect=mock_get_setting):
-        with mock.patch('requests.get') as mock_get:
-            mock_response = mock.Mock()
-            mock_response.raise_for_status.return_value = None
-            mock_response.json.return_value = {"count": 1, "results": ["invalid_string_not_dict"]}
-            mock_get.return_value = mock_response
+    # Mock use_controller_password to True to enable controller authentication
+    mock_response = mock.Mock()
+    mock_response.raise_for_status.return_value = None
+    mock_response.json.return_value = {"count": 1, "results": ["invalid_string_not_dict"]}
 
-            with expected_log('ansible_base.authentication.authenticator_plugins.local.logger', "warning", "user was not a dictionary"):
-                result = plugin._can_authenticate_from_controller(user.username, "password")
-                assert result is False
+    with (
+        mock.patch.object(user, 'use_controller_password', True, create=True),
+        mock.patch('ansible_base.authentication.authenticator_plugins.local.UserModel._default_manager.get_by_natural_key', return_value=user),
+        mock.patch('ansible_base.authentication.authenticator_plugins.local.get_setting', side_effect=mock_get_setting),
+        mock.patch('requests.get', return_value=mock_response),
+        expected_log('ansible_base.authentication.authenticator_plugins.local.logger', "warning", "user was not a dictionary"),
+    ):
+        result = plugin._can_authenticate_from_controller(user.username, "password")
+        assert result is False
 
 
 @pytest.mark.django_db()
@@ -436,10 +458,15 @@ def test_can_authenticate_from_controller_logs_warning_not_local_user(user, expe
     user.password = "$encrypted$"
     user.save()
 
-    with mock.patch.object(plugin, '_get_controller_user', return_value={"ldap_dn": "cn=user,dc=example,dc=com", "password": "$encrypted$"}):
-        with expected_log('ansible_base.authentication.authenticator_plugins.local.logger', "warning", "is an ldap user and can not be authenticated"):
-            result = plugin._can_authenticate_from_controller(user.username, "password")
-            assert result is False
+    # Mock use_controller_password to True to enable controller authentication
+    with (
+        mock.patch.object(user, 'use_controller_password', True, create=True),
+        mock.patch('ansible_base.authentication.authenticator_plugins.local.UserModel._default_manager.get_by_natural_key', return_value=user),
+        mock.patch.object(plugin, '_get_controller_user', return_value={"ldap_dn": "cn=user,dc=example,dc=com", "password": "$encrypted$"}),
+        expected_log('ansible_base.authentication.authenticator_plugins.local.logger', "warning", "is an ldap user and can not be authenticated"),
+    ):
+        result = plugin._can_authenticate_from_controller(user.username, "password")
+        assert result is False
 
 
 @pytest.mark.django_db()
@@ -629,23 +656,28 @@ def test_authenticate_successful_controller_validation_full_flow(user, local_aut
     user.password = "$encrypted$"
     user.save()
 
-    # Mock all the components for a successful flow
-    with (
-        mock.patch.object(plugin, '_get_controller_user', return_value={"ldap_dn": "", "password": "$encrypted$"}),
-        mock.patch('django.contrib.auth.backends.ModelBackend.authenticate') as mock_auth,
-    ):
-        mock_auth.side_effect = [None, user]  # First call fails, second succeeds after password update
+    # Mock use_controller_password to True to enable controller authentication
+    with mock.patch.object(user, 'use_controller_password', True, create=True):
+        # Mock the database lookup to return our mocked user
+        with mock.patch('ansible_base.authentication.authenticator_plugins.local.UserModel._default_manager.get_by_natural_key', return_value=user):
+            # Mock all the components for a successful flow
+            with (
+                mock.patch.object(plugin, '_get_controller_user', return_value={"ldap_dn": "", "password": "$encrypted$"}),
+                mock.patch('django.contrib.auth.backends.ModelBackend.authenticate') as mock_auth,
+                mock.patch.object(plugin, 'update_gateway_user') as mock_update,
+            ):
+                mock_auth.side_effect = [None, user]  # First call fails, second succeeds after password update
 
-        # Create request with gateway login path
-        request = RequestFactory().get('/api/gateway/v1/login/')
-        result = plugin.authenticate(request=request, username=user.username, password="password")
+                # Create request with gateway login path
+                request = RequestFactory().get('/api/gateway/v1/login/')
+                result = plugin.authenticate(request=request, username=user.username, password="password")
 
-        # Verify successful authentication
-        assert result is not None
-        assert result == user
+                # Verify successful authentication
+                assert result is not None
+                assert result == user
 
-        # Verify user was updated
-        user.refresh_from_db()
+                # Verify update_gateway_user was called
+                mock_update.assert_called_once_with(user.username, "password")
 
 
 # Test connection and timeout errors
@@ -790,11 +822,15 @@ def test_can_authenticate_from_controller_enterprise_user(user, expected_log):
     """
     plugin = AuthenticatorPlugin()
 
-    # Mock the controller user response with regular password (not encrypted)
-    with mock.patch.object(plugin, '_get_controller_user', return_value={"ldap_dn": "", "password": "regular_password"}):
-        with expected_log('ansible_base.authentication.authenticator_plugins.local.logger', "warning", "is an enterprise user and can not be authenticated"):
-            result = plugin._can_authenticate_from_controller(user.username, "password")
-            assert result is False
+    # Mock use_controller_password to True to enable controller authentication
+    with (
+        mock.patch.object(user, 'use_controller_password', True, create=True),
+        mock.patch('ansible_base.authentication.authenticator_plugins.local.UserModel._default_manager.get_by_natural_key', return_value=user),
+        mock.patch.object(plugin, '_get_controller_user', return_value={"ldap_dn": "", "password": "regular_password"}),
+        expected_log('ansible_base.authentication.authenticator_plugins.local.logger', "warning", "is an enterprise user and can not be authenticated"),
+    ):
+        result = plugin._can_authenticate_from_controller(user.username, "password")
+        assert result is False
 
 
 @pytest.mark.django_db()
@@ -804,11 +840,15 @@ def test_can_authenticate_from_controller_enterprise_user_missing_password(user,
     """
     plugin = AuthenticatorPlugin()
 
-    # Mock the controller user response without password field
-    with mock.patch.object(plugin, '_get_controller_user', return_value={"ldap_dn": "", "username": "testuser"}):
-        with expected_log('ansible_base.authentication.authenticator_plugins.local.logger', "warning", "is an enterprise user and can not be authenticated"):
-            result = plugin._can_authenticate_from_controller(user.username, "password")
-            assert result is False
+    # Mock use_controller_password to True to enable controller authentication
+    with (
+        mock.patch.object(user, 'use_controller_password', True, create=True),
+        mock.patch('ansible_base.authentication.authenticator_plugins.local.UserModel._default_manager.get_by_natural_key', return_value=user),
+        mock.patch.object(plugin, '_get_controller_user', return_value={"ldap_dn": "", "username": "testuser"}),
+        expected_log('ansible_base.authentication.authenticator_plugins.local.logger', "warning", "is an enterprise user and can not be authenticated"),
+    ):
+        result = plugin._can_authenticate_from_controller(user.username, "password")
+        assert result is False
 
 
 @pytest.mark.django_db()
@@ -818,11 +858,15 @@ def test_can_authenticate_from_controller_enterprise_user_none_password(user, ex
     """
     plugin = AuthenticatorPlugin()
 
-    # Mock the controller user response with None password
-    with mock.patch.object(plugin, '_get_controller_user', return_value={"ldap_dn": "", "password": None}):
-        with expected_log('ansible_base.authentication.authenticator_plugins.local.logger', "warning", "is an enterprise user and can not be authenticated"):
-            result = plugin._can_authenticate_from_controller(user.username, "password")
-            assert result is False
+    # Mock use_controller_password to True to enable controller authentication
+    with (
+        mock.patch.object(user, 'use_controller_password', True, create=True),
+        mock.patch('ansible_base.authentication.authenticator_plugins.local.UserModel._default_manager.get_by_natural_key', return_value=user),
+        mock.patch.object(plugin, '_get_controller_user', return_value={"ldap_dn": "", "password": None}),
+        expected_log('ansible_base.authentication.authenticator_plugins.local.logger', "warning", "is an enterprise user and can not be authenticated"),
+    ):
+        result = plugin._can_authenticate_from_controller(user.username, "password")
+        assert result is False
 
 
 # Test for timeout handling - if not timeout

test_app/tests/authentication/utils/test_authentication.py

@@ -31,7 +31,7 @@ def test_get_local_user_username_existing_user(self, random_user):
         [
             (None, 'is able to authenticate user', True),
             ('local', 'already authenticated', True),
-            ('ldap', 'username is already in use by another authenticator', False),
+            ('ldap', 'is already associated with an existing user, creating a new user with username', False),
             ('multiple', 'already authenticated', True),
         ],
     )
@@ -43,7 +43,9 @@ def test_determine_username_from_uid(
             AuthenticatorUser.objects.create(uid=random_user.username, user=random_user, provider=local_authenticator)
         elif related_authenticator in ['ldap', 'multiple']:
             AuthenticatorUser.objects.create(uid=random_user.username, user=random_user, provider=ldap_authenticator)
-        with expected_log(self.logger, 'info', info_message):
+        # 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)
             if expected_username:
                 assert new_username == random_user.username