Merge branch 'devel' into AAP-45980

awx/api/templates/instance_install_bundle/requirements.yml

@@ -1,4 +1,4 @@
 ---
 collections:
   - name: ansible.receptor
-    version: 2.0.6
+    version: 2.0.8

awx/api/views/root.py

@@ -344,13 +344,22 @@ class ApiV2ConfigView(APIView):
             become_methods=PRIVILEGE_ESCALATION_METHODS,
         )
 
-        if (
+        # Check superuser/auditor first
-            request.user.is_superuser
+        if request.user.is_superuser or request.user.is_system_auditor:
-            or request.user.is_system_auditor
+            has_org_access = True
-            or Organization.accessible_objects(request.user, 'admin_role').exists()
+        else:
-            or Organization.accessible_objects(request.user, 'auditor_role').exists()
+            # Single query checking all three organization role types at once
-            or Organization.accessible_objects(request.user, 'project_admin_role').exists()
+            has_org_access = (
-        ):
+                (
+                    Organization.access_qs(request.user, 'change')
+                    | Organization.access_qs(request.user, 'audit')
+                    | Organization.access_qs(request.user, 'add_project')
+                )
+                .distinct()
+                .exists()
+            )
+
+        if has_org_access:
             data.update(
                 dict(
                     project_base_dir=settings.PROJECTS_ROOT,
@@ -358,8 +367,10 @@ class ApiV2ConfigView(APIView):
                     custom_virtualenvs=get_custom_venv_choices(),
                 )
             )
-        elif JobTemplate.accessible_objects(request.user, 'admin_role').exists():
+        else:
-            data['custom_virtualenvs'] = get_custom_venv_choices()
+            # Only check JobTemplate access if org check failed
+            if JobTemplate.accessible_objects(request.user, 'admin_role').exists():
+                data['custom_virtualenvs'] = get_custom_venv_choices()
 
         return Response(data)
 

awx/main/tasks/jobs.py

@@ -228,16 +228,19 @@ class BaseTask(object):
         # Convert to list to prevent re-evaluation of QuerySet
         return list(credentials_list)
 
-    def populate_workload_identity_tokens(self):
+    def populate_workload_identity_tokens(self, additional_credentials=None):
         """
         Populate credentials with workload identity tokens.
 
         Sets the context on Credential objects that have input sources
         using compatible external credential types.
         """
+        credentials = list(self._credentials)
+        if additional_credentials:
+            credentials.extend(additional_credentials)
         credential_input_sources = (
             (credential.context, src)
-            for credential in self._credentials
+            for credential in credentials
             for src in credential.input_sources.all()
             if any(
                 field.get('id') == 'workload_identity_token' and field.get('internal')
@@ -1863,6 +1866,24 @@ class RunInventoryUpdate(SourceControlMixin, BaseTask):
         # All credentials not used by inventory source injector
         return inventory_update.get_extra_credentials()
 
+    def populate_workload_identity_tokens(self, additional_credentials=None):
+        """Also generate OIDC tokens for the cloud credential.
+
+        The cloud credential is not in _credentials (it is handled by the
+        inventory source injector), but it may still need a workload identity
+        token generated for it.
+        """
+        cloud_cred = self.instance.get_cloud_credential()
+        creds = list(additional_credentials or [])
+        if cloud_cred:
+            creds.append(cloud_cred)
+        super().populate_workload_identity_tokens(additional_credentials=creds or None)
+        # Override get_cloud_credential on this instance so the injector
+        # uses the credential with OIDC context instead of doing a fresh
+        # DB fetch that would lose it.
+        if cloud_cred and cloud_cred.context:
+            self.instance.get_cloud_credential = lambda: cloud_cred
+
     def build_project_dir(self, inventory_update, private_data_dir):
         source_project = None
         if inventory_update.inventory_source:

awx/main/tests/functional/api/test_config_endpoint.py

@@ -0,0 +1,84 @@
+import pytest
+from awx.api.versioning import reverse
+from rest_framework import status
+
+from awx.main.models.jobs import JobTemplate
+
+
+@pytest.mark.django_db
+class TestConfigEndpointFields:
+    def test_base_fields_all_users(self, get, rando):
+        url = reverse('api:api_v2_config_view')
+        response = get(url, rando, expect=200)
+
+        assert 'time_zone' in response.data
+        assert 'license_info' in response.data
+        assert 'version' in response.data
+        assert 'eula' in response.data
+        assert 'analytics_status' in response.data
+        assert 'analytics_collectors' in response.data
+        assert 'become_methods' in response.data
+
+    @pytest.mark.parametrize(
+        "role_type",
+        [
+            "superuser",
+            "system_auditor",
+            "org_admin",
+            "org_auditor",
+            "org_project_admin",
+        ],
+    )
+    def test_privileged_users_conditional_fields(self, get, user, organization, admin, role_type):
+        url = reverse('api:api_v2_config_view')
+
+        if role_type == "superuser":
+            test_user = admin
+        elif role_type == "system_auditor":
+            test_user = user('system-auditor', is_superuser=False)
+            test_user.is_system_auditor = True
+            test_user.save()
+        elif role_type == "org_admin":
+            test_user = user('org-admin', is_superuser=False)
+            organization.admin_role.members.add(test_user)
+        elif role_type == "org_auditor":
+            test_user = user('org-auditor', is_superuser=False)
+            organization.auditor_role.members.add(test_user)
+        elif role_type == "org_project_admin":
+            test_user = user('org-project-admin', is_superuser=False)
+            organization.project_admin_role.members.add(test_user)
+
+        response = get(url, test_user, expect=200)
+
+        assert 'project_base_dir' in response.data
+        assert 'project_local_paths' in response.data
+        assert 'custom_virtualenvs' in response.data
+
+    def test_job_template_admin_gets_venvs_only(self, get, user, organization, project, inventory):
+        """Test that JobTemplate admin without org access gets only custom_virtualenvs"""
+        jt_admin = user('jt-admin', is_superuser=False)
+
+        jt = JobTemplate.objects.create(name='test-jt', organization=organization, project=project, inventory=inventory)
+        jt.admin_role.members.add(jt_admin)
+
+        url = reverse('api:api_v2_config_view')
+        response = get(url, jt_admin, expect=200)
+
+        assert 'custom_virtualenvs' in response.data
+        assert 'project_base_dir' not in response.data
+        assert 'project_local_paths' not in response.data
+
+    def test_normal_user_no_conditional_fields(self, get, rando):
+        url = reverse('api:api_v2_config_view')
+        response = get(url, rando, expect=200)
+
+        assert 'project_base_dir' not in response.data
+        assert 'project_local_paths' not in response.data
+        assert 'custom_virtualenvs' not in response.data
+
+    def test_unauthenticated_denied(self, get):
+        """Test that unauthenticated requests are denied"""
+        url = reverse('api:api_v2_config_view')
+        response = get(url, None, expect=401)
+
+        assert response.status_code == status.HTTP_401_UNAUTHORIZED

awx/main/tests/unit/tasks/test_jobs.py

@@ -590,3 +590,67 @@ def test_populate_workload_identity_tokens_passes_get_instance_timeout_to_client
         scope=AutomationControllerJobScope.name,
         workload_ttl_seconds=expected_ttl,
     )
+
+
+class TestRunInventoryUpdatePopulateWorkloadIdentityTokens:
+    """Tests for RunInventoryUpdate.populate_workload_identity_tokens."""
+
+    def test_cloud_credential_passed_as_additional_credential(self):
+        """The cloud credential is forwarded to super().populate_workload_identity_tokens via additional_credentials."""
+        cloud_cred = mock.MagicMock(name='cloud_cred')
+        cloud_cred.context = {}
+
+        task = jobs.RunInventoryUpdate()
+        task.instance = mock.MagicMock()
+        task.instance.get_cloud_credential.return_value = cloud_cred
+        task._credentials = []
+
+        with mock.patch.object(jobs.BaseTask, 'populate_workload_identity_tokens') as mock_super:
+            task.populate_workload_identity_tokens()
+
+        mock_super.assert_called_once_with(additional_credentials=[cloud_cred])
+
+    def test_no_cloud_credential_calls_super_with_none(self):
+        """When there is no cloud credential, super() is called with additional_credentials=None."""
+        task = jobs.RunInventoryUpdate()
+        task.instance = mock.MagicMock()
+        task.instance.get_cloud_credential.return_value = None
+        task._credentials = []
+
+        with mock.patch.object(jobs.BaseTask, 'populate_workload_identity_tokens') as mock_super:
+            task.populate_workload_identity_tokens()
+
+        mock_super.assert_called_once_with(additional_credentials=None)
+
+    def test_additional_credentials_combined_with_cloud_credential(self):
+        """Caller-supplied additional_credentials are combined with the cloud credential."""
+        cloud_cred = mock.MagicMock(name='cloud_cred')
+        cloud_cred.context = {}
+        extra_cred = mock.MagicMock(name='extra_cred')
+
+        task = jobs.RunInventoryUpdate()
+        task.instance = mock.MagicMock()
+        task.instance.get_cloud_credential.return_value = cloud_cred
+        task._credentials = []
+
+        with mock.patch.object(jobs.BaseTask, 'populate_workload_identity_tokens') as mock_super:
+            task.populate_workload_identity_tokens(additional_credentials=[extra_cred])
+
+        mock_super.assert_called_once_with(additional_credentials=[extra_cred, cloud_cred])
+
+    def test_cloud_credential_override_after_context_set(self):
+        """After OIDC processing, get_cloud_credential is overridden on the instance when context is populated."""
+        cloud_cred = mock.MagicMock(name='cloud_cred')
+        # Simulate that super().populate_workload_identity_tokens populates context
+        cloud_cred.context = {'workload_identity_token': 'eyJ.test.jwt'}
+
+        task = jobs.RunInventoryUpdate()
+        task.instance = mock.MagicMock()
+        task.instance.get_cloud_credential.return_value = cloud_cred
+        task._credentials = []
+
+        with mock.patch.object(jobs.BaseTask, 'populate_workload_identity_tokens'):
+            task.populate_workload_identity_tokens()
+
+        # The instance's get_cloud_credential should now return the same object with context
+        assert task.instance.get_cloud_credential() is cloud_cred