Resolving cert files relative path using kube-config file path.

Author: mbohlool

kubernetes/config/kube_config.py

@@ -28,8 +28,13 @@
 
 
 def _cleanup_temp_files():
+    global _temp_files
     for temp_file in _temp_files.values():
-        os.remove(temp_file)
+        try:
+            os.remove(temp_file)
+        except OSError:
+            pass
+    _temp_files = {}
 
 
 def _create_temp_file_with_content(content):
@@ -54,15 +59,17 @@ class FileOrData(object):
      obj['%data_key_name'] is not set or empty. Assumption is file content is
      raw data and data field is base64 string."""
 
-    def __init__(self, obj, file_key_name, data_key_name=None):
+    def __init__(self, obj, file_key_name, data_key_name=None,
+                 file_base_path=""):
         if not data_key_name:
             data_key_name = file_key_name + "-data"
         self._file = None
         self._data = None
         if data_key_name in obj:
             self._data = obj[data_key_name]
         elif file_key_name in obj:
-            self._file = obj[file_key_name]
+            self._file = os.path.normpath(
+                os.path.join(file_base_path, obj[file_key_name]))
 
     def as_file(self):
         """If obj[%data_key_name] exists, return name of a file with base64
@@ -71,6 +78,8 @@ def as_file(self):
         if use_data_if_no_file:
             self._file = _create_temp_file_with_content(
                 base64.decodestring(self._data.encode()))
+        if not os.path.isfile(self._file):
+            raise ConfigException("File does not exists: %s" % self._file)
         return self._file
 
     def as_data(self):
@@ -87,12 +96,14 @@ def as_data(self):
 class KubeConfigLoader(object):
 
     def __init__(self, config_dict, active_context=None,
-                 get_google_credentials=None, client_configuration=None):
+                 get_google_credentials=None, client_configuration=None,
+                 config_base_path=""):
         self._config = ConfigNode('kube-config', config_dict)
         self._current_context = None
         self._user = None
         self._cluster = None
         self.set_active_context(active_context)
+        self._config_base_path = config_base_path
         if get_google_credentials:
             self._get_google_credentials = get_google_credentials
         else:
@@ -151,7 +162,9 @@ def _load_gcp_token(self):
         return self.token
 
     def _load_user_token(self):
-        token = FileOrData(self._user, 'tokenFile', 'token').as_data()
+        token = FileOrData(
+            self._user, 'tokenFile', 'token',
+            file_base_path=self._config_base_path).as_data()
         if token:
             self.token = token
             return True
@@ -168,10 +181,14 @@ def _load_cluster_info(self):
             self.host = self._cluster['server']
             if self.host.startswith("https"):
                 self.ssl_ca_cert = FileOrData(
-                    self._cluster, 'certificate-authority').as_file()
+                    self._cluster, 'certificate-authority',
+                    file_base_path=self._config_base_path).as_file()
                 self.cert_file = FileOrData(
-                    self._user, 'client-certificate').as_file()
-                self.key_file = FileOrData(self._user, 'client-key').as_file()
+                    self._user, 'client-certificate',
+                    file_base_path=self._config_base_path).as_file()
+                self.key_file = FileOrData(
+                    self._user, 'client-key',
+                    file_base_path=self._config_base_path).as_file()
 
     def _set_config(self):
         if 'token' in self.__dict__:
@@ -244,9 +261,16 @@ def get_with_name(self, name):
             'Expected object with name %s in %s list' % (name, self.name))
 
 
+def _get_kube_config_loader_for_yaml_file(filename, **kwargs):
+    with open(filename) as f:
+        return KubeConfigLoader(
+            config_dict=yaml.load(f),
+            config_base_path=os.path.abspath(os.path.dirname(filename)),
+            **kwargs)
+
+
 def list_kube_config_contexts(config_file):
-    with open(config_file) as f:
-        loader = KubeConfigLoader(config_dict=yaml.load(f))
+    loader = _get_kube_config_loader_for_yaml_file(config_file)
     return loader.list_contexts(), loader.current_context
 
 
@@ -259,6 +283,5 @@ def load_kube_config(config_file, context=None):
     from config file will be used.
     """
 
-    with open(config_file) as f:
-        KubeConfigLoader(
-            config_dict=yaml.load(f), active_context=context).load_and_set()
+    _get_kube_config_loader_for_yaml_file(
+        config_file, active_context=context).load_and_set()

kubernetes/config/kube_config_test.py

@@ -14,12 +14,15 @@
 
 import base64
 import os
+import shutil
 import tempfile
 import unittest
 
 from .config_exception import ConfigException
 from .kube_config import (ConfigNode, FileOrData, KubeConfigLoader,
-                          _create_temp_file_with_content)
+                          _create_temp_file_with_content, _cleanup_temp_files)
+
+NON_EXISTING_FILE = "zz_non_existing_file_472398324"
 
 
 def _base64(string):
@@ -67,6 +70,11 @@ def _create_temp_file(self, content=""):
         os.close(handler)
         return name
 
+    def expect_exception(self, func, message_part):
+        with self.assertRaises(ConfigException) as context:
+            func()
+        self.assertIn(message_part, str(context.exception))
+
 
 class TestFileOrData(BaseTestCase):
 
@@ -76,9 +84,16 @@ def get_file_content(filename):
             return f.read()
 
     def test_file_given_file(self):
-        obj = {TEST_FILE_KEY: TEST_FILENAME}
+        temp_filename = _create_temp_file_with_content(TEST_DATA)
+        obj = {TEST_FILE_KEY: temp_filename}
         t = FileOrData(obj=obj, file_key_name=TEST_FILE_KEY)
-        self.assertEqual(TEST_FILENAME, t.as_file())
+        self.assertEqual(TEST_DATA, self.get_file_content(t.as_file()))
+
+    def test_file_given_non_existing_file(self):
+        temp_filename = NON_EXISTING_FILE
+        obj = {TEST_FILE_KEY: temp_filename}
+        t = FileOrData(obj=obj, file_key_name=TEST_FILE_KEY)
+        self.expect_exception(t.as_file, "does not exists")
 
     def test_file_given_data(self):
         obj = {TEST_DATA_KEY: TEST_DATA_BASE64}
@@ -116,10 +131,20 @@ def test_file_given_file_and_data(self):
                        data_key_name=TEST_DATA_KEY)
         self.assertEqual(TEST_DATA, self.get_file_content(t.as_file()))
 
+    def test_file_with_custom_dirname(self):
+        tempfile = self._create_temp_file(content=TEST_DATA)
+        tempfile_dir = os.path.dirname(tempfile)
+        tempfile_basename = os.path.basename(tempfile)
+        obj = {TEST_FILE_KEY: tempfile_basename}
+        t = FileOrData(obj=obj, file_key_name=TEST_FILE_KEY,
+                       file_base_path=tempfile_dir)
+        self.assertEqual(TEST_DATA, self.get_file_content(t.as_file()))
+
     def test_create_temp_file_with_content(self):
         self.assertEqual(TEST_DATA,
                          self.get_file_content(
                              _create_temp_file_with_content(TEST_DATA)))
+        _cleanup_temp_files()
 
 
 class TestConfigNode(BaseTestCase):
@@ -163,11 +188,6 @@ def test_get_with_name(self):
         self.assertEqual("test_obj/with_names[name=test_name3]",
                          node.get_with_name("test_name3").name)
 
-    def expect_exception(self, func, message_part):
-        with self.assertRaises(ConfigException) as context:
-            func()
-        self.assertIn(message_part, str(context.exception))
-
     def test_key_does_not_exists(self):
         self.expect_exception(lambda: self.node['not-exists-key'],
                               "Expected key not-exists-key in test_obj")
@@ -281,6 +301,13 @@ class TestKubeConfigLoader(BaseTestCase):
                     "user": "ssl-no_file"
                 }
             },
+            {
+                "name": "ssl-local-file",
+                "context": {
+                    "cluster": "ssl-local-file",
+                    "user": "ssl-local-file"
+                }
+            },
         ],
         "clusters": [
             {
@@ -296,6 +323,13 @@ class TestKubeConfigLoader(BaseTestCase):
                     "certificate-authority": TEST_CERTIFICATE_AUTH,
                 }
             },
+            {
+                "name": "ssl-local-file",
+                "cluster": {
+                    "server": TEST_SSL_HOST,
+                    "certificate-authority": "cert_test",
+                }
+            },
             {
                 "name": "ssl",
                 "cluster": {
@@ -340,6 +374,14 @@ class TestKubeConfigLoader(BaseTestCase):
                     "client-key": TEST_CLIENT_KEY,
                 }
             },
+            {
+                "name": "ssl-local-file",
+                "user": {
+                    "tokenFile": "token_file",
+                    "client-certificate": "client_cert",
+                    "client-key": "client_key",
+                }
+            },
             {
                 "name": "ssl",
                 "user": {
@@ -420,11 +462,11 @@ def test_ssl_no_cert_files(self):
             ssl_ca_cert=TEST_CERTIFICATE_AUTH
         )
         actual = FakeConfig()
-        KubeConfigLoader(
+        loader = KubeConfigLoader(
             config_dict=self.TEST_KUBE_CONFIG,
             active_context="ssl-no_file",
-            client_configuration=actual).load_and_set()
-        self.assertEqual(expected, actual)
+            client_configuration=actual)
+        self.expect_exception(loader.load_and_set, "does not exists")
 
     def test_ssl(self):
         expected = FakeConfig(
@@ -464,6 +506,34 @@ def test_set_active_context(self):
         self.assertEqual(expected_contexts.get_with_name("ssl").value,
                          loader.current_context)
 
+    def test_ssl_with_relative_ssl_files(self):
+        expected = FakeConfig(
+            host=TEST_SSL_HOST,
+            token=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)
+        )
+        try:
+            temp_dir = tempfile.mkdtemp()
+            actual = FakeConfig()
+            with open(os.path.join(temp_dir, "cert_test"), "wb") as fd:
+                fd.write(TEST_CERTIFICATE_AUTH.encode())
+            with open(os.path.join(temp_dir, "client_cert"), "wb") as fd:
+                fd.write(TEST_CLIENT_CERT.encode())
+            with open(os.path.join(temp_dir, "client_key"), "wb") as fd:
+                fd.write(TEST_CLIENT_KEY.encode())
+            with open(os.path.join(temp_dir, "token_file"), "wb") as fd:
+                fd.write(TEST_DATA.encode())
+            KubeConfigLoader(
+                config_dict=self.TEST_KUBE_CONFIG,
+                active_context="ssl-local-file",
+                config_base_path=temp_dir,
+                client_configuration=actual).load_and_set()
+            self.assertEqual(expected, actual)
+        finally:
+            shutil.rmtree(temp_dir)
+
 
 if __name__ == '__main__':
     unittest.main()