Periodically refresh ServiceAccount tokens

Periodically refresh ServiceAccount tokens. This is required to avoid
authentication errors when time-bound tokens [1] are rotated, and the
initially-read token expires. Time bound tokens are beta in Kubernetes
1.21, and GA in 1.22 [2].

[1]: https://github.com/kubernetes/enhancements/tree/master/keps/sig-auth/1205-bound-service-account-tokens
[2]: kubernetes/enhancements#542

kubernetes_asyncio/config/incluster_config.py

@@ -12,6 +12,7 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
+import datetime
 import os
 
 from kubernetes_asyncio.client import Configuration
@@ -35,11 +36,16 @@ def _join_host_port(host, port):
 
 class InClusterConfigLoader(object):
 
-    def __init__(self, token_filename,
-                 cert_filename, environ=os.environ):
+    def __init__(self,
+                 token_filename,
+                 cert_filename,
+                 try_refresh_token=True,
+                 environ=os.environ):
         self._token_filename = token_filename
         self._cert_filename = cert_filename
         self._environ = environ
+        self._try_refresh_token = try_refresh_token
+        self._token_refresh_period = datetime.timedelta(minutes=1)
 
     def load_and_set(self, client_configuration=None):
         self._load_config()
@@ -64,12 +70,9 @@ def _load_config(self):
                                          self._environ[SERVICE_PORT_ENV_NAME]))
 
         if not os.path.isfile(self._token_filename):
-            raise ConfigException("Service token file does not exists.")
+            raise ConfigException("Service token file does not exist.")
 
-        with open(self._token_filename) as f:
-            self.token = f.read()
-            if not self.token:
-                raise ConfigException("Token file exists but empty.")
+        self._read_token_file()
 
         if not os.path.isfile(self._cert_filename):
             raise ConfigException(
@@ -84,10 +87,29 @@ def _load_config(self):
     def _set_config(self, configuration):
         configuration.host = self.host
         configuration.ssl_ca_cert = self.ssl_ca_cert
-        configuration.api_key['BearerToken'] = "Bearer " + self.token
+        if self.token is not None:
+            configuration.api_key['BearerToken'] = self.token
+        if not self._try_refresh_token:
+            return
+
+        def load_token_from_file(*args):
+            if self.token_expires_at <= datetime.datetime.now():
+                self._read_token_file()
+            return self.token
+
+        configuration.get_api_key_with_prefix = load_token_from_file
+
+    def _read_token_file(self):
+        with open(self._token_filename) as f:
+            content = f.read()
+            if not content:
+                raise ConfigException("Token file exists but empty.")
+            self.token = "Bearer " + content
+            self.token_expires_at = datetime.datetime.now(
+            ) + self._token_refresh_period
 
 
-def load_incluster_config(client_configuration=None):
+def load_incluster_config(client_configuration=None, try_refresh_token=True):
     """Use the service account kubernetes gives to pods to connect to kubernetes
     cluster. It's intended for clients that expect to be running inside a pod
     running on kubernetes. It will raise an exception if called from a process
@@ -96,5 +118,7 @@ def load_incluster_config(client_configuration=None):
     :param client_configuration: The kubernetes.client.Configuration to
     set configs to.
     """
-    InClusterConfigLoader(token_filename=SERVICE_TOKEN_FILENAME,
-                          cert_filename=SERVICE_CERT_FILENAME).load_and_set(client_configuration)
+    InClusterConfigLoader(
+        token_filename=SERVICE_TOKEN_FILENAME,
+        cert_filename=SERVICE_CERT_FILENAME,
+        try_refresh_token=try_refresh_token).load_and_set(client_configuration)

kubernetes_asyncio/config/incluster_config_test.py

@@ -12,6 +12,7 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
+import datetime
 import os
 import tempfile
 import unittest
@@ -25,6 +26,7 @@
 )
 
 _TEST_TOKEN = "temp_token"
+_TEST_NEW_TOKEN = "temp_new_token"
 _TEST_CERT = "temp_cert"
 _TEST_HOST = "127.0.0.1"
 _TEST_PORT = "80"
@@ -66,6 +68,7 @@ def get_test_loader(
         return InClusterConfigLoader(
             token_filename=token_filename,
             cert_filename=cert_filename,
+            try_refresh_token=True,
             environ=environ)
 
     def test_join_host_port(self):
@@ -80,7 +83,29 @@ def test_load_config(self):
         loader._load_config()
         self.assertEqual("https://" + _TEST_HOST_PORT, loader.host)
         self.assertEqual(cert_filename, loader.ssl_ca_cert)
-        self.assertEqual(_TEST_TOKEN, loader.token)
+        self.assertEqual('Bearer ' + _TEST_TOKEN, loader.token)
+
+    def test_refresh_token(self):
+        loader = self.get_test_loader()
+        config = Configuration()
+        loader.load_and_set(config)
+
+        self.assertEqual('Bearer ' + _TEST_TOKEN,
+                         config.get_api_key_with_prefix('authorization'))
+        self.assertEqual('Bearer ' + _TEST_TOKEN, loader.token)
+        self.assertIsNotNone(loader.token_expires_at)
+
+        old_token_expires_at = loader.token_expires_at
+        loader._token_filename = self._create_file_with_temp_content(
+            _TEST_NEW_TOKEN)
+        self.assertEqual('Bearer ' + _TEST_TOKEN,
+                         config.get_api_key_with_prefix('authorization'))
+
+        loader.token_expires_at = datetime.datetime.now()
+        self.assertEqual('Bearer ' + _TEST_NEW_TOKEN,
+                         config.get_api_key_with_prefix('authorization'))
+        self.assertEqual('Bearer ' + _TEST_NEW_TOKEN, loader.token)
+        self.assertGreater(loader.token_expires_at, old_token_expires_at)
 
     def _should_fail_load(self, config_loader, reason):
         try: