diff --git a/kubernetes/config/incluster_config.py b/kubernetes/config/incluster_config.py index 144e52576..3ba1113f1 100644 --- a/kubernetes/config/incluster_config.py +++ b/kubernetes/config/incluster_config.py @@ -18,10 +18,10 @@ from kubernetes.client import configuration from .config_exception import ConfigException -_SERVICE_HOST_ENV_NAME = "KUBERNETES_SERVICE_HOST" -_SERVICE_PORT_ENV_NAME = "KUBERNETES_SERVICE_PORT" -_SERVICE_TOKEN_FILENAME = "/var/run/secrets/kubernetes.io/serviceaccount/token" -_SERVICE_CERT_FILENAME = "/var/run/secrets/kubernetes.io/serviceaccount/ca.crt" +SERVICE_HOST_ENV_NAME = "KUBERNETES_SERVICE_HOST" +SERVICE_PORT_ENV_NAME = "KUBERNETES_SERVICE_PORT" +SERVICE_TOKEN_FILENAME = "/var/run/secrets/kubernetes.io/serviceaccount/token" +SERVICE_CERT_FILENAME = "/var/run/secrets/kubernetes.io/serviceaccount/ca.crt" def _join_host_port(host, port): @@ -35,10 +35,8 @@ def _join_host_port(host, port): class InClusterConfigLoader(object): - def __init__(self, host_env_name, port_env_name, token_filename, + def __init__(self, token_filename, cert_filename, environ=os.environ): - self._host_env_name = host_env_name - self._port_env_name = port_env_name self._token_filename = token_filename self._cert_filename = cert_filename self._environ = environ @@ -48,24 +46,34 @@ class InClusterConfigLoader(object): self._set_config() def _load_config(self): - if (self._host_env_name not in self._environ or - self._port_env_name not in self._environ): + if (SERVICE_HOST_ENV_NAME not in self._environ or + SERVICE_PORT_ENV_NAME not in self._environ): raise ConfigException("Service host/port is not set.") + if (not self._environ[SERVICE_HOST_ENV_NAME] or + not self._environ[SERVICE_PORT_ENV_NAME]): + raise ConfigException("Service host/port is set but empty.") + self.host = ( - "https://" + _join_host_port(self._environ[self._host_env_name], - self._environ[self._port_env_name])) + "https://" + _join_host_port(self._environ[SERVICE_HOST_ENV_NAME], + self._environ[SERVICE_PORT_ENV_NAME])) if not os.path.isfile(self._token_filename): raise ConfigException("Service token file does not exists.") with open(self._token_filename) as f: self.token = f.read() + if not self.token: + raise ConfigException("Token file exists but empty.") if not os.path.isfile(self._cert_filename): raise ConfigException( "Service certification file does not exists.") + with open(self._cert_filename) as f: + if not f.read(): + raise ConfigException("Cert file exists but empty.") + self.ssl_ca_cert = self._cert_filename def _set_config(self): @@ -79,7 +87,5 @@ def load_incluster_config(): 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 not running in a kubernetes environment.""" - InClusterConfigLoader(host_env_name=_SERVICE_HOST_ENV_NAME, - port_env_name=_SERVICE_PORT_ENV_NAME, - token_filename=_SERVICE_TOKEN_FILENAME, - cert_filename=_SERVICE_CERT_FILENAME).load_and_set() + InClusterConfigLoader(token_filename=SERVICE_TOKEN_FILENAME, + cert_filename=SERVICE_CERT_FILENAME).load_and_set() diff --git a/kubernetes/config/incluster_config_test.py b/kubernetes/config/incluster_config_test.py index a807f164d..622b31b37 100644 --- a/kubernetes/config/incluster_config_test.py +++ b/kubernetes/config/incluster_config_test.py @@ -17,17 +17,21 @@ import tempfile import unittest from .config_exception import ConfigException -from .incluster_config import (_SERVICE_HOST_ENV_NAME, _SERVICE_PORT_ENV_NAME, - InClusterConfigLoader) +from .incluster_config import (SERVICE_HOST_ENV_NAME, SERVICE_PORT_ENV_NAME, + InClusterConfigLoader, _join_host_port) _TEST_TOKEN = "temp_token" +_TEST_CERT = "temp_cert" _TEST_HOST = "127.0.0.1" -_TEST_IPV6_HOST = "::1" _TEST_PORT = "80" -_TEST_ENVIRON = {_SERVICE_HOST_ENV_NAME: _TEST_HOST, - _SERVICE_PORT_ENV_NAME: _TEST_PORT} -_TEST_IPV6_ENVIRON = {_SERVICE_HOST_ENV_NAME: _TEST_IPV6_HOST, - _SERVICE_PORT_ENV_NAME: _TEST_PORT} +_TEST_HOST_PORT = "127.0.0.1:80" +_TEST_IPV6_HOST = "::1" +_TEST_IPV6_HOST_PORT = "[::1]:80" + +_TEST_ENVIRON = {SERVICE_HOST_ENV_NAME: _TEST_HOST, + SERVICE_PORT_ENV_NAME: _TEST_PORT} +_TEST_IPV6_ENVIRON = {SERVICE_HOST_ENV_NAME: _TEST_IPV6_HOST, + SERVICE_PORT_ENV_NAME: _TEST_PORT} class InClusterConfigTest(unittest.TestCase): @@ -48,38 +52,29 @@ class InClusterConfigTest(unittest.TestCase): def get_test_loader( self, - host_env_name=_SERVICE_HOST_ENV_NAME, - port_env_name=_SERVICE_PORT_ENV_NAME, token_filename=None, cert_filename=None, environ=_TEST_ENVIRON): if not token_filename: token_filename = self._create_file_with_temp_content(_TEST_TOKEN) if not cert_filename: - cert_filename = self._create_file_with_temp_content() + cert_filename = self._create_file_with_temp_content(_TEST_CERT) return InClusterConfigLoader( - host_env_name=host_env_name, - port_env_name=port_env_name, token_filename=token_filename, cert_filename=cert_filename, environ=environ) + def test_join_host_port(self): + self.assertEqual(_TEST_HOST_PORT, + _join_host_port(_TEST_HOST, _TEST_PORT)) + self.assertEqual(_TEST_IPV6_HOST_PORT, + _join_host_port(_TEST_IPV6_HOST, _TEST_PORT)) + def test_load_config(self): - cert_filename = self._create_file_with_temp_content() + cert_filename = self._create_file_with_temp_content(_TEST_CERT) loader = self.get_test_loader(cert_filename=cert_filename) loader._load_config() - self.assertEqual("https://%s:%s" % (_TEST_HOST, _TEST_PORT), - loader.host) - self.assertEqual(cert_filename, loader.ssl_ca_cert) - self.assertEqual(_TEST_TOKEN, loader.token) - - def test_load_config_with_bracketed_hostname(self): - cert_filename = self._create_file_with_temp_content() - loader = self.get_test_loader(cert_filename=cert_filename, - environ=_TEST_IPV6_ENVIRON) - loader._load_config() - self.assertEqual("https://[%s]:%s" % (_TEST_IPV6_HOST, _TEST_PORT), - loader.host) + self.assertEqual("https://" + _TEST_HOST_PORT, loader.host) self.assertEqual(cert_filename, loader.ssl_ca_cert) self.assertEqual(_TEST_TOKEN, loader.token) @@ -92,21 +87,45 @@ class InClusterConfigTest(unittest.TestCase): pass def test_no_port(self): - loader = self.get_test_loader(port_env_name="not_exists_port") + loader = self.get_test_loader( + environ={SERVICE_HOST_ENV_NAME: _TEST_HOST}) self._should_fail_load(loader, "no port specified") + def test_empty_port(self): + loader = self.get_test_loader( + environ={SERVICE_HOST_ENV_NAME: _TEST_HOST, + SERVICE_PORT_ENV_NAME: ""}) + self._should_fail_load(loader, "empty port specified") + def test_no_host(self): - loader = self.get_test_loader(host_env_name="not_exists_host") + loader = self.get_test_loader( + environ={SERVICE_PORT_ENV_NAME: _TEST_PORT}) self._should_fail_load(loader, "no host specified") + def test_empty_host(self): + loader = self.get_test_loader( + environ={SERVICE_HOST_ENV_NAME: "", + SERVICE_PORT_ENV_NAME: _TEST_PORT}) + self._should_fail_load(loader, "empty host specified") + def test_no_cert_file(self): loader = self.get_test_loader(cert_filename="not_exists_file_1123") self._should_fail_load(loader, "cert file does not exists") + def test_empty_cert_file(self): + loader = self.get_test_loader( + cert_filename=self._create_file_with_temp_content()) + self._should_fail_load(loader, "empty cert file provided") + def test_no_token_file(self): loader = self.get_test_loader(token_filename="not_exists_file_1123") self._should_fail_load(loader, "token file does not exists") + def test_empty_token_file(self): + loader = self.get_test_loader( + token_filename=self._create_file_with_temp_content()) + self._should_fail_load(loader, "empty token file provided") + if __name__ == '__main__': unittest.main()