blacklist special env vars from being used in CredentialType injectors

see: #5877

awx/main/models/credential.py

@@ -457,6 +457,14 @@ class CredentialType(CommonModelNameNotUnique):
 
     defaults = OrderedDict()
 
+    ENV_BLACKLIST = set((
+        'VIRTUAL_ENV', 'PATH', 'PYTHONPATH', 'PROOT_TMP_DIR', 'JOB_ID',
+        'INVENTORY_ID', 'INVENTORY_SOURCE_ID', 'INVENTORY_UPDATE_ID',
+        'AD_HOC_COMMAND_ID', 'REST_API_URL', 'REST_API_TOKEN', 'TOWER_HOST',
+        'MAX_EVENT_RES', 'CALLBACK_QUEUE', 'CALLBACK_CONNECTION', 'CACHE',
+        'JOB_CALLBACK_DEBUG', 'INVENTORY_HOSTVARS', 'FACT_QUEUE',
+    ))
+
     class Meta:
         app_label = 'main'
         ordering = ('kind', 'name')
@@ -613,6 +621,8 @@ class CredentialType(CommonModelNameNotUnique):
             namespace['tower'].filename = path
 
         for env_var, tmpl in self.injectors.get('env', {}).items():
+            if env_var.startswith('ANSIBLE_') or env_var in self.ENV_BLACKLIST:
+                continue
             env[env_var] = Template(tmpl).render(**namespace)
             safe_env[env_var] = Template(tmpl).render(**safe_namespace)
 

awx/main/tests/unit/test_tasks.py

@@ -619,6 +619,36 @@ class TestJobCredentials(TestJobExecution):
 
         assert env['MY_CLOUD_API_TOKEN'] == 'ABC123'
 
+    def test_custom_environment_injectors_with_reserved_env_var(self):
+        some_cloud = CredentialType(
+            kind='cloud',
+            name='SomeCloud',
+            managed_by_tower=False,
+            inputs={
+                'fields': [{
+                    'id': 'api_token',
+                    'label': 'API Token',
+                    'type': 'string'
+                }]
+            },
+            injectors={
+                'env': {
+                    'JOB_ID': 'reserved'
+                }
+            }
+        )
+        self.instance.cloud_credential = Credential(
+            credential_type=some_cloud,
+            inputs = {'api_token': 'ABC123'}
+        )
+        self.task.run(self.pk)
+
+        assert self.task.run_pexpect.call_count == 1
+        call_args, _ = self.task.run_pexpect.call_args_list[0]
+        job, args, cwd, env, passwords, stdout = call_args
+
+        assert env['JOB_ID'] == str(self.instance.pk)
+
     def test_custom_environment_injectors_with_secret_field(self):
         some_cloud = CredentialType(
             kind='cloud',