Refresh GCP tokens on retrieval via custom dict

[Fix kubernetes-client#59]

config/kube_config.py

@@ -14,6 +14,7 @@
 
 import atexit
 import base64
+import collections
 import datetime
 import json
 import logging
@@ -126,6 +127,35 @@ def as_data(self):
         return self._data
 
 
+class GcpTokenApiKeyDict(collections.MutableMapping):
+    """Custom dict which allows GCP auth tokens to be refreshed."""
+
+    def __init__(self, copy_target, kube_config, provider):
+        self._dict = dict.copy(copy_target)
+        self.kube_config = kube_config
+        self.provider = provider
+
+    def __getitem__(self, key):
+        if key == 'authorization':
+            return self.kube_config.load_gcp_token(self.provider)
+        return self._dict[key]
+
+    def __setitem__(self, key, value):
+        self._dict[key] = value
+
+    def __delitem__(self, key):
+        del self._dict[key]
+
+    def __iter__(self):
+        return iter(self._dict)
+
+    def __len__(self):
+        return len(self._dict)
+
+    def __keytransform__(self, key):
+        return key
+
+
 class KubeConfigLoader(object):
 
     def __init__(self, config_dict, active_context=None,
@@ -200,7 +230,7 @@ def _load_auth_provider_token(self):
         if 'name' not in provider:
             return
         if provider['name'] == 'gcp':
-            return self._load_gcp_token(provider)
+            return self.load_gcp_token(provider)
         if provider['name'] == 'azure':
             return self._load_azure_token(provider)
         if provider['name'] == 'oidc':
@@ -234,7 +264,7 @@ def _refresh_azure_token(self, config):
         if self._config_persister:
             self._config_persister(self._config.value)
 
-    def _load_gcp_token(self, provider):
+    def load_gcp_token(self, provider):
         if (('config' not in provider) or
                 ('access-token' not in provider['config']) or
                 ('expiry' in provider['config'] and
@@ -394,7 +424,14 @@ def _load_cluster_info(self):
 
     def _set_config(self, client_configuration):
         if 'token' in self.__dict__:
-            client_configuration.api_key['authorization'] = self.token
+            if 'auth-provider' in self._user and 'name' in self._user['auth-provider'] and self._user['auth-provider']['name'] == 'gcp':
+                # GCP authorization tokens must be regularly refreshed.
+                gcp_dict = GcpTokenApiKeyDict(client_configuration.api_key, self, self._user['auth-provider'])
+                client_configuration.api_key = gcp_dict
+                # for tests only.
+                client_configuration.api_key['authorization'] = self.token
+            else:
+                client_configuration.api_key['authorization'] = self.token
         # copy these keys directly from self to configuration object
         keys = ['host', 'ssl_ca_cert', 'cert_file', 'key_file', 'verify_ssl']
         for key in keys:

config/kube_config_test.py

@@ -34,7 +34,9 @@
 
 EXPIRY_DATETIME_FORMAT = "%Y-%m-%dT%H:%M:%SZ"
 # should be less than kube_config.EXPIRY_SKEW_PREVENTION_DELAY
-EXPIRY_TIMEDELTA = 2
+PAST_EXPIRY_TIMEDELTA = 2
+# should be more than kube_config.EXPIRY_SKEW_PREVENTION_DELAY
+FUTURE_EXPIRY_TIMEDELTA = 60
 
 NON_EXISTING_FILE = "zz_non_existing_file_472398324"
 
@@ -73,8 +75,10 @@ def _raise_exception(st):
 TEST_PASSWORD = "pass"
 # token for me:pass
 TEST_BASIC_TOKEN = "Basic bWU6cGFzcw=="
-TEST_TOKEN_EXPIRY = _format_expiry_datetime(
-    datetime.datetime.utcnow() - datetime.timedelta(minutes=EXPIRY_TIMEDELTA))
+DATETIME_EXPIRY_PAST = datetime.datetime.utcnow() - datetime.timedelta(minutes=PAST_EXPIRY_TIMEDELTA)
+DATETIME_EXPIRY_FUTURE = datetime.datetime.utcnow() + datetime.timedelta(minutes=FUTURE_EXPIRY_TIMEDELTA)
+TEST_TOKEN_EXPIRY_PAST = _format_expiry_datetime(DATETIME_EXPIRY_PAST)
+TEST_TOKEN_EXPIRY_FUTURE = _format_expiry_datetime(DATETIME_EXPIRY_FUTURE)
 
 TEST_SSL_HOST = "https://test-host"
 TEST_CERTIFICATE_AUTH = "cert-auth"
@@ -495,6 +499,7 @@ class TestKubeConfigLoader(BaseTestCase):
                         "name": "gcp",
                         "config": {
                             "access-token": TEST_DATA_BASE64,
+                            "expiry": TEST_TOKEN_EXPIRY_FUTURE,
                         }
                     },
                     "token": TEST_DATA_BASE64,  # should be ignored
@@ -509,7 +514,7 @@ class TestKubeConfigLoader(BaseTestCase):
                         "name": "gcp",
                         "config": {
                             "access-token": TEST_DATA_BASE64,
-                            "expiry": TEST_TOKEN_EXPIRY,  # always in past
+                            "expiry": TEST_TOKEN_EXPIRY_PAST,  # always in past
                         }
                     },
                     "token": TEST_DATA_BASE64,  # should be ignored
@@ -654,7 +659,7 @@ def test_load_gcp_token_no_refresh(self):
     def test_load_gcp_token_with_refresh(self):
         def cred(): return None
         cred.token = TEST_ANOTHER_DATA_BASE64
-        cred.expiry = datetime.datetime.now()
+        cred.expiry = datetime.datetime.utcnow()
 
         loader = KubeConfigLoader(
             config_dict=self.TEST_KUBE_CONFIG,
@@ -668,6 +673,40 @@ def cred(): return None
         self.assertEqual(BEARER_TOKEN_FORMAT % TEST_ANOTHER_DATA_BASE64,
                          loader.token)
 
+    def test_load_gcp_token_with_refresh_dict_get(self):
+        class nonlocal:
+            accessed = False
+
+        class cred_old:
+            token = TEST_DATA_BASE64
+            expiry = DATETIME_EXPIRY_PAST
+
+        class cred_new:
+            token = TEST_ANOTHER_DATA_BASE64
+            expiry = DATETIME_EXPIRY_FUTURE
+        actual = FakeConfig()
+
+        def get_google_credentials():
+            if nonlocal.accessed:
+                return cred_new
+            nonlocal.accessed = True
+            return cred_old
+
+        loader = KubeConfigLoader(
+            config_dict=self.TEST_KUBE_CONFIG,
+            active_context="expired_gcp",
+            get_google_credentials=get_google_credentials)
+        loader.load_and_set(actual)
+        original_expiry = _get_expiry(loader)
+        token = actual.api_key['authorization']
+        new_expiry = _get_expiry(loader)
+        # assert that the configs expiry actually updates
+        self.assertTrue(new_expiry > original_expiry)
+        self.assertEqual(BEARER_TOKEN_FORMAT % TEST_ANOTHER_DATA_BASE64,
+                         loader.token)
+        self.assertEqual(BEARER_TOKEN_FORMAT % TEST_ANOTHER_DATA_BASE64,
+                         token)
+
     def test_oidc_no_refresh(self):
         loader = KubeConfigLoader(
             config_dict=self.TEST_KUBE_CONFIG,