Load certificates from plugin exec (#269)

* Load certificates from plugin exec

This mostly imports what was done in the sync client in kubernetes-client/python@b85aff2

This may solve #231: if the configuration from #213 was trying to load a
configuration file containing these client certificates instead of a
bearer token, the log message would display and the connection to
Kubernetes would fail.

* linter

* move expirationTimestamp next to the token authentication block

kubernetes_asyncio/config/kube_config.py

@@ -314,13 +314,34 @@ async def load_from_exec_plugin(self):
         try:
             if hasattr(self, 'exec_plugin_expiry') and not _is_expired(self.exec_plugin_expiry):
                 return True
+            base_path = self._get_base_path(self._cluster.path)
             status = await ExecProvider(self._user['exec']).run()
-            if 'token' not in status:
-                logger.error('exec: missing token field in plugin output')
-                return None
-            self.token = "Bearer %s" % status['token']
-            if 'expirationTimestamp' in status:
-                self.exec_plugin_expiry = parse_rfc3339(status['expirationTimestamp'])
+            if 'token' in status:
+                self.token = "Bearer %s" % status['token']
+                if 'expirationTimestamp' in status:
+                    self.exec_plugin_expiry = parse_rfc3339(status['expirationTimestamp'])
+            elif 'clientCertificateData' in status:
+                # https://kubernetes.io/docs/reference/access-authn-authz/authentication/#input-and-output-formats
+                # Plugin has provided certificates instead of a token.
+                if 'clientKeyData' not in status:
+                    logger.error('exec: missing clientKeyData field in '
+                                 'plugin output')
+                    return None
+                self.cert_file = FileOrData(
+                    status, None,
+                    data_key_name='clientCertificateData',
+                    file_base_path=base_path,
+                    base64_file_content=False,
+                    temp_file_path=self._temp_file_path).as_file()
+                self.key_file = FileOrData(
+                    status, None,
+                    data_key_name='clientKeyData',
+                    file_base_path=base_path,
+                    base64_file_content=False,
+                    temp_file_path=self._temp_file_path).as_file()
+            else:
+                logger.error('exec: missing token or clientCertificateData '
+                             'field in plugin output')
             return True
         except Exception as e:
             logger.error(str(e))
@@ -358,14 +379,18 @@ def _load_cluster_info(self):
                     self._cluster, 'certificate-authority',
                     file_base_path=base_path,
                     temp_file_path=self._temp_file_path).as_file()
-                self.cert_file = FileOrData(
-                    self._user, 'client-certificate',
-                    file_base_path=base_path,
-                    temp_file_path=self._temp_file_path).as_file()
-                self.key_file = FileOrData(
-                    self._user, 'client-key',
-                    file_base_path=base_path,
-                    temp_file_path=self._temp_file_path).as_file()
+                if 'cert_file' not in self.__dict__:
+                    # cert_file could have been provided by
+                    # _load_from_exec_plugin; only load from the _user
+                    # section if we need it.
+                    self.cert_file = FileOrData(
+                        self._user, 'client-certificate',
+                        file_base_path=base_path,
+                        temp_file_path=self._temp_file_path).as_file()
+                    self.key_file = FileOrData(
+                        self._user, 'client-key',
+                        file_base_path=base_path,
+                        temp_file_path=self._temp_file_path).as_file()
         if 'insecure-skip-tls-verify' in self._cluster:
             self.verify_ssl = not self._cluster['insecure-skip-tls-verify']
 

kubernetes_asyncio/config/kube_config_test.py

@@ -436,6 +436,13 @@ class TestKubeConfigLoader(BaseTestCase):
                     "user": "exec_cred_user"
                 }
             },
+            {
+                "name": "exec_cred_user_certificate",
+                "context": {
+                    "cluster": "ssl",
+                    "user": "exec_cred_user_certificate"
+                }
+            },
         ],
         "clusters": [
             {
@@ -462,7 +469,9 @@ class TestKubeConfigLoader(BaseTestCase):
                 "name": "ssl",
                 "cluster": {
                     "server": TEST_SSL_HOST,
-                    "certificate-authority-data": TEST_CERTIFICATE_AUTH_BASE64,
+                    "certificate-authority-data":
+                        TEST_CERTIFICATE_AUTH_BASE64,
+                    "insecure-skip-tls-verify": False,
                 }
             },
             {
@@ -601,6 +610,16 @@ class TestKubeConfigLoader(BaseTestCase):
                     }
                 }
             },
+            {
+                "name": "exec_cred_user_certificate",
+                "user": {
+                    "exec": {
+                        "apiVersion": "client.authentication.k8s.io/v1beta1",
+                        "command": "custom-certificate-authenticator",
+                        "args": []
+                    }
+                }
+            },
         ]
     }
 
@@ -757,6 +776,24 @@ async def test_user_exec_auth(self, mock):
             active_context="exec_cred_user").load_and_set(actual)
         self.assertEqual(expected, actual)
 
+    @patch('kubernetes_asyncio.config.kube_config.ExecProvider.run')
+    async def test_user_exec_auth_certificates(self, mock):
+        mock.return_value = {
+            "clientCertificateData": TEST_CLIENT_CERT,
+            "clientKeyData": TEST_CLIENT_KEY,
+        }
+        expected = FakeConfig(
+            host=TEST_SSL_HOST,
+            cert_file=self._create_temp_file(TEST_CLIENT_CERT),
+            key_file=self._create_temp_file(TEST_CLIENT_KEY),
+            ssl_ca_cert=self._create_temp_file(TEST_CERTIFICATE_AUTH),
+            verify_ssl=True)
+        actual = FakeConfig()
+        await KubeConfigLoader(
+            config_dict=self.TEST_KUBE_CONFIG,
+            active_context="exec_cred_user_certificate").load_and_set(actual)
+        self.assertEqual(expected, actual)
+
     async def test_user_pass(self):
         expected = FakeConfig(host=TEST_HOST, token=TEST_BASIC_TOKEN)
         actual = FakeConfig()
@@ -787,7 +824,8 @@ async def test_ssl(self):
             token=BEARER_TOKEN_FORMAT % TEST_DATA_BASE64,
             cert_file=self._create_temp_file(TEST_CLIENT_CERT),
             key_file=self._create_temp_file(TEST_CLIENT_KEY),
-            ssl_ca_cert=self._create_temp_file(TEST_CERTIFICATE_AUTH)
+            ssl_ca_cert=self._create_temp_file(TEST_CERTIFICATE_AUTH),
+            verify_ssl=True,
         )
         actual = FakeConfig()
         await KubeConfigLoader(
@@ -900,7 +938,8 @@ async def test_load_kube_config_from_dict_with_temp_file_path(self):
             token=BEARER_TOKEN_FORMAT % TEST_DATA_BASE64,
             cert_file=self._create_temp_file(TEST_CLIENT_CERT),
             key_file=self._create_temp_file(TEST_CLIENT_KEY),
-            ssl_ca_cert=self._create_temp_file(TEST_CERTIFICATE_AUTH)
+            ssl_ca_cert=self._create_temp_file(TEST_CERTIFICATE_AUTH),
+            verify_ssl=True,
         )
         actual = FakeConfig()