feat: workload identity credentials integration (#16286)

* feat: workload identity credentials integration

* feat: cache credentials and add context property to Credential

Assisted-by: Claude

* feat: include safeguard in case feature flag is disabled

* feat: tests to validate workload identity credentials integration

* fix: affected tests by the credential cache mechanism

* feat: remove word cache from variables and comments, use standard library decorators

* fix: reorder tests in correct files

* Use better error catching mechanisms

* Adjust logic to support multiple credential input sources and use internal field

* Remove hardcoded credential type names

* Add tests for the internal field

Assited-by: Claude

awx/main/fields.py

@@ -428,6 +428,9 @@ class CredentialInputField(JSONSchemaField):
         # determine the defined fields for the associated credential type
         properties = {}
         for field in model_instance.credential_type.inputs.get('fields', []):
+            # Prevent users from providing values for internally resolved fields
+            if 'internal' in field:
+                continue
             field = field.copy()
             properties[field['id']] = field
             if field.get('choices', []):
@@ -566,6 +569,7 @@ class CredentialTypeInputField(JSONSchemaField):
                             },
                             'label': {'type': 'string'},
                             'help_text': {'type': 'string'},
+                            'internal': {'type': 'boolean'},
                             'multiline': {'type': 'boolean'},
                             'secret': {'type': 'boolean'},
                             'ask_at_runtime': {'type': 'boolean'},

awx/main/models/credential.py

@@ -242,6 +242,29 @@ class Credential(PasswordFieldsModel, CommonModelNameNotUnique, ResourceMixin):
                 needed.append('vault_password')
         return needed
 
+    @functools.cached_property
+    def context(self):
+        """
+        Property for storing runtime context during credential resolution.
+
+        The context is a dict keyed by CredentialInputSource PK, where each value
+        is a dict of runtime fields for that input source. Example::
+
+            {
+                <input_source_pk>: {
+                    "workload_identity_token": "<jwt_token>"
+                },
+                <another_input_source_pk>: {
+                    "workload_identity_token": "<different_jwt_token>"
+                },
+            }
+
+        This structure allows each input source to have its own set of runtime
+        values, avoiding conflicts when a credential has multiple input sources
+        with different configurations (e.g., different JWT audiences).
+        """
+        return {}
+
     @cached_property
     def dynamic_input_fields(self):
         # if the credential is not yet saved we can't access the input_sources
@@ -367,7 +390,7 @@ class Credential(PasswordFieldsModel, CommonModelNameNotUnique, ResourceMixin):
     def _get_dynamic_input(self, field_name):
         for input_source in self.input_sources.all():
             if input_source.input_field_name == field_name:
-                return input_source.get_input_value()
+                return input_source.get_input_value(context=self.context)
         else:
             raise ValueError('{} is not a dynamic input field'.format(field_name))
 
@@ -622,7 +645,15 @@ class CredentialInputSource(PrimordialModel):
             raise ValidationError(_('Input field must be defined on target credential (options are {}).'.format(', '.join(sorted(defined_fields)))))
         return self.input_field_name
 
-    def get_input_value(self):
+    def get_input_value(self, context: dict | None = None):
+        """
+        Retrieve the value from the external credential backend.
+
+        Args:
+            context: Optional runtime context dict passed from the target credential.
+        """
+        if context is None:
+            context = {}
         backend = self.source_credential.credential_type.plugin.backend
         backend_kwargs = {}
         for field_name, value in self.source_credential.inputs.items():
@@ -633,6 +664,17 @@ class CredentialInputSource(PrimordialModel):
 
         backend_kwargs.update(self.metadata)
 
+        # Resolve internal fields from the per-input-source context.
+        # The context dict is keyed by input source PK, e.g.:
+        #   {42: {"workload_identity_token": "eyJ..."}, 43: {"workload_identity_token": "eyX..."}}
+        # This allows each input source to carry its own runtime values.
+        input_source_context = context.get(self.pk, {})
+        for field in self.source_credential.credential_type.inputs.get('fields', []):
+            if field.get('internal'):
+                value = input_source_context.get(field['id'])
+                if value is not None:
+                    backend_kwargs[field['id']] = value
+
         with set_environ(**settings.AWX_TASK_ENV):
             return backend(**backend_kwargs)
 

awx/main/tasks/jobs.py

@@ -219,6 +219,55 @@ class BaseTask(object):
         self.update_attempts = int(getattr(settings, 'DISPATCHER_DB_DOWNTOWN_TOLLERANCE', settings.DISPATCHER_DB_DOWNTIME_TOLERANCE) / 5)
         self.runner_callback = self.callback_class(model=self.model)
 
+    @functools.cached_property
+    def _credentials(self):
+        """
+        Credentials for the task execution.
+        Fetches credentials once using build_credentials_list() and stores
+        them for the duration of the task to avoid redundant database queries.
+        """
+        credentials_list = self.build_credentials_list(self.instance)
+        # Convert to list to prevent re-evaluation of QuerySet
+        return list(credentials_list)
+
+    def populate_workload_identity_tokens(self):
+        """
+        Populate credentials with workload identity tokens.
+
+        Sets the context on Credential objects that have input sources
+        using compatible external credential types.
+        """
+        credential_input_sources = (
+            (credential.context, src)
+            for credential in self._credentials
+            for src in credential.input_sources.all()
+            if any(
+                field.get('id') == 'workload_identity_token' and field.get('internal')
+                for field in src.source_credential.credential_type.inputs.get('fields', [])
+            )
+        )
+        for credential_ctx, input_src in credential_input_sources:
+            if flag_enabled("FEATURE_OIDC_WORKLOAD_IDENTITY_ENABLED"):
+                try:
+                    jwt = retrieve_workload_identity_jwt(
+                        self.instance, audience=input_src.source_credential.get_input('jwt_aud'), scope=AutomationControllerJobScope.name
+                    )
+                    # Store token keyed by input source PK, since a credential can have
+                    # multiple input sources (one per field), each potentially with a different audience
+                    credential_ctx[input_src.pk] = {"workload_identity_token": jwt}
+                except Exception as e:
+                    self.instance.job_explanation = (
+                        f'Could not generate workload identity token for credential {input_src.source_credential.name} used in this job. Error:\n{e}'
+                    )
+                    self.instance.status = 'error'
+                    self.instance.save()
+            else:
+                self.instance.job_explanation = (
+                    f'Flag FEATURE_OIDC_WORKLOAD_IDENTITY_ENABLED is not enabled, required for credential {input_src.source_credential.name} used in this job.'
+                )
+                self.instance.status = 'error'
+                self.instance.save()
+
     def update_model(self, pk, _attempt=0, **updates):
         return update_model(self.model, pk, _attempt=0, _max_attempts=self.update_attempts, **updates)
 
@@ -615,6 +664,12 @@ class BaseTask(object):
             if not os.path.exists(settings.AWX_ISOLATION_BASE_PATH):
                 raise RuntimeError('AWX_ISOLATION_BASE_PATH=%s does not exist' % settings.AWX_ISOLATION_BASE_PATH)
 
+            if flag_enabled("FEATURE_OIDC_WORKLOAD_IDENTITY_ENABLED"):
+                logger.info(f'Generating workload identity tokens for job {self.instance.id}')
+                self.populate_workload_identity_tokens()
+                if self.instance.status == 'error':
+                    raise RuntimeError('not starting %s task' % self.instance.status)
+
             # May have to serialize the value
             private_data_files, ssh_key_data = self.build_private_data_files(self.instance, private_data_dir)
             passwords = self.build_passwords(self.instance, kwargs)
@@ -632,7 +687,7 @@ class BaseTask(object):
 
             self.runner_callback.job_created = str(self.instance.created)
 
-            credentials = self.build_credentials_list(self.instance)
+            credentials = self._credentials
 
             container_root = None
             if settings.IS_K8S and isinstance(self.instance, ProjectUpdate):
@@ -927,6 +982,29 @@ class RunJob(SourceControlMixin, BaseTask):
     model = Job
     event_model = JobEvent
 
+    def _extract_credentials_of_kind(self, kind: str):
+        return (cred for cred in self._credentials if cred.credential_type.kind == kind)
+
+    @property
+    def _machine_credential(self) -> object:
+        """Get machine credential."""
+        return next(self._extract_credentials_of_kind('ssh'), None)
+
+    @property
+    def _vault_credentials(self) -> list[object]:
+        """Get vault credentials."""
+        return list(self._extract_credentials_of_kind('vault'))
+
+    @property
+    def _network_credentials(self) -> list[object]:
+        """Get network credentials."""
+        return list(self._extract_credentials_of_kind('net'))
+
+    @property
+    def _cloud_credentials(self) -> list[object]:
+        """Get cloud credentials."""
+        return list(self._extract_credentials_of_kind('cloud'))
+
     def build_private_data(self, job, private_data_dir):
         """
         Returns a dict of the form
@@ -944,7 +1022,7 @@ class RunJob(SourceControlMixin, BaseTask):
         }
         """
         private_data = {'credentials': {}}
-        for credential in job.credentials.prefetch_related('input_sources__source_credential').all():
+        for credential in self._credentials:
             # If we were sent SSH credentials, decrypt them and send them
             # back (they will be written to a temporary file).
             if credential.has_input('ssh_key_data'):
@@ -960,14 +1038,14 @@ class RunJob(SourceControlMixin, BaseTask):
         and ansible-vault.
         """
         passwords = super(RunJob, self).build_passwords(job, runtime_passwords)
-        cred = job.machine_credential
+        cred = self._machine_credential
         if cred:
             for field in ('ssh_key_unlock', 'ssh_password', 'become_password', 'vault_password'):
                 value = runtime_passwords.get(field, cred.get_input('password' if field == 'ssh_password' else field, default=''))
                 if value not in ('', 'ASK'):
                     passwords[field] = value
 
-        for cred in job.vault_credentials:
+        for cred in self._vault_credentials:
             field = 'vault_password'
             vault_id = cred.get_input('vault_id', default=None)
             if vault_id:
@@ -983,7 +1061,7 @@ class RunJob(SourceControlMixin, BaseTask):
         key unlock over network key unlock.
         '''
         if 'ssh_key_unlock' not in passwords:
-            for cred in job.network_credentials:
+            for cred in self._network_credentials:
                 if cred.inputs.get('ssh_key_unlock'):
                     passwords['ssh_key_unlock'] = runtime_passwords.get('ssh_key_unlock', cred.get_input('ssh_key_unlock', default=''))
                     break
@@ -1018,11 +1096,11 @@ class RunJob(SourceControlMixin, BaseTask):
 
         # Set environment variables for cloud credentials.
         cred_files = private_data_files.get('credentials', {})
-        for cloud_cred in job.cloud_credentials:
+        for cloud_cred in self._cloud_credentials:
             if cloud_cred and cloud_cred.credential_type.namespace == 'openstack' and cred_files.get(cloud_cred, ''):
                 env['OS_CLIENT_CONFIG_FILE'] = get_incontainer_path(cred_files.get(cloud_cred, ''), private_data_dir)
 
-        for network_cred in job.network_credentials:
+        for network_cred in self._network_credentials:
             env['ANSIBLE_NET_USERNAME'] = network_cred.get_input('username', default='')
             env['ANSIBLE_NET_PASSWORD'] = network_cred.get_input('password', default='')
 
@@ -1072,7 +1150,7 @@ class RunJob(SourceControlMixin, BaseTask):
         Build command line argument list for running ansible-playbook,
         optionally using ssh-agent for public/private key authentication.
         """
-        creds = job.machine_credential
+        creds = self._machine_credential
 
         ssh_username, become_username, become_method = '', '', ''
         if creds:

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

@@ -1,5 +1,7 @@
 import pytest
 
+from ansible_base.lib.testing.util import feature_flag_enabled, feature_flag_disabled
+
 from awx.main.models import CredentialInputSource
 from awx.api.versioning import reverse
 
@@ -316,3 +318,60 @@ def test_create_credential_input_source_with_already_used_input_returns_400(post
     ]
     all_responses = [post(list_url, params, admin) for params in all_params]
     assert all_responses.pop().status_code == 400
+
+
+@pytest.mark.django_db
+def test_credential_input_source_passes_workload_identity_token_when_flag_enabled(vault_credential, external_credential, mocker):
+    """Test that workload_identity_token is passed to backend when flag is enabled."""
+    with feature_flag_enabled('FEATURE_OIDC_WORKLOAD_IDENTITY_ENABLED'):
+        # Add workload_identity_token as an internal field on the external credential type
+        # so get_input_value resolves it from the per-input-source context
+        external_credential.credential_type.inputs['fields'].append(
+            {'id': 'workload_identity_token', 'label': 'Workload Identity Token', 'type': 'string', 'internal': True}
+        )
+
+        # Create an input source
+        input_source = CredentialInputSource.objects.create(
+            target_credential=vault_credential,
+            source_credential=external_credential,
+            input_field_name='vault_password',
+            metadata={'key': 'test_key'},
+        )
+
+        # Mock the credential plugin backend
+        mock_backend = mocker.patch.object(external_credential.credential_type.plugin, 'backend', autospec=True, return_value='test_value')
+
+        # Call with context keyed by input source PK
+        test_context = {input_source.pk: {'workload_identity_token': 'jwt_token_here'}}
+        result = input_source.get_input_value(context=test_context)
+
+        # Verify backend was called with workload_identity_token
+        assert result == 'test_value'
+        call_kwargs = mock_backend.call_args[1]
+        assert call_kwargs['workload_identity_token'] == 'jwt_token_here'
+        assert call_kwargs['key'] == 'test_key'
+
+
+@pytest.mark.django_db
+def test_credential_input_source_skips_workload_identity_token_when_flag_disabled(vault_credential, external_credential, mocker):
+    """Test that workload_identity_token is NOT passed when flag is disabled."""
+    with feature_flag_disabled('FEATURE_OIDC_WORKLOAD_IDENTITY_ENABLED'):
+        # Create an input source
+        input_source = CredentialInputSource.objects.create(
+            target_credential=vault_credential,
+            source_credential=external_credential,
+            input_field_name='vault_password',
+            metadata={'key': 'test_key'},
+        )
+        # Mock the credential plugin backend
+        mock_backend = mocker.patch.object(external_credential.credential_type.plugin, 'backend', autospec=True, return_value='test_value')
+        # Call with context containing workload_identity_token but NO internal field defined,
+        # simulating a flag-disabled scenario where tokens are not generated upstream
+        test_context = {input_source.pk: {'workload_identity_token': 'jwt_token_here'}}
+        result = input_source.get_input_value(context=test_context)
+        # Verify backend was called WITHOUT workload_identity_token since the credential type
+        # does not define it as an internal field (flag-disabled path doesn't register it)
+        assert result == 'test_value'
+        call_kwargs = mock_backend.call_args[1]
+        assert 'workload_identity_token' not in call_kwargs
+        assert call_kwargs['key'] == 'test_key'

awx/main/tests/functional/test_credential.py

@@ -93,6 +93,8 @@ def test_default_cred_types():
             'gpg_public_key',
             'hashivault_kv',
             'hashivault_ssh',
+            'hashivault-kv-oidc',
+            'hashivault-ssh-oidc',
             'hcp_terraform',
             'insights',
             'kubernetes_bearer_token',

awx/main/tests/functional/test_jobs.py

@@ -8,6 +8,7 @@ from awx.main.models import (
     Instance,
     Host,
     JobHostSummary,
+    Inventory,
     InventoryUpdate,
     InventorySource,
     Project,
@@ -17,14 +18,60 @@ from awx.main.models import (
     InstanceGroup,
     Label,
     ExecutionEnvironment,
+    Credential,
+    CredentialType,
+    CredentialInputSource,
+    Organization,
+    JobTemplate,
 )
+from awx.main.tasks import jobs
 from awx.main.tasks.system import cluster_node_heartbeat
 from awx.main.utils.db import bulk_update_sorted_by_id
+from ansible_base.lib.testing.util import feature_flag_enabled, feature_flag_disabled
 
 from django.db import OperationalError
 from django.test.utils import override_settings
 
 
+@pytest.fixture
+def job_template_with_credentials():
+    """
+    Factory fixture that creates a job template with specified credentials.
+
+    Usage:
+        job = job_template_with_credentials(ssh_cred, vault_cred)
+    """
+
+    def _create_job_template(
+        *credentials, org_name='test-org', project_name='test-project', inventory_name='test-inventory', jt_name='test-jt', playbook='test.yml'
+    ):
+        """
+        Create a job template with the given credentials.
+
+        Args:
+            *credentials: Variable number of Credential objects to attach to the job template
+            org_name: Name for the organization
+            project_name: Name for the project
+            inventory_name: Name for the inventory
+            jt_name: Name for the job template
+            playbook: Playbook filename
+
+        Returns:
+            Job instance created from the job template
+        """
+        org = Organization.objects.create(name=org_name)
+        proj = Project.objects.create(name=project_name, organization=org)
+        inv = Inventory.objects.create(name=inventory_name, organization=org)
+        jt = JobTemplate.objects.create(name=jt_name, project=proj, inventory=inv, playbook=playbook)
+
+        if credentials:
+            jt.credentials.add(*credentials)
+
+        return jt.create_unified_job()
+
+    return _create_job_template
+
+
 @pytest.mark.django_db
 def test_orphan_unified_job_creation(instance, inventory):
     job = Job.objects.create(job_template=None, inventory=inventory, name='hi world')
@@ -262,3 +309,393 @@ class TestLaunchConfig:
         assert config.execution_environment
         # We just write the PK instead of trying to assign an item, that happens on the save
         assert config.execution_environment_id == ee.id
+
+
+@pytest.mark.django_db
+def test_base_task_credentials_property(job_template_with_credentials):
+    """Test that _credentials property caches credentials and doesn't re-query."""
+    task = jobs.RunJob()
+
+    # Create real credentials
+    ssh_type = CredentialType.defaults['ssh']()
+    ssh_type.save()
+    vault_type = CredentialType.defaults['vault']()
+    vault_type.save()
+
+    ssh_cred = Credential.objects.create(credential_type=ssh_type, name='ssh-cred')
+    vault_cred = Credential.objects.create(credential_type=vault_type, name='vault-cred')
+
+    # Create a job with credentials using fixture
+    job = job_template_with_credentials(ssh_cred, vault_cred)
+    task.instance = job
+
+    # First access should build credentials
+    result1 = task._credentials
+    assert len(result1) == 2
+    assert isinstance(result1, list)
+
+    # Second access should return cached value (we can verify by checking it's the same list object)
+    result2 = task._credentials
+    assert result2 is result1  # Same object reference
+
+
+@pytest.mark.django_db
+def test_run_job_machine_credential(job_template_with_credentials):
+    """Test _machine_credential returns ssh credential from cache."""
+    task = jobs.RunJob()
+
+    # Create credentials
+    ssh_type = CredentialType.defaults['ssh']()
+    ssh_type.save()
+    vault_type = CredentialType.defaults['vault']()
+    vault_type.save()
+
+    ssh_cred = Credential.objects.create(credential_type=ssh_type, name='ssh-cred')
+    vault_cred = Credential.objects.create(credential_type=vault_type, name='vault-cred')
+
+    # Create a job using fixture
+    job = job_template_with_credentials(ssh_cred, vault_cred)
+    task.instance = job
+
+    # Set cached credentials
+    task._credentials = [ssh_cred, vault_cred]
+
+    # Get machine credential
+    result = task._machine_credential
+    assert result == ssh_cred
+    assert result.credential_type.kind == 'ssh'
+
+
+@pytest.mark.django_db
+def test_run_job_machine_credential_none(job_template_with_credentials):
+    """Test _machine_credential returns None when no ssh credential exists."""
+    task = jobs.RunJob()
+
+    # Create only vault credential
+    vault_type = CredentialType.defaults['vault']()
+    vault_type.save()
+    vault_cred = Credential.objects.create(credential_type=vault_type, name='vault-cred')
+
+    job = job_template_with_credentials(vault_cred)
+    task.instance = job
+
+    # Set cached credentials
+    task._credentials = [vault_cred]
+
+    # Get machine credential
+    result = task._machine_credential
+    assert result is None
+
+
+@pytest.mark.django_db
+def test_run_job_vault_credentials(job_template_with_credentials):
+    """Test _vault_credentials returns all vault credentials from cache."""
+    task = jobs.RunJob()
+
+    # Create credentials
+    vault_type = CredentialType.defaults['vault']()
+    vault_type.save()
+    ssh_type = CredentialType.defaults['ssh']()
+    ssh_type.save()
+
+    vault_cred1 = Credential.objects.create(credential_type=vault_type, name='vault-1')
+    vault_cred2 = Credential.objects.create(credential_type=vault_type, name='vault-2')
+    ssh_cred = Credential.objects.create(credential_type=ssh_type, name='ssh-cred')
+
+    job = job_template_with_credentials(vault_cred1, ssh_cred, vault_cred2)
+    task.instance = job
+
+    # Set cached credentials
+    task._credentials = [vault_cred1, ssh_cred, vault_cred2]
+
+    # Get vault credentials
+    result = task._vault_credentials
+    assert len(result) == 2
+    assert vault_cred1 in result
+    assert vault_cred2 in result
+    assert ssh_cred not in result
+
+
+@pytest.mark.django_db
+def test_run_job_network_credentials(job_template_with_credentials):
+    """Test _network_credentials returns all network credentials from cache."""
+    task = jobs.RunJob()
+
+    # Create credentials
+    net_type = CredentialType.defaults['net']()
+    net_type.save()
+    ssh_type = CredentialType.defaults['ssh']()
+    ssh_type.save()
+
+    net_cred = Credential.objects.create(credential_type=net_type, name='net-cred')
+    ssh_cred = Credential.objects.create(credential_type=ssh_type, name='ssh-cred')
+
+    job = job_template_with_credentials(net_cred, ssh_cred)
+    task.instance = job
+
+    # Set cached credentials
+    task._credentials = [net_cred, ssh_cred]
+
+    # Get network credentials
+    result = task._network_credentials
+    assert len(result) == 1
+    assert result[0] == net_cred
+
+
+@pytest.mark.django_db
+def test_run_job_cloud_credentials(job_template_with_credentials):
+    """Test _cloud_credentials returns all cloud credentials from cache."""
+    task = jobs.RunJob()
+
+    # Create credentials
+    aws_type = CredentialType.defaults['aws']()
+    aws_type.save()
+    ssh_type = CredentialType.defaults['ssh']()
+    ssh_type.save()
+
+    aws_cred = Credential.objects.create(credential_type=aws_type, name='aws-cred')
+    ssh_cred = Credential.objects.create(credential_type=ssh_type, name='ssh-cred')
+
+    job = job_template_with_credentials(aws_cred, ssh_cred)
+    task.instance = job
+
+    # Set cached credentials
+    task._credentials = [aws_cred, ssh_cred]
+
+    # Get cloud credentials
+    result = task._cloud_credentials
+    assert len(result) == 1
+    assert result[0] == aws_cred
+
+
+@pytest.mark.django_db
+@override_settings(RESOURCE_SERVER={'URL': 'https://gateway.example.com', 'SECRET_KEY': 'test-secret-key', 'VALIDATE_HTTPS': False})
+def test_populate_workload_identity_tokens_with_flag_enabled(job_template_with_credentials, mocker):
+    """Test populate_workload_identity_tokens sets context when flag is enabled."""
+    with feature_flag_enabled('FEATURE_OIDC_WORKLOAD_IDENTITY_ENABLED'):
+        task = jobs.RunJob()
+
+        # Create credential types
+        ssh_type = CredentialType.defaults['ssh']()
+        ssh_type.save()
+
+        # Create a workload identity credential type
+        hashivault_type = CredentialType(
+            name='HashiCorp Vault Secret Lookup (OIDC)',
+            kind='cloud',
+            managed=False,
+            inputs={
+                'fields': [
+                    {'id': 'jwt_aud', 'type': 'string', 'label': 'JWT Audience'},
+                    {'id': 'workload_identity_token', 'type': 'string', 'label': 'Workload Identity Token', 'secret': True, 'internal': True},
+                ]
+            },
+        )
+        hashivault_type.save()
+
+        # Create credentials
+        ssh_cred = Credential.objects.create(credential_type=ssh_type, name='ssh-cred')
+        source_cred = Credential.objects.create(credential_type=hashivault_type, name='vault-source', inputs={'jwt_aud': 'https://vault.example.com'})
+        target_cred = Credential.objects.create(credential_type=ssh_type, name='target-cred', inputs={'username': 'testuser'})
+
+        # Create input source linking source credential to target credential
+        input_source = CredentialInputSource.objects.create(
+            target_credential=target_cred, source_credential=source_cred, input_field_name='password', metadata={'path': 'secret/data/password'}
+        )
+
+        # Create a job using fixture
+        job = job_template_with_credentials(target_cred, ssh_cred)
+        task.instance = job
+
+        # Override cached_property so the loop uses these exact Python objects
+        task._credentials = [target_cred, ssh_cred]
+
+        # Mock only the HTTP response from the Gateway workload identity endpoint
+        mock_response = mocker.Mock(status_code=200)
+        mock_response.json.return_value = {'jwt': 'eyJ.test.jwt'}
+
+        mock_request = mocker.patch('requests.request', return_value=mock_response, autospec=True)
+
+        task.populate_workload_identity_tokens()
+
+        # Verify the HTTP call was made to the correct endpoint
+        mock_request.assert_called_once()
+        call_kwargs = mock_request.call_args.kwargs
+        assert call_kwargs['method'] == 'POST'
+        assert '/api/gateway/v1/workload_identity_tokens' in call_kwargs['url']
+
+        # Verify context was set on the credential, keyed by input source PK
+        assert input_source.pk in target_cred.context
+        assert target_cred.context[input_source.pk]['workload_identity_token'] == 'eyJ.test.jwt'
+
+
+@pytest.mark.django_db
+def test_populate_workload_identity_tokens_with_flag_disabled(job_template_with_credentials):
+    """Test populate_workload_identity_tokens sets error status when flag is disabled."""
+    with feature_flag_disabled('FEATURE_OIDC_WORKLOAD_IDENTITY_ENABLED'):
+        task = jobs.RunJob()
+
+        # Create credential types
+        ssh_type = CredentialType.defaults['ssh']()
+        ssh_type.save()
+
+        # Create a workload identity credential type
+        hashivault_type = CredentialType(
+            name='HashiCorp Vault Secret Lookup (OIDC)',
+            kind='cloud',
+            managed=False,
+            inputs={
+                'fields': [
+                    {'id': 'jwt_aud', 'type': 'string', 'label': 'JWT Audience'},
+                    {'id': 'workload_identity_token', 'type': 'string', 'label': 'Workload Identity Token', 'secret': True, 'internal': True},
+                ]
+            },
+        )
+        hashivault_type.save()
+
+        # Create credentials
+        source_cred = Credential.objects.create(credential_type=hashivault_type, name='vault-source')
+        target_cred = Credential.objects.create(credential_type=ssh_type, name='target-cred', inputs={'username': 'testuser'})
+
+        # Create input source linking source credential to target credential
+        # Note: Creates the relationship that will trigger the feature flag check
+        CredentialInputSource.objects.create(
+            target_credential=target_cred, source_credential=source_cred, input_field_name='password', metadata={'path': 'secret/data/password'}
+        )
+
+        # Create a job using fixture
+        job = job_template_with_credentials(target_cred)
+        task.instance = job
+
+        # Set cached credentials
+        task._credentials = [target_cred]
+
+        task.populate_workload_identity_tokens()
+
+        # Verify job status was set to error
+        job.refresh_from_db()
+        assert job.status == 'error'
+        assert 'FEATURE_OIDC_WORKLOAD_IDENTITY_ENABLED' in job.job_explanation
+        assert 'vault-source' in job.job_explanation
+
+
+@pytest.mark.django_db
+@override_settings(RESOURCE_SERVER={'URL': 'https://gateway.example.com', 'SECRET_KEY': 'test-secret-key', 'VALIDATE_HTTPS': False})
+def test_populate_workload_identity_tokens_multiple_input_sources_per_credential(job_template_with_credentials, mocker):
+    """Test that a single credential with two input sources from different workload identity
+    credential types gets a separate JWT token for each input source, keyed by input source PK."""
+    with feature_flag_enabled('FEATURE_OIDC_WORKLOAD_IDENTITY_ENABLED'):
+        task = jobs.RunJob()
+
+        # Create credential types
+        ssh_type = CredentialType.defaults['ssh']()
+        ssh_type.save()
+
+        # Create two different workload identity credential types
+        hashivault_kv_type = CredentialType(
+            name='HashiCorp Vault Secret Lookup (OIDC)',
+            kind='cloud',
+            managed=False,
+            inputs={
+                'fields': [
+                    {'id': 'jwt_aud', 'type': 'string', 'label': 'JWT Audience'},
+                    {'id': 'workload_identity_token', 'type': 'string', 'label': 'Workload Identity Token', 'secret': True, 'internal': True},
+                ]
+            },
+        )
+        hashivault_kv_type.save()
+
+        hashivault_ssh_type = CredentialType(
+            name='HashiCorp Vault Signed SSH (OIDC)',
+            kind='cloud',
+            managed=False,
+            inputs={
+                'fields': [
+                    {'id': 'jwt_aud', 'type': 'string', 'label': 'JWT Audience'},
+                    {'id': 'workload_identity_token', 'type': 'string', 'label': 'Workload Identity Token', 'secret': True, 'internal': True},
+                ]
+            },
+        )
+        hashivault_ssh_type.save()
+
+        # Create source credentials with different audiences
+        source_cred_kv = Credential.objects.create(
+            credential_type=hashivault_kv_type, name='vault-kv-source', inputs={'jwt_aud': 'https://vault-kv.example.com'}
+        )
+        source_cred_ssh = Credential.objects.create(
+            credential_type=hashivault_ssh_type, name='vault-ssh-source', inputs={'jwt_aud': 'https://vault-ssh.example.com'}
+        )
+
+        # Create target credential that uses both sources for different fields
+        target_cred = Credential.objects.create(credential_type=ssh_type, name='target-cred', inputs={'username': 'testuser'})
+
+        # Create two input sources on the same target credential, each for a different field
+        input_source_password = CredentialInputSource.objects.create(
+            target_credential=target_cred, source_credential=source_cred_kv, input_field_name='password', metadata={'path': 'secret/data/password'}
+        )
+        input_source_ssh_key = CredentialInputSource.objects.create(
+            target_credential=target_cred, source_credential=source_cred_ssh, input_field_name='ssh_key_data', metadata={'path': 'secret/data/ssh_key'}
+        )
+
+        # Create a job using fixture
+        job = job_template_with_credentials(target_cred)
+        task.instance = job
+
+        # Override cached_property so the loop uses this exact Python object
+        task._credentials = [target_cred]
+
+        # Mock HTTP responses - return different JWTs for each call
+        response_kv = mocker.Mock(status_code=200)
+        response_kv.json.return_value = {'jwt': 'eyJ.kv.jwt'}
+
+        response_ssh = mocker.Mock(status_code=200)
+        response_ssh.json.return_value = {'jwt': 'eyJ.ssh.jwt'}
+
+        mock_request = mocker.patch('requests.request', side_effect=[response_kv, response_ssh], autospec=True)
+
+        task.populate_workload_identity_tokens()
+
+        # Verify two separate HTTP calls were made (one per input source)
+        assert mock_request.call_count == 2
+
+        # Verify each call used the correct audience from its source credential
+        audiences_requested = {call.kwargs.get('json', {}).get('audience', '') for call in mock_request.call_args_list}
+        assert 'https://vault-kv.example.com' in audiences_requested
+        assert 'https://vault-ssh.example.com' in audiences_requested
+
+        # Verify context on the target credential has both tokens, keyed by input source PK
+        assert input_source_password.pk in target_cred.context
+        assert input_source_ssh_key.pk in target_cred.context
+        assert target_cred.context[input_source_password.pk]['workload_identity_token'] == 'eyJ.kv.jwt'
+        assert target_cred.context[input_source_ssh_key.pk]['workload_identity_token'] == 'eyJ.ssh.jwt'
+
+
+@pytest.mark.django_db
+def test_populate_workload_identity_tokens_without_workload_identity_credentials(job_template_with_credentials, mocker):
+    """Test populate_workload_identity_tokens does nothing when no workload identity credentials."""
+    with feature_flag_enabled('FEATURE_OIDC_WORKLOAD_IDENTITY_ENABLED'):
+        task = jobs.RunJob()
+
+        # Create only standard credentials (no workload identity)
+        ssh_type = CredentialType.defaults['ssh']()
+        ssh_type.save()
+        vault_type = CredentialType.defaults['vault']()
+        vault_type.save()
+
+        ssh_cred = Credential.objects.create(credential_type=ssh_type, name='ssh-cred')
+        vault_cred = Credential.objects.create(credential_type=vault_type, name='vault-cred')
+
+        # Create a job using fixture
+        job = job_template_with_credentials(ssh_cred, vault_cred)
+        task.instance = job
+
+        # Set cached credentials
+        task._credentials = [ssh_cred, vault_cred]
+
+        mocker.patch('awx.main.tasks.jobs.populate_claims_for_workload', return_value={'job_id': 123}, autospec=True)
+
+        task.populate_workload_identity_tokens()
+
+        # Verify no context was set
+        assert not hasattr(ssh_cred, '_context') or ssh_cred.context == {}
+        assert not hasattr(vault_cred, '_context') or vault_cred.context == {}

awx/main/tests/unit/models/test_credential.py

@@ -47,3 +47,34 @@ def test__get_credential_type_class_invalid_params():
 
     assert type(e.value) is ValueError
     assert str(e.value) == 'Expected only apps or app_config to be defined, not both'
+
+
+def test_credential_context_property():
+    """Test that credential context property initializes empty dict and persists across accesses."""
+    ct = CredentialType(name='Test Cred', kind='vault')
+    cred = Credential(id=1, name='Test Credential', credential_type=ct, inputs={})
+
+    # First access should return empty dict
+    context = cred.context
+    assert context == {}
+
+    # Modify the context
+    context['test_key'] = 'test_value'
+
+    # Second access should return the same dict with modifications
+    assert cred.context == {'test_key': 'test_value'}
+    assert cred.context is context  # Same object reference
+
+
+def test_credential_context_property_independent_instances():
+    """Test that context property is independent between credential instances."""
+    ct = CredentialType(name='Test Cred', kind='vault')
+    cred1 = Credential(id=1, name='Cred 1', credential_type=ct, inputs={})
+    cred2 = Credential(id=2, name='Cred 2', credential_type=ct, inputs={})
+
+    cred1.context['key1'] = 'value1'
+    cred2.context['key2'] = 'value2'
+
+    assert cred1.context == {'key1': 'value1'}
+    assert cred2.context == {'key2': 'value2'}
+    assert cred1.context is not cred2.context

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

@@ -41,6 +41,45 @@ def private_data_dir():
     shutil.rmtree(private_data, True)
 
 
+@pytest.fixture
+def job_template_with_credentials():
+    """
+    Factory fixture that creates a job template with specified credentials.
+
+    Usage:
+        job = job_template_with_credentials(ssh_cred, vault_cred)
+    """
+
+    def _create_job_template(
+        *credentials, org_name='test-org', project_name='test-project', inventory_name='test-inventory', jt_name='test-jt', playbook='test.yml'
+    ):
+        """
+        Create a job template with the given credentials.
+
+        Args:
+            *credentials: Variable number of Credential objects to attach to the job template
+            org_name: Name for the organization
+            project_name: Name for the project
+            inventory_name: Name for the inventory
+            jt_name: Name for the job template
+            playbook: Playbook filename
+
+        Returns:
+            Job instance created from the job template
+        """
+        org = Organization.objects.create(name=org_name)
+        proj = Project.objects.create(name=project_name, organization=org)
+        inv = Inventory.objects.create(name=inventory_name, organization=org)
+        jt = JobTemplate.objects.create(name=jt_name, project=proj, inventory=inv, playbook=playbook)
+
+        if credentials:
+            jt.credentials.add(*credentials)
+
+        return jt.create_unified_job()
+
+    return _create_job_template
+
+
 @mock.patch('awx.main.tasks.facts.settings')
 @mock.patch('awx.main.tasks.jobs.create_partition', return_value=True)
 def test_pre_post_run_hook_facts(mock_create_partition, mock_facts_settings, private_data_dir, execution_environment):

awx/main/tests/unit/test_fields.py

@@ -76,6 +76,9 @@ def test_custom_error_messages(schema, given, message):
         ({'fields': [{'id': 'token', 'label': 'Token', 'secret': 'bad'}]}, False),
         ({'fields': [{'id': 'token', 'label': 'Token', 'ask_at_runtime': True}]}, True),
         ({'fields': [{'id': 'token', 'label': 'Token', 'ask_at_runtime': 'bad'}]}, False),  # noqa
+        ({'fields': [{'id': 'token', 'label': 'Token', 'internal': True}]}, True),
+        ({'fields': [{'id': 'token', 'label': 'Token', 'internal': False}]}, True),
+        ({'fields': [{'id': 'token', 'label': 'Token', 'internal': 'bad'}]}, False),
         ({'fields': [{'id': 'become_method', 'label': 'Become', 'choices': 'not-a-list'}]}, False),  # noqa
         ({'fields': [{'id': 'become_method', 'label': 'Become', 'choices': []}]}, False),
         ({'fields': [{'id': 'become_method', 'label': 'Become', 'choices': ['su', 'sudo']}]}, True),  # noqa
@@ -204,6 +207,68 @@ def test_credential_creation_validation_failure(inputs):
     assert e.type in (ValidationError, DRFValidationError)
 
 
+def test_credential_input_field_excludes_internal_fields():
+    """Internal fields should be excluded from the schema generated by CredentialInputField,
+    preventing users from providing values for internally resolved fields."""
+    type_ = CredentialType(
+        kind='cloud',
+        name='SomeCloud',
+        managed=True,
+        inputs={
+            'fields': [
+                {'id': 'username', 'label': 'Username', 'type': 'string'},
+                {'id': 'resolved_token', 'label': 'Token', 'type': 'string', 'internal': True},
+            ]
+        },
+    )
+    cred = Credential(credential_type=type_, name="Test Credential", inputs={'username': 'joe'})
+    field = cred._meta.get_field('inputs')
+    schema = field.schema(cred)
+
+    assert 'username' in schema['properties']
+    assert 'resolved_token' not in schema['properties']
+
+
+def test_credential_input_field_rejects_values_for_internal_fields():
+    """Users should not be able to provide values for fields marked as internal."""
+    type_ = CredentialType(
+        kind='cloud',
+        name='SomeCloud',
+        managed=True,
+        inputs={
+            'fields': [
+                {'id': 'username', 'label': 'Username', 'type': 'string'},
+                {'id': 'resolved_token', 'label': 'Token', 'type': 'string', 'internal': True},
+            ]
+        },
+    )
+    cred = Credential(credential_type=type_, name="Test Credential", inputs={'username': 'joe', 'resolved_token': 'secret'})
+    field = cred._meta.get_field('inputs')
+
+    with pytest.raises(Exception) as e:
+        field.validate(cred.inputs, cred)
+    assert e.type in (ValidationError, DRFValidationError)
+
+
+def test_credential_input_field_accepts_non_internal_fields_only():
+    """Credentials with only non-internal field values should validate successfully."""
+    type_ = CredentialType(
+        kind='cloud',
+        name='SomeCloud',
+        managed=True,
+        inputs={
+            'fields': [
+                {'id': 'username', 'label': 'Username', 'type': 'string'},
+                {'id': 'resolved_token', 'label': 'Token', 'type': 'string', 'internal': True},
+            ]
+        },
+    )
+    cred = Credential(credential_type=type_, name="Test Credential", inputs={'username': 'joe'})
+    field = cred._meta.get_field('inputs')
+    # Should not raise
+    field.validate(cred.inputs, cred)
+
+
 def test_implicit_role_field_parents():
     """This assures that every ImplicitRoleField only references parents
     which are relationships that actually exist

awx/main/tests/unit/test_tasks.py

@@ -556,7 +556,8 @@ class TestGenericRun:
         task._write_extra_vars_file = mock.Mock()
 
         with mock.patch('awx.main.tasks.jobs.settings.AWX_TASK_ENV', {'FOO': 'BAR'}):
-            env = task.build_env(job, private_data_dir)
+            with mock.patch.object(task, 'build_credentials_list', return_value=[], autospec=True):
+                env = task.build_env(job, private_data_dir)
         assert env['FOO'] == 'BAR'
 
 
@@ -649,7 +650,9 @@ class TestJobCredentials(TestJobExecution):
         )
 
         with mock.patch.object(UnifiedJob, 'credentials', credentials_mock):
-            yield job
+            # Mock build_credentials_list to work with the cached credentials mechanism
+            with mock.patch.object(jobs.RunJob, 'build_credentials_list', return_value=job._credentials, autospec=True):
+                yield job
 
     @pytest.fixture
     def update_model_wrapper(self, job):