kubernetes-client/python
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

incluster_config loader should fail on empty host/port or cert/token file

Browse Source
This commit is contained in:
mbohlool 2016-11-30 04:11:19 -08:00
parent 66e4f5ff70
commit acc07d720b
2 changed files with 67 additions and 42 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

36
kubernetes/config/incluster_config.py
View File

@ -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()

73
kubernetes/config/incluster_config_test.py
View File

@ -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()