refactor k8s credential injectors to properly handle verify=f

awx/main/models/credential/__init__.py

@@ -1169,18 +1169,7 @@ ManagedCredentialType(
             'multiline': True,
         }],
         'required': ['host', 'bearer_token'],
-    },
+    }
-    injectors={
-        'file': {
-            'template': '{{ ssl_ca_cert }}'
-        },
-        'env': {
-            'K8S_AUTH_HOST': '{{ host }}',
-            'K8S_AUTH_API_KEY': '{{ bearer_token }}',
-            'K8S_AUTH_VERIFY_SSL': '{{ verify_ssl }}',
-            'K8S_AUTH_SSL_CA_CERT': '{{ tower.filename }}',
-        },
-    },
 )
 
 

awx/main/models/credential/injectors.py

@@ -101,3 +101,17 @@ def openstack(cred, env, private_data_dir):
     f.close()
     os.chmod(path, stat.S_IRUSR | stat.S_IWUSR)
     env['OS_CLIENT_CONFIG_FILE'] = path
+
+
+def kubernetes_bearer_token(cred, env, private_data_dir):
+    env['K8S_AUTH_HOST'] = cred.get_input('host', default='')
+    env['K8S_AUTH_API_KEY'] = cred.get_input('bearer_token', default='')
+    if cred.get_input('verify_ssl') and 'ssl_ca_cert' in cred.inputs:
+        env['K8S_AUTH_VERIFY_SSL'] = 'True'
+        handle, path = tempfile.mkstemp(dir=private_data_dir)
+        with os.fdopen(handle, 'w') as f:
+            os.chmod(path, stat.S_IRUSR | stat.S_IWUSR)
+            f.write(cred.get_input('ssl_ca_cert'))
+        env['K8S_AUTH_SSL_CA_CERT'] = path
+    else:
+        env['K8S_AUTH_VERIFY_SSL'] = 'False'

awx/main/tests/unit/test_tasks.py

@@ -1037,17 +1037,20 @@ class TestJobCredentials(TestJobExecution):
         assert '--vault-id dev@prompt' in ' '.join(args)
         assert '--vault-id prod@prompt' in ' '.join(args)
 
-    def test_k8s_credential(self, job, private_data_dir):
+    @pytest.mark.parametrize("verify", (True, False))
+    def test_k8s_credential(self, job, private_data_dir, verify):
         k8s = CredentialType.defaults['kubernetes_bearer_token']()
+        inputs = {
+            'host': 'https://example.org/',
+            'bearer_token': 'token123',
+        }
+        if verify:
+            inputs['verify_ssl'] = True
+            inputs['ssl_ca_cert'] = 'CERTDATA'
         credential = Credential(
             pk=1,
             credential_type=k8s,
-            inputs = {
+            inputs = inputs,
-                'host': 'https://example.org/',
-                'bearer_token': 'token123',
-                'verify_ssl': True,
-                'ssl_ca_cert': 'CERTDATA'
-            }
         )
         credential.inputs['bearer_token'] = encrypt_field(credential, 'bearer_token')
         job.credentials.add(credential)
@@ -1060,9 +1063,15 @@ class TestJobCredentials(TestJobExecution):
 
         assert env['K8S_AUTH_HOST'] == 'https://example.org/'
         assert env['K8S_AUTH_API_KEY'] == 'token123'
-        assert env['K8S_AUTH_VERIFY_SSL'] == 'True'
+
-        cert = open(env['K8S_AUTH_SSL_CA_CERT'], 'r').read()
+        if verify:
-        assert cert == 'CERTDATA'
+            assert env['K8S_AUTH_VERIFY_SSL'] == 'True'
+            cert = open(env['K8S_AUTH_SSL_CA_CERT'], 'r').read()
+            assert cert == 'CERTDATA'
+        else:
+            assert env['K8S_AUTH_VERIFY_SSL'] == 'False'
+            assert 'K8S_AUTH_SSL_CA_CERT' not in env
+
         assert safe_env['K8S_AUTH_API_KEY'] == tasks.HIDDEN_PASSWORD
 
     def test_aws_cloud_credential(self, job, private_data_dir):