[fix] support refreshing exec api credentials (#258)

Co-authored-by: Harry Harpel <[email protected]>

Author: harryharpel

examples/tail.py

@@ -14,7 +14,7 @@
 import asyncio
 
 from kubernetes_asyncio import client, config
-from kubernetes_asyncio.client.api_client import ApiClient
+from kubernetes_asyncio.client.api_client import ApiClient, Configuration
 
 
 def parse_args():
@@ -51,8 +51,9 @@ async def print_pod_log(v1_api, pod, namespace, container, lines, follow):
 async def main():
     args = parse_args()
 
-    loader = await config.load_kube_config()
-    api = ApiClient()
+    client_configuration = Configuration()
+    loader = await config.load_kube_config(client_configuration=client_configuration)
+    api = ApiClient(configuration=client_configuration)
     v1_api = client.CoreV1Api(api)
     ret = await v1_api.list_namespaced_pod(args.namespace)
     cmd = []
@@ -72,7 +73,9 @@ async def main():
 
     if args.follow:
         # autorefresh gcp token
-        cmd.append(config.refresh_token(loader))
+        cmd.append(config.refresh_token(
+            loader=loader,
+            client_configuration=client_configuration))
 
     await asyncio.wait(cmd)
     await api.close()

kubernetes_asyncio/config/kube_config.py

@@ -208,7 +208,7 @@ async def _load_authentication(self):
 
         if 'exec' in self._user:
             logging.debug('Try to use exec provider')
-            res_exec_plugin = await self._load_from_exec_plugin()
+            res_exec_plugin = await self.load_from_exec_plugin()
             if res_exec_plugin:
                 return
 
@@ -309,13 +309,17 @@ def _retrieve_oidc_cacert(self, provider):
 
         return None
 
-    async def _load_from_exec_plugin(self):
+    async def load_from_exec_plugin(self):
         try:
+            if hasattr(self, 'exec_plugin_expiry') and not _is_expired(self.exec_plugin_expiry):
+                return True
             status = await ExecProvider(self._user['exec']).run()
             if 'token' not in status:
                 logging.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'])
             return True
         except Exception as e:
             logging.error(str(e))
@@ -617,16 +621,20 @@ async def refresh_token(loader, client_configuration=None, interval=60):
     :param interval: how often check if token is up-to-date
 
     """
-    if loader.provider != 'gcp':
-        return
 
     if client_configuration is None:
-        client_configuration = Configuration()
-
-    while 1:
-        await asyncio.sleep(interval)
-        await loader.load_gcp_token()
-        client_configuration.api_key['BearerToken'] = loader.token
+        raise NotImplementedError
+
+    if loader.provider == 'gcp':
+        while 1:
+            await asyncio.sleep(interval)
+            await loader.load_gcp_token()
+            client_configuration.api_key['BearerToken'] = loader.token
+    elif 'exec' in loader._user:
+        while 1:
+            await asyncio.sleep(interval)
+            await loader.load_from_exec_plugin()
+            client_configuration.api_key['BearerToken'] = loader.token
 
 
 async def new_client_from_config(config_file=None, context=None, persist_config=True,

kubernetes_asyncio/config/kube_config_test.py

@@ -961,7 +961,7 @@ async def test_non_existing_user(self):
             active_context="non_existing_user").load_and_set(actual)
         self.assertEqual(expected, actual)
 
-    async def test_refresh_token(self):
+    async def test_refresh_gcp_token(self):
         loader = KubeConfigLoader(
             config_dict=self.TEST_KUBE_CONFIG,
             active_context="gcp",
@@ -979,6 +979,27 @@ async def test_refresh_token(self):
         self.assertEqual(BEARER_TOKEN_FORMAT % TEST_DATA_BASE64,
                          loader.token)
 
+    async def test_refresh_exec_token(self):
+        class MockKubeConfigLoader(KubeConfigLoader):
+            async def load_from_exec_plugin(self):
+                self.token = TEST_ANOTHER_DATA_BASE64
+
+        loader = MockKubeConfigLoader(
+            config_dict=self.TEST_KUBE_CONFIG,
+            active_context="exec_cred_user")
+        mock_sleep = patch('asyncio.sleep').start()
+        mock_sleep.side_effect = [0, AssertionError]
+
+        mock_config = Mock()
+        mock_config.api_key = {}
+
+        loader.exec_plugin_expiry = datetime.datetime.now()
+
+        with self.assertRaises(AssertionError):
+            await refresh_token(loader, mock_config)
+
+        self.assertEqual(TEST_ANOTHER_DATA_BASE64, mock_config.api_key["BearerToken"])
+
 
 class TestKubeConfigMerger(BaseTestCase):
     TEST_KUBE_CONFIG_PART1 = {