diff --git a/awx/api/serializers.py b/awx/api/serializers.py
index b0cabac93b..4cd4353380 100644
--- a/awx/api/serializers.py
+++ b/awx/api/serializers.py
@@ -1412,6 +1412,11 @@ class ExecutionEnvironmentSerializer(BaseSerializer):
             res['credential'] = self.reverse('api:credential_detail', kwargs={'pk': obj.credential.pk})
         return res
 
+    def validate_credential(self, value):
+        if value and value.kind != 'registry':
+            raise serializers.ValidationError(_('Only Container Registry credentials can be associated with an Execution Environment'))
+        return value
+
     def validate(self, attrs):
         # prevent changing organization of ee. Unsetting (change to null) is allowed
         if self.instance:
diff --git a/awx/main/models/credential/__init__.py b/awx/main/models/credential/__init__.py
index f37bfee884..01e71302bf 100644
--- a/awx/main/models/credential/__init__.py
+++ b/awx/main/models/credential/__init__.py
@@ -295,6 +295,15 @@ class Credential(PasswordFieldsModel, CommonModelNameNotUnique, ResourceMixin):
             return True
         return field_name in self.inputs and self.inputs[field_name] not in ('', None)
 
+    def has_inputs(self, field_names=()):
+        for name in field_names:
+            if name in self.inputs:
+                if self.inputs[name] in ('', None):
+                    return False
+            else:
+                raise ValueError('{} is not an input field'.format(name))
+        return True
+
     def _get_dynamic_input(self, field_name):
         for input_source in self.input_sources.all():
             if input_source.input_field_name == field_name:
@@ -1097,16 +1106,10 @@ ManagedCredentialType(
             },
             {
                 'id': 'password',
-                'label': ugettext_noop('Password'),
+                'label': ugettext_noop('Password or Token'),
                 'type': 'string',
                 'secret': True,
-            },
-            {
-                'id': 'token',
-                'label': ugettext_noop('Access Token'),
-                'type': 'string',
-                'secret': True,
-                'help_text': ugettext_noop('A token to use to authenticate with. ' 'This should not be set if username/password are being used.'),
+                'help_text': ugettext_noop('A password or token used to authenticate with'),
             },
         ],
         'required': ['host'],
diff --git a/awx/main/tasks.py b/awx/main/tasks.py
index 0acaee3b9c..fd01a6140e 100644
--- a/awx/main/tasks.py
+++ b/awx/main/tasks.py
@@ -834,7 +834,7 @@ class BaseTask(object):
         """
         return os.path.abspath(os.path.join(os.path.dirname(__file__), *args))
 
-    def build_execution_environment_params(self, instance):
+    def build_execution_environment_params(self, instance, private_data_dir):
         if settings.IS_K8S:
             return {}
 
@@ -851,6 +851,23 @@ class BaseTask(object):
             "container_options": ['--user=root'],
         }
 
+        if instance.execution_environment.credential:
+            cred = instance.execution_environment.credential
+            if cred.has_inputs(field_names=('host', 'username', 'password')):
+                path = os.path.split(private_data_dir)[0]
+                with open(path + '/auth.json', 'w') as authfile:
+                    host = cred.get_input('host')
+                    username = cred.get_input('username')
+                    password = cred.get_input('password')
+                    token = "{}:{}".format(username, password)
+                    auth_data = {'auths': {host: {'auth': b64encode(token.encode('ascii')).decode()}}}
+                    authfile.write(json.dumps(auth_data, indent=4))
+                authfile.close()
+                os.chmod(authfile.name, stat.S_IRUSR | stat.S_IWUSR | stat.S_IXUSR)
+                params["container_options"].append(f'--authfile={authfile.name}')
+            else:
+                raise RuntimeError('Please recheck that your host, username, and password fields are all filled.')
+
         pull = instance.execution_environment.pull
         if pull:
             params['container_options'].append(f'--pull={pull}')
@@ -1709,11 +1726,11 @@ class RunJob(BaseTask):
         """
         return settings.AWX_RESOURCE_PROFILING_ENABLED
 
-    def build_execution_environment_params(self, instance):
+    def build_execution_environment_params(self, instance, private_data_dir):
         if settings.IS_K8S:
             return {}
 
-        params = super(RunJob, self).build_execution_environment_params(instance)
+        params = super(RunJob, self).build_execution_environment_params(instance, private_data_dir)
         # If this has an insights agent and it is not already mounted then show it
         insights_dir = os.path.dirname(settings.INSIGHTS_SYSTEM_ID_FILE)
         if instance.use_fact_cache and os.path.exists(insights_dir):
@@ -2324,11 +2341,11 @@ class RunProjectUpdate(BaseTask):
             if status == 'successful' and instance.launch_type != 'sync':
                 self._update_dependent_inventories(instance, dependent_inventory_sources)
 
-    def build_execution_environment_params(self, instance):
+    def build_execution_environment_params(self, instance, private_data_dir):
         if settings.IS_K8S:
             return {}
 
-        params = super(RunProjectUpdate, self).build_execution_environment_params(instance)
+        params = super(RunProjectUpdate, self).build_execution_environment_params(instance, private_data_dir)
         project_path = instance.get_project_path(check_if_exists=False)
         cache_path = instance.get_cache_path()
         params.setdefault('container_volume_mounts', [])
@@ -2831,7 +2848,7 @@ class RunSystemJob(BaseTask):
     event_model = SystemJobEvent
     event_data_key = 'system_job_id'
 
-    def build_execution_environment_params(self, system_job):
+    def build_execution_environment_params(self, system_job, private_data_dir):
         return {}
 
     def build_args(self, system_job, private_data_dir, passwords):
@@ -2947,7 +2964,7 @@ class AWXReceptorJob:
         self.unit_id = None
 
         if self.task and not self.task.instance.is_container_group_task:
-            execution_environment_params = self.task.build_execution_environment_params(self.task.instance)
+            execution_environment_params = self.task.build_execution_environment_params(self.task.instance, runner_params['private_data_dir'])
             self.runner_params['settings'].update(execution_environment_params)
 
     def run(self):