Repository: tomplus/kubernetes_asyncio (python)

Issue:
Refresh exec-based API credentials when they expire
This issue is to track the problem found here in the non-async library https://github.com/kubernetes-client/python/issues/741.

Currently the authorization token/api key is only refreshed when the config is initially loaded and the client does not support refreshing this credential when it expires. 

This can be fixed in this library in a very similar way as it was fixed in the non-async library. Non-async fix: https://github.com/kubernetes-client/python-base/pull/250. 

Pull request:
[fix] support refreshing exec api credentials
Fixes #257. This PR is based heavily on https://github.com/kubernetes-client/python-base/pull/250. It also includes the fix made [here](https://github.com/kubernetes-client/python/commit/a07531b93233a9565d973d861dc711fa26b17cc8) to the non-asyncio library. 

This change has been tested in multiple services that rely on a long running client that was previously failing due to expiring credentials.

Patch:
diff --git a/examples/tail.py b/examples/tail.py
index e92e01ee..17239e44 100644
--- a/examples/tail.py
+++ b/examples/tail.py
@@ -14,7 +14,7 @@ import argparse
 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()
diff --git a/kubernetes_asyncio/__init__.py b/kubernetes_asyncio/__init__.py
index c32957b0..9d80c74d 100644
--- a/kubernetes_asyncio/__init__.py
+++ b/kubernetes_asyncio/__init__.py
@@ -18,6 +18,7 @@ __version__ = "24.2.0+snapshot"
 
 import kubernetes_asyncio.client
 import kubernetes_asyncio.config
+import kubernetes_asyncio.dynamic
 import kubernetes_asyncio.stream
 import kubernetes_asyncio.utils
 import kubernetes_asyncio.watch
diff --git a/kubernetes_asyncio/config/kube_config.py b/kubernetes_asyncio/config/kube_config.py
index 736c2b62..e479b5c2 100644
--- a/kubernetes_asyncio/config/kube_config.py
+++ b/kubernetes_asyncio/config/kube_config.py
@@ -208,7 +208,7 @@ class KubeConfigLoader(object):
 
         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 @@ class KubeConfigLoader(object):
 
         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,
diff --git a/setup.py b/setup.py
index 4bad79d5..f5be2164 100644
--- a/setup.py
+++ b/setup.py
@@ -51,6 +51,7 @@ setup(
         'kubernetes_asyncio',
         'kubernetes_asyncio.client',
         'kubernetes_asyncio.config',
+        'kubernetes_asyncio.dynamic',
         'kubernetes_asyncio.watch',
         'kubernetes_asyncio.client.api',
         'kubernetes_asyncio.stream',


Test patch:
diff --git a/kubernetes_asyncio/config/kube_config_test.py b/kubernetes_asyncio/config/kube_config_test.py
index 5a279a1a..ac0c741b 100644
--- a/kubernetes_asyncio/config/kube_config_test.py
+++ b/kubernetes_asyncio/config/kube_config_test.py
@@ -961,7 +961,7 @@ class TestKubeConfigLoader(BaseTestCase):
             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 @@ class TestKubeConfigLoader(BaseTestCase):
         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 = {