codekingpro/portable-devtools
114k
1# Copyright 2018 The Kubernetes Authors.2#3# Licensed under the Apache License, Version 2.0 (the "License");4# you may not use this file except in compliance with the License.5# You may obtain a copy of the License at6#7# http://www.apache.org/licenses/LICENSE-2.08#9# Unless required by applicable law or agreed to in writing, software10# distributed under the License is distributed on an "AS IS" BASIS,11# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.12# See the License for the specific language governing permissions and13# limitations under the License.14 15import base6416import datetime17import io18import json19import os20from pprint import pprint21import shutil22import tempfile23import unittest24from collections import namedtuple25 26from unittest import mock27import yaml28from six import PY3, next29 30from kubernetes.client import Configuration31 32from .config_exception import ConfigException33from .dateutil import UTC, format_rfc3339, parse_rfc333934from .kube_config import (ENV_KUBECONFIG_PATH_SEPARATOR, CommandTokenSource,35 ConfigNode, FileOrData, KubeConfigLoader,36 KubeConfigMerger, _cleanup_temp_files,37 _create_temp_file_with_content,38 _get_kube_config_loader,39 _get_kube_config_loader_for_yaml_file,40 list_kube_config_contexts, load_kube_config,41 load_kube_config_from_dict, new_client_from_config, new_client_from_config_dict)42 43BEARER_TOKEN_FORMAT = "Bearer %s"44 45EXPIRY_DATETIME_FORMAT = "%Y-%m-%dT%H:%M:%SZ"46# should be less than kube_config.EXPIRY_SKEW_PREVENTION_DELAY47PAST_EXPIRY_TIMEDELTA = 248# should be more than kube_config.EXPIRY_SKEW_PREVENTION_DELAY49FUTURE_EXPIRY_TIMEDELTA = 6050 51NON_EXISTING_FILE = "zz_non_existing_file_472398324"52 53 54def _base64(string):55 return base64.standard_b64encode(string.encode()).decode()56 57 58def _urlsafe_unpadded_b64encode(string):59 return base64.urlsafe_b64encode(string.encode()).decode().rstrip('=')60 61 62def _format_expiry_datetime(dt):63 return dt.strftime(EXPIRY_DATETIME_FORMAT)64 65 66def _get_expiry(loader, active_context):67 expired_gcp_conf = (item for item in loader._config.value.get("users")68 if item.get("name") == active_context)69 return next(expired_gcp_conf).get("user").get("auth-provider") \70 .get("config").get("expiry")71 72 73def _raise_exception(st):74 raise Exception(st)75 76 77TEST_FILE_KEY = "file"78TEST_DATA_KEY = "data"79TEST_FILENAME = "test-filename"80 81TEST_DATA = "test-data"82TEST_DATA_BASE64 = _base64(TEST_DATA)83 84TEST_ANOTHER_DATA = "another-test-data"85TEST_ANOTHER_DATA_BASE64 = _base64(TEST_ANOTHER_DATA)86 87TEST_HOST = "test-host"88TEST_USERNAME = "me"89TEST_PASSWORD = "pass"90# token for me:pass91TEST_BASIC_TOKEN = "Basic bWU6cGFzcw=="92DATETIME_EXPIRY_PAST = datetime.datetime.now(tz=UTC93 ).replace(tzinfo=None) - datetime.timedelta(minutes=PAST_EXPIRY_TIMEDELTA)94DATETIME_EXPIRY_FUTURE = datetime.datetime.now(tz=UTC95 ).replace(tzinfo=None) + datetime.timedelta(minutes=FUTURE_EXPIRY_TIMEDELTA)96TEST_TOKEN_EXPIRY_PAST = _format_expiry_datetime(DATETIME_EXPIRY_PAST)97 98TEST_SSL_HOST = "https://test-host"99TEST_CERTIFICATE_AUTH = "cert-auth"100TEST_CERTIFICATE_AUTH_BASE64 = _base64(TEST_CERTIFICATE_AUTH)101TEST_CLIENT_KEY = "client-key"102TEST_CLIENT_KEY_BASE64 = _base64(TEST_CLIENT_KEY)103TEST_CLIENT_CERT = "client-cert"104TEST_CLIENT_CERT_BASE64 = _base64(TEST_CLIENT_CERT)105TEST_TLS_SERVER_NAME = "kubernetes.io"106 107TEST_OIDC_TOKEN = "test-oidc-token"108TEST_OIDC_INFO = "{\"name\": \"test\"}"109TEST_OIDC_BASE = ".".join([110 _urlsafe_unpadded_b64encode(TEST_OIDC_TOKEN),111 _urlsafe_unpadded_b64encode(TEST_OIDC_INFO)112])113TEST_OIDC_LOGIN = ".".join([114 TEST_OIDC_BASE,115 _urlsafe_unpadded_b64encode(TEST_CLIENT_CERT_BASE64)116])117TEST_OIDC_TOKEN = "Bearer %s" % TEST_OIDC_LOGIN118TEST_OIDC_EXP = "{\"name\": \"test\",\"exp\": 536457600}"119TEST_OIDC_EXP_BASE = _urlsafe_unpadded_b64encode(120 TEST_OIDC_TOKEN) + "." + _urlsafe_unpadded_b64encode(TEST_OIDC_EXP)121TEST_OIDC_EXPIRED_LOGIN = ".".join([122 TEST_OIDC_EXP_BASE,123 _urlsafe_unpadded_b64encode(TEST_CLIENT_CERT)124])125TEST_OIDC_CONTAINS_RESERVED_CHARACTERS = ".".join([126 _urlsafe_unpadded_b64encode(TEST_OIDC_TOKEN),127 _urlsafe_unpadded_b64encode(TEST_OIDC_INFO).replace("a", "+"),128 _urlsafe_unpadded_b64encode(TEST_CLIENT_CERT)129])130TEST_OIDC_INVALID_PADDING_LENGTH = ".".join([131 _urlsafe_unpadded_b64encode(TEST_OIDC_TOKEN),132 "aaaaa",133 _urlsafe_unpadded_b64encode(TEST_CLIENT_CERT)134])135 136TEST_OIDC_CA = _base64(TEST_CERTIFICATE_AUTH)137 138TEST_AZURE_LOGIN = TEST_OIDC_LOGIN139TEST_AZURE_TOKEN = "test-azure-token"140TEST_AZURE_TOKEN_FULL = "Bearer " + TEST_AZURE_TOKEN141 142 143class BaseTestCase(unittest.TestCase):144 145 def setUp(self):146 self._temp_files = []147 148 def tearDown(self):149 for f in self._temp_files:150 os.remove(f)151 152 def _create_temp_file(self, content=""):153 handler, name = tempfile.mkstemp()154 self._temp_files.append(name)155 os.write(handler, str.encode(content))156 os.close(handler)157 return name158 159 def expect_exception(self, func, message_part, *args, **kwargs):160 with self.assertRaises(ConfigException) as context:161 func(*args, **kwargs)162 self.assertIn(message_part, str(context.exception))163 164 165class TestFileOrData(BaseTestCase):166 167 @staticmethod168 def get_file_content(filename):169 with open(filename) as f:170 return f.read()171 172 def test_file_given_file(self):173 temp_filename = _create_temp_file_with_content(TEST_DATA)174 obj = {TEST_FILE_KEY: temp_filename}175 t = FileOrData(obj=obj, file_key_name=TEST_FILE_KEY)176 self.assertEqual(TEST_DATA, self.get_file_content(t.as_file()))177 178 def test_file_given_non_existing_file(self):179 temp_filename = NON_EXISTING_FILE180 obj = {TEST_FILE_KEY: temp_filename}181 t = FileOrData(obj=obj, file_key_name=TEST_FILE_KEY)182 self.expect_exception(t.as_file, "does not exist")183 184 def test_file_given_data(self):185 obj = {TEST_DATA_KEY: TEST_DATA_BASE64}186 t = FileOrData(obj=obj, file_key_name=TEST_FILE_KEY,187 data_key_name=TEST_DATA_KEY)188 self.assertEqual(TEST_DATA, self.get_file_content(t.as_file()))189 190 def test_file_given_data_no_base64(self):191 obj = {TEST_DATA_KEY: TEST_DATA}192 t = FileOrData(obj=obj, file_key_name=TEST_FILE_KEY,193 data_key_name=TEST_DATA_KEY, base64_file_content=False)194 self.assertEqual(TEST_DATA, self.get_file_content(t.as_file()))195 196 def test_data_given_data(self):197 obj = {TEST_DATA_KEY: TEST_DATA_BASE64}198 t = FileOrData(obj=obj, file_key_name=TEST_FILE_KEY,199 data_key_name=TEST_DATA_KEY)200 self.assertEqual(TEST_DATA_BASE64, t.as_data())201 202 def test_data_given_file(self):203 obj = {204 TEST_FILE_KEY: self._create_temp_file(content=TEST_DATA)}205 t = FileOrData(obj=obj, file_key_name=TEST_FILE_KEY)206 self.assertEqual(TEST_DATA_BASE64, t.as_data())207 208 def test_data_given_file_no_base64(self):209 obj = {210 TEST_FILE_KEY: self._create_temp_file(content=TEST_DATA)}211 t = FileOrData(obj=obj, file_key_name=TEST_FILE_KEY,212 base64_file_content=False)213 self.assertEqual(TEST_DATA, t.as_data())214 215 def test_data_given_file_and_data(self):216 obj = {217 TEST_DATA_KEY: TEST_DATA_BASE64,218 TEST_FILE_KEY: self._create_temp_file(219 content=TEST_ANOTHER_DATA)}220 t = FileOrData(obj=obj, file_key_name=TEST_FILE_KEY,221 data_key_name=TEST_DATA_KEY)222 self.assertEqual(TEST_DATA_BASE64, t.as_data())223 224 def test_file_given_file_and_data(self):225 obj = {226 TEST_DATA_KEY: TEST_DATA_BASE64,227 TEST_FILE_KEY: self._create_temp_file(228 content=TEST_ANOTHER_DATA)}229 t = FileOrData(obj=obj, file_key_name=TEST_FILE_KEY,230 data_key_name=TEST_DATA_KEY)231 self.assertEqual(TEST_DATA, self.get_file_content(t.as_file()))232 233 def test_file_with_custom_dirname(self):234 tempfile = self._create_temp_file(content=TEST_DATA)235 tempfile_dir = os.path.dirname(tempfile)236 tempfile_basename = os.path.basename(tempfile)237 obj = {TEST_FILE_KEY: tempfile_basename}238 t = FileOrData(obj=obj, file_key_name=TEST_FILE_KEY,239 file_base_path=tempfile_dir)240 self.assertEqual(TEST_DATA, self.get_file_content(t.as_file()))241 242 def test_create_temp_file_with_content(self):243 self.assertEqual(TEST_DATA,244 self.get_file_content(245 _create_temp_file_with_content(TEST_DATA)))246 _cleanup_temp_files()247 248 def test_file_given_data_bytes(self):249 obj = {TEST_DATA_KEY: TEST_DATA_BASE64.encode()}250 t = FileOrData(obj=obj, file_key_name=TEST_FILE_KEY,251 data_key_name=TEST_DATA_KEY)252 self.assertEqual(TEST_DATA, self.get_file_content(t.as_file()))253 254 def test_file_given_data_bytes_no_base64(self):255 obj = {TEST_DATA_KEY: TEST_DATA.encode()}256 t = FileOrData(obj=obj, file_key_name=TEST_FILE_KEY,257 data_key_name=TEST_DATA_KEY, base64_file_content=False)258 self.assertEqual(TEST_DATA, self.get_file_content(t.as_file()))259 260 def test_file_given_no_object(self):261 t = FileOrData(obj=None, file_key_name=TEST_FILE_KEY,262 data_key_name=TEST_DATA_KEY)263 self.assertEqual(t.as_file(), None)264 265 def test_file_given_no_object_data(self):266 t = FileOrData(obj=None, file_key_name=TEST_FILE_KEY,267 data_key_name=TEST_DATA_KEY)268 self.assertEqual(t.as_data(), None)269 270 def test_file_recreation(self):271 obj = {TEST_DATA_KEY: TEST_DATA_BASE64}272 t1 = FileOrData(273 obj=obj,274 file_key_name=TEST_FILE_KEY,275 data_key_name=TEST_DATA_KEY,276 )277 first_file_path = t1.as_file()278 # We manually remove the file from the disk leaving it in the cache279 os.remove(first_file_path)280 281 t2 = FileOrData(282 obj=obj,283 file_key_name=TEST_FILE_KEY,284 data_key_name=TEST_DATA_KEY,285 )286 287 second_file_path = t2.as_file()288 self.assertEqual(TEST_DATA, self.get_file_content(second_file_path))289 290 291class TestConfigNode(BaseTestCase):292 293 test_obj = {"key1": "test", "key2": ["a", "b", "c"],294 "key3": {"inner_key": "inner_value"},295 "with_names": [{"name": "test_name", "value": "test_value"},296 {"name": "test_name2",297 "value": {"key1", "test"}},298 {"name": "test_name3", "value": [1, 2, 3]}],299 "with_names_dup": [300 {"name": "test_name", "value": "test_value"},301 {"name": "test_name",302 "value": {"key1", "test"}},303 {"name": "test_name3", "value": [1, 2, 3]}304 ]}305 306 def setUp(self):307 super(TestConfigNode, self).setUp()308 self.node = ConfigNode("test_obj", self.test_obj)309 310 def test_normal_map_array_operations(self):311 self.assertEqual("test", self.node['key1'])312 self.assertEqual(5, len(self.node))313 314 self.assertEqual("test_obj/key2", self.node['key2'].name)315 self.assertEqual(["a", "b", "c"], self.node['key2'].value)316 self.assertEqual("b", self.node['key2'][1])317 self.assertEqual(3, len(self.node['key2']))318 319 self.assertEqual("test_obj/key3", self.node['key3'].name)320 self.assertEqual({"inner_key": "inner_value"},321 self.node['key3'].value)322 self.assertEqual("inner_value", self.node['key3']["inner_key"])323 self.assertEqual(1, len(self.node['key3']))324 325 def test_get_with_name(self):326 node = self.node["with_names"]327 self.assertEqual(328 "test_value",329 node.get_with_name("test_name")["value"])330 self.assertTrue(331 isinstance(node.get_with_name("test_name2"), ConfigNode))332 self.assertTrue(333 isinstance(node.get_with_name("test_name3"), ConfigNode))334 self.assertEqual("test_obj/with_names[name=test_name2]",335 node.get_with_name("test_name2").name)336 self.assertEqual("test_obj/with_names[name=test_name3]",337 node.get_with_name("test_name3").name)338 339 def test_key_does_not_exists(self):340 self.expect_exception(lambda: self.node['not-exists-key'],341 "Expected key not-exists-key in test_obj")342 self.expect_exception(lambda: self.node['key3']['not-exists-key'],343 "Expected key not-exists-key in test_obj/key3")344 345 def test_get_with_name_on_invalid_object(self):346 self.expect_exception(347 lambda: self.node['key2'].get_with_name('no-name'),348 "Expected all values in test_obj/key2 list to have \'name\' key")349 350 def test_get_with_name_on_non_list_object(self):351 self.expect_exception(352 lambda: self.node['key3'].get_with_name('no-name'),353 "Expected test_obj/key3 to be a list")354 355 def test_get_with_name_on_name_does_not_exists(self):356 self.expect_exception(357 lambda: self.node['with_names'].get_with_name('no-name'),358 "Expected object with name no-name in test_obj/with_names list")359 360 def test_get_with_name_on_duplicate_name(self):361 self.expect_exception(362 lambda: self.node['with_names_dup'].get_with_name('test_name'),363 "Expected only one object with name test_name in "364 "test_obj/with_names_dup list")365 366 367class FakeConfig:368 369 FILE_KEYS = ["ssl_ca_cert", "key_file", "cert_file"]370 IGNORE_KEYS = ["refresh_api_key_hook"]371 372 def __init__(self, token=None, **kwargs):373 self.api_key = {}374 # Provided by the OpenAPI-generated Configuration class375 self.refresh_api_key_hook = None376 if token:377 self.api_key['authorization'] = token378 379 self.__dict__.update(kwargs)380 381 def __eq__(self, other):382 if len(self.__dict__) != len(other.__dict__):383 return384 for k, v in self.__dict__.items():385 if k in self.IGNORE_KEYS:386 continue387 if k not in other.__dict__:388 return389 if k in self.FILE_KEYS:390 if v and other.__dict__[k]:391 try:392 with open(v) as f1, open(other.__dict__[k]) as f2:393 if f1.read() != f2.read():394 return395 except OSError:396 # fall back to only compare filenames in case we are397 # testing the passing of filenames to the config398 if other.__dict__[k] != v:399 return400 else:401 if other.__dict__[k] != v:402 return403 else:404 if other.__dict__[k] != v:405 return406 return True407 408 def __repr__(self):409 rep = "\n"410 for k, v in self.__dict__.items():411 val = v412 if k in self.FILE_KEYS:413 try:414 with open(v) as f:415 val = "FILE: %s" % str.decode(f.read())416 except OSError as e:417 val = "ERROR: %s" % str(e)418 rep += "\t%s: %s\n" % (k, val)419 return "Config(%s\n)" % rep420 421 422class TestKubeConfigLoader(BaseTestCase):423 TEST_KUBE_CONFIG = {424 "current-context": "no_user",425 "contexts": [426 {427 "name": "no_user",428 "context": {429 "cluster": "default"430 }431 },432 {433 "name": "simple_token",434 "context": {435 "cluster": "default",436 "user": "simple_token"437 }438 },439 {440 "name": "gcp",441 "context": {442 "cluster": "default",443 "user": "gcp"444 }445 },446 {447 "name": "expired_gcp",448 "context": {449 "cluster": "default",450 "user": "expired_gcp"451 }452 },453 {454 "name": "expired_gcp_refresh",455 "context": {456 "cluster": "default",457 "user": "expired_gcp_refresh"458 }459 },460 {461 "name": "oidc",462 "context": {463 "cluster": "default",464 "user": "oidc"465 }466 },467 {468 "name": "azure",469 "context": {470 "cluster": "default",471 "user": "azure"472 }473 },474 {475 "name": "azure_num",476 "context": {477 "cluster": "default",478 "user": "azure_num"479 }480 },481 {482 "name": "azure_str",483 "context": {484 "cluster": "default",485 "user": "azure_str"486 }487 },488 {489 "name": "azure_num_error",490 "context": {491 "cluster": "default",492 "user": "azure_str_error"493 }494 },495 {496 "name": "azure_str_error",497 "context": {498 "cluster": "default",499 "user": "azure_str_error"500 }501 },502 {503 "name": "expired_oidc",504 "context": {505 "cluster": "default",506 "user": "expired_oidc"507 }508 },509 {510 "name": "expired_oidc_with_idp_ca_file",511 "context": {512 "cluster": "default",513 "user": "expired_oidc_with_idp_ca_file"514 }515 },516 {517 "name": "expired_oidc_nocert",518 "context": {519 "cluster": "default",520 "user": "expired_oidc_nocert"521 }522 },523 {524 "name": "oidc_contains_reserved_character",525 "context": {526 "cluster": "default",527 "user": "oidc_contains_reserved_character"528 529 }530 },531 {532 "name": "oidc_invalid_padding_length",533 "context": {534 "cluster": "default",535 "user": "oidc_invalid_padding_length"536 537 }538 },539 {540 "name": "user_pass",541 "context": {542 "cluster": "default",543 "user": "user_pass"544 }545 },546 {547 "name": "ssl",548 "context": {549 "cluster": "ssl",550 "user": "ssl"551 }552 },553 {554 "name": "no_ssl_verification",555 "context": {556 "cluster": "no_ssl_verification",557 "user": "ssl"558 }559 },560 {561 "name": "ssl-no_file",562 "context": {563 "cluster": "ssl-no_file",564 "user": "ssl-no_file"565 }566 },567 {568 "name": "ssl-local-file",569 "context": {570 "cluster": "ssl-local-file",571 "user": "ssl-local-file"572 }573 },574 {575 "name": "non_existing_user",576 "context": {577 "cluster": "default",578 "user": "non_existing_user"579 }580 },581 {582 "name": "exec_cred_user",583 "context": {584 "cluster": "default",585 "user": "exec_cred_user"586 }587 },588 {589 "name": "exec_cred_user_certificate",590 "context": {591 "cluster": "ssl",592 "user": "exec_cred_user_certificate"593 }594 },595 {596 "name": "contexttestcmdpath",597 "context": {598 "cluster": "clustertestcmdpath",599 "user": "usertestcmdpath"600 }601 },602 {603 "name": "contexttestcmdpathempty",604 "context": {605 "cluster": "clustertestcmdpath",606 "user": "usertestcmdpathempty"607 }608 },609 {610 "name": "contexttestcmdpathscope",611 "context": {612 "cluster": "clustertestcmdpath",613 "user": "usertestcmdpathscope"614 }615 },616 {617 "name": "tls-server-name",618 "context": {619 "cluster": "tls-server-name",620 "user": "ssl"621 }622 },623 ],624 "clusters": [625 {626 "name": "default",627 "cluster": {628 "server": TEST_HOST629 }630 },631 {632 "name": "ssl-no_file",633 "cluster": {634 "server": TEST_SSL_HOST,635 "certificate-authority": TEST_CERTIFICATE_AUTH,636 }637 },638 {639 "name": "ssl-local-file",640 "cluster": {641 "server": TEST_SSL_HOST,642 "certificate-authority": "cert_test",643 }644 },645 {646 "name": "ssl",647 "cluster": {648 "server": TEST_SSL_HOST,649 "certificate-authority-data":650 TEST_CERTIFICATE_AUTH_BASE64,651 "insecure-skip-tls-verify": False,652 }653 },654 {655 "name": "no_ssl_verification",656 "cluster": {657 "server": TEST_SSL_HOST,658 "insecure-skip-tls-verify": True,659 }660 },661 {662 "name": "clustertestcmdpath",663 "cluster": {}664 },665 {666 "name": "tls-server-name",667 "cluster": {668 "server": TEST_SSL_HOST,669 "certificate-authority-data":670 TEST_CERTIFICATE_AUTH_BASE64,671 "insecure-skip-tls-verify": False,672 "tls-server-name": TEST_TLS_SERVER_NAME,673 }674 },675 ],676 "users": [677 {678 "name": "simple_token",679 "user": {680 "token": TEST_DATA_BASE64,681 "username": TEST_USERNAME, # should be ignored682 "password": TEST_PASSWORD, # should be ignored683 }684 },685 {686 "name": "gcp",687 "user": {688 "auth-provider": {689 "name": "gcp",690 "config": {691 "access-token": TEST_DATA_BASE64,692 }693 },694 "token": TEST_DATA_BASE64, # should be ignored695 "username": TEST_USERNAME, # should be ignored696 "password": TEST_PASSWORD, # should be ignored697 }698 },699 {700 "name": "expired_gcp",701 "user": {702 "auth-provider": {703 "name": "gcp",704 "config": {705 "access-token": TEST_DATA_BASE64,706 "expiry": TEST_TOKEN_EXPIRY_PAST, # always in past707 }708 },709 "token": TEST_DATA_BASE64, # should be ignored710 "username": TEST_USERNAME, # should be ignored711 "password": TEST_PASSWORD, # should be ignored712 }713 },714 # Duplicated from "expired_gcp" so test_load_gcp_token_with_refresh715 # is isolated from test_gcp_get_api_key_with_prefix.716 {717 "name": "expired_gcp_refresh",718 "user": {719 "auth-provider": {720 "name": "gcp",721 "config": {722 "access-token": TEST_DATA_BASE64,723 "expiry": TEST_TOKEN_EXPIRY_PAST, # always in past724 }725 },726 "token": TEST_DATA_BASE64, # should be ignored727 "username": TEST_USERNAME, # should be ignored728 "password": TEST_PASSWORD, # should be ignored729 }730 },731 {732 "name": "oidc",733 "user": {734 "auth-provider": {735 "name": "oidc",736 "config": {737 "id-token": TEST_OIDC_LOGIN738 }739 }740 }741 },742 {743 "name": "azure",744 "user": {745 "auth-provider": {746 "config": {747 "access-token": TEST_AZURE_TOKEN,748 "apiserver-id": "00000002-0000-0000-c000-"749 "000000000000",750 "environment": "AzurePublicCloud",751 "refresh-token": "refreshToken",752 "tenant-id": "9d2ac018-e843-4e14-9e2b-4e0ddac75433"753 },754 "name": "azure"755 }756 }757 },758 {759 "name": "azure_num",760 "user": {761 "auth-provider": {762 "config": {763 "access-token": TEST_AZURE_TOKEN,764 "apiserver-id": "00000002-0000-0000-c000-"765 "000000000000",766 "environment": "AzurePublicCloud",767 "expires-in": "0",768 "expires-on": "156207275",769 "refresh-token": "refreshToken",770 "tenant-id": "9d2ac018-e843-4e14-9e2b-4e0ddac75433"771 },772 "name": "azure"773 }774 }775 },776 {777 "name": "azure_str",778 "user": {779 "auth-provider": {780 "config": {781 "access-token": TEST_AZURE_TOKEN,782 "apiserver-id": "00000002-0000-0000-c000-"783 "000000000000",784 "environment": "AzurePublicCloud",785 "expires-in": "0",786 "expires-on": "2018-10-18 00:52:29.044727",787 "refresh-token": "refreshToken",788 "tenant-id": "9d2ac018-e843-4e14-9e2b-4e0ddac75433"789 },790 "name": "azure"791 }792 }793 },794 {795 "name": "azure_str_error",796 "user": {797 "auth-provider": {798 "config": {799 "access-token": TEST_AZURE_TOKEN,800 "apiserver-id": "00000002-0000-0000-c000-"801 "000000000000",802 "environment": "AzurePublicCloud",803 "expires-in": "0",804 "expires-on": "2018-10-18 00:52",805 "refresh-token": "refreshToken",806 "tenant-id": "9d2ac018-e843-4e14-9e2b-4e0ddac75433"807 },808 "name": "azure"809 }810 }811 },812 {813 "name": "azure_num_error",814 "user": {815 "auth-provider": {816 "config": {817 "access-token": TEST_AZURE_TOKEN,818 "apiserver-id": "00000002-0000-0000-c000-"819 "000000000000",820 "environment": "AzurePublicCloud",821 "expires-in": "0",822 "expires-on": "-1",823 "refresh-token": "refreshToken",824 "tenant-id": "9d2ac018-e843-4e14-9e2b-4e0ddac75433"825 },826 "name": "azure"827 }828 }829 },830 {831 "name": "expired_oidc",832 "user": {833 "auth-provider": {834 "name": "oidc",835 "config": {836 "client-id": "tectonic-kubectl",837 "client-secret": "FAKE_SECRET",838 "id-token": TEST_OIDC_EXPIRED_LOGIN,839 "idp-certificate-authority-data": TEST_OIDC_CA,840 "idp-issuer-url": "https://example.org/identity",841 "refresh-token":842 "lucWJjEhlxZW01cXI3YmVlcYnpxNGhzk"843 }844 }845 }846 },847 {848 "name": "expired_oidc_with_idp_ca_file",849 "user": {850 "auth-provider": {851 "name": "oidc",852 "config": {853 "client-id": "tectonic-kubectl",854 "client-secret": "FAKE_SECRET",855 "id-token": TEST_OIDC_EXPIRED_LOGIN,856 "idp-certificate-authority": TEST_CERTIFICATE_AUTH,857 "idp-issuer-url": "https://example.org/identity",858 "refresh-token":859 "lucWJjEhlxZW01cXI3YmVlcYnpxNGhzk"860 }861 }862 }863 },864 {865 "name": "expired_oidc_nocert",866 "user": {867 "auth-provider": {868 "name": "oidc",869 "config": {870 "client-id": "tectonic-kubectl",871 "client-secret": "FAKE_SECRET",872 "id-token": TEST_OIDC_EXPIRED_LOGIN,873 "idp-issuer-url": "https://example.org/identity",874 "refresh-token":875 "lucWJjEhlxZW01cXI3YmVlcYnpxNGhzk"876 }877 }878 }879 },880 {881 "name": "oidc_contains_reserved_character",882 "user": {883 "auth-provider": {884 "name": "oidc",885 "config": {886 "client-id": "tectonic-kubectl",887 "client-secret": "FAKE_SECRET",888 "id-token": TEST_OIDC_CONTAINS_RESERVED_CHARACTERS,889 "idp-issuer-url": "https://example.org/identity",890 "refresh-token":891 "lucWJjEhlxZW01cXI3YmVlcYnpxNGhzk"892 }893 }894 }895 },896 {897 "name": "oidc_invalid_padding_length",898 "user": {899 "auth-provider": {900 "name": "oidc",901 "config": {902 "client-id": "tectonic-kubectl",903 "client-secret": "FAKE_SECRET",904 "id-token": TEST_OIDC_INVALID_PADDING_LENGTH,905 "idp-issuer-url": "https://example.org/identity",906 "refresh-token":907 "lucWJjEhlxZW01cXI3YmVlcYnpxNGhzk"908 }909 }910 }911 },912 {913 "name": "user_pass",914 "user": {915 "username": TEST_USERNAME, # should be ignored916 "password": TEST_PASSWORD, # should be ignored917 }918 },919 {920 "name": "ssl-no_file",921 "user": {922 "token": TEST_DATA_BASE64,923 "client-certificate": TEST_CLIENT_CERT,924 "client-key": TEST_CLIENT_KEY,925 }926 },927 {928 "name": "ssl-local-file",929 "user": {930 "tokenFile": "token_file",931 "client-certificate": "client_cert",932 "client-key": "client_key",933 }934 },935 {936 "name": "ssl",937 "user": {938 "token": TEST_DATA_BASE64,939 "client-certificate-data": TEST_CLIENT_CERT_BASE64,940 "client-key-data": TEST_CLIENT_KEY_BASE64,941 }942 },943 {944 "name": "exec_cred_user",945 "user": {946 "exec": {947 "apiVersion": "client.authentication.k8s.io/v1beta1",948 "command": "aws-iam-authenticator",949 "args": ["token", "-i", "dummy-cluster"]950 }951 }952 },953 {954 "name": "exec_cred_user_certificate",955 "user": {956 "exec": {957 "apiVersion": "client.authentication.k8s.io/v1beta1",958 "command": "custom-certificate-authenticator",959 "args": []960 }961 }962 },963 {964 "name": "usertestcmdpath",965 "user": {966 "auth-provider": {967 "name": "gcp",968 "config": {969 "cmd-path": "cmdtorun"970 }971 }972 }973 },974 {975 "name": "usertestcmdpathempty",976 "user": {977 "auth-provider": {978 "name": "gcp",979 "config": {980 "cmd-path": ""981 }982 }983 }984 },985 {986 "name": "usertestcmdpathscope",987 "user": {988 "auth-provider": {989 "name": "gcp",990 "config": {991 "cmd-path": "cmd",992 "scopes": "scope"993 }994 }995 }996 }997 ]998 }999 1000 def test_no_user_context(self):1001 expected = FakeConfig(host=TEST_HOST)1002 actual = FakeConfig()1003 KubeConfigLoader(1004 config_dict=self.TEST_KUBE_CONFIG,1005 active_context="no_user").load_and_set(actual)1006 self.assertEqual(expected, actual)1007 1008 def test_simple_token(self):1009 expected = FakeConfig(host=TEST_HOST,1010 token=BEARER_TOKEN_FORMAT % TEST_DATA_BASE64)1011 actual = FakeConfig()1012 KubeConfigLoader(1013 config_dict=self.TEST_KUBE_CONFIG,1014 active_context="simple_token").load_and_set(actual)1015 self.assertEqual(expected, actual)1016 1017 def test_load_user_token(self):1018 loader = KubeConfigLoader(1019 config_dict=self.TEST_KUBE_CONFIG,1020 active_context="simple_token")1021 self.assertTrue(loader._load_user_token())1022 self.assertEqual(BEARER_TOKEN_FORMAT % TEST_DATA_BASE64, loader.token)1023 1024 def test_gcp_no_refresh(self):1025 fake_config = FakeConfig()1026 self.assertIsNone(fake_config.refresh_api_key_hook)1027 KubeConfigLoader(1028 config_dict=self.TEST_KUBE_CONFIG,1029 active_context="gcp",1030 get_google_credentials=lambda: _raise_exception(1031 "SHOULD NOT BE CALLED")).load_and_set(fake_config)1032 # Should now be populated with a gcp token fetcher.1033 self.assertIsNotNone(fake_config.refresh_api_key_hook)1034 self.assertEqual(TEST_HOST, fake_config.host)1035 self.assertEqual(BEARER_TOKEN_FORMAT % TEST_DATA_BASE64,1036 fake_config.api_key['authorization'])1037 1038 def test_load_gcp_token_no_refresh(self):1039 loader = KubeConfigLoader(1040 config_dict=self.TEST_KUBE_CONFIG,1041 active_context="gcp",1042 get_google_credentials=lambda: _raise_exception(1043 "SHOULD NOT BE CALLED"))1044 self.assertTrue(loader._load_auth_provider_token())1045 self.assertEqual(BEARER_TOKEN_FORMAT % TEST_DATA_BASE64,1046 loader.token)1047 1048 def test_load_gcp_token_with_refresh(self):1049 def cred(): return None1050 cred.token = TEST_ANOTHER_DATA_BASE641051 cred.expiry = datetime.datetime.now(tz=UTC).replace(tzinfo=None)1052 1053 loader = KubeConfigLoader(1054 config_dict=self.TEST_KUBE_CONFIG,1055 active_context="expired_gcp",1056 get_google_credentials=lambda: cred)1057 original_expiry = _get_expiry(loader, "expired_gcp")1058 self.assertTrue(loader._load_auth_provider_token())1059 new_expiry = _get_expiry(loader, "expired_gcp")1060 # assert that the configs expiry actually updates1061 self.assertTrue(new_expiry > original_expiry)1062 self.assertEqual(BEARER_TOKEN_FORMAT % TEST_ANOTHER_DATA_BASE64,1063 loader.token)1064 1065 def test_gcp_refresh_api_key_hook(self):1066 class cred_old:1067 token = TEST_DATA_BASE641068 expiry = DATETIME_EXPIRY_PAST1069 1070 class cred_new:1071 token = TEST_ANOTHER_DATA_BASE641072 expiry = DATETIME_EXPIRY_FUTURE1073 fake_config = FakeConfig()1074 _get_google_credentials = mock.Mock()1075 _get_google_credentials.side_effect = [cred_old, cred_new]1076 1077 loader = KubeConfigLoader(1078 config_dict=self.TEST_KUBE_CONFIG,1079 active_context="expired_gcp_refresh",1080 get_google_credentials=_get_google_credentials)1081 loader.load_and_set(fake_config)1082 original_expiry = _get_expiry(loader, "expired_gcp_refresh")1083 # Refresh the GCP token.1084 fake_config.refresh_api_key_hook(fake_config)1085 new_expiry = _get_expiry(loader, "expired_gcp_refresh")1086 1087 self.assertTrue(new_expiry > original_expiry)1088 self.assertEqual(BEARER_TOKEN_FORMAT % TEST_ANOTHER_DATA_BASE64,1089 loader.token)1090 1091 def test_oidc_no_refresh(self):1092 loader = KubeConfigLoader(1093 config_dict=self.TEST_KUBE_CONFIG,1094 active_context="oidc",1095 )1096 self.assertTrue(loader._load_auth_provider_token())1097 self.assertEqual(TEST_OIDC_TOKEN, loader.token)1098 1099 @mock.patch('kubernetes.config.kube_config.OAuth2Session.refresh_token')1100 @mock.patch('kubernetes.config.kube_config.ApiClient.request')1101 def test_oidc_with_refresh(self, mock_ApiClient, mock_OAuth2Session):1102 mock_response = mock.MagicMock()1103 type(mock_response).status = mock.PropertyMock(1104 return_value=2001105 )1106 type(mock_response).data = mock.PropertyMock(1107 return_value=json.dumps({1108 "token_endpoint": "https://example.org/identity/token"1109 })1110 )1111 1112 mock_ApiClient.return_value = mock_response1113 1114 mock_OAuth2Session.return_value = {"id_token": "abc123",1115 "refresh_token": "newtoken123"}1116 1117 loader = KubeConfigLoader(1118 config_dict=self.TEST_KUBE_CONFIG,1119 active_context="expired_oidc",1120 )1121 self.assertTrue(loader._load_auth_provider_token())1122 self.assertEqual("Bearer abc123", loader.token)1123 1124 @mock.patch('kubernetes.config.kube_config.OAuth2Session.refresh_token')1125 @mock.patch('kubernetes.config.kube_config.ApiClient.request')1126 def test_oidc_with_idp_ca_file_refresh(self, mock_ApiClient, mock_OAuth2Session):1127 mock_response = mock.MagicMock()1128 type(mock_response).status = mock.PropertyMock(1129 return_value=2001130 )1131 type(mock_response).data = mock.PropertyMock(1132 return_value=json.dumps({1133 "token_endpoint": "https://example.org/identity/token"1134 })1135 )1136 1137 mock_ApiClient.return_value = mock_response1138 1139 mock_OAuth2Session.return_value = {"id_token": "abc123",1140 "refresh_token": "newtoken123"}1141 1142 loader = KubeConfigLoader(1143 config_dict=self.TEST_KUBE_CONFIG,1144 active_context="expired_oidc_with_idp_ca_file",1145 )1146 1147 self.assertTrue(loader._load_auth_provider_token())1148 self.assertEqual("Bearer abc123", loader.token)1149 1150 @mock.patch('kubernetes.config.kube_config.OAuth2Session.refresh_token')1151 @mock.patch('kubernetes.config.kube_config.ApiClient.request')1152 def test_oidc_with_refresh_nocert(1153 self, mock_ApiClient, mock_OAuth2Session):1154 mock_response = mock.MagicMock()1155 type(mock_response).status = mock.PropertyMock(1156 return_value=2001157 )1158 type(mock_response).data = mock.PropertyMock(1159 return_value=json.dumps({1160 "token_endpoint": "https://example.org/identity/token"1161 })1162 )1163 1164 mock_ApiClient.return_value = mock_response1165 1166 mock_OAuth2Session.return_value = {"id_token": "abc123",1167 "refresh_token": "newtoken123"}1168 1169 loader = KubeConfigLoader(1170 config_dict=self.TEST_KUBE_CONFIG,1171 active_context="expired_oidc_nocert",1172 )1173 self.assertTrue(loader._load_auth_provider_token())1174 self.assertEqual("Bearer abc123", loader.token)1175 1176 def test_oidc_fails_if_contains_reserved_chars(self):1177 loader = KubeConfigLoader(1178 config_dict=self.TEST_KUBE_CONFIG,1179 active_context="oidc_contains_reserved_character",1180 )1181 self.assertEqual(1182 loader._load_oid_token("oidc_contains_reserved_character"),1183 None,1184 )1185 1186 def test_oidc_fails_if_invalid_padding_length(self):1187 loader = KubeConfigLoader(1188 config_dict=self.TEST_KUBE_CONFIG,1189 active_context="oidc_invalid_padding_length",1190 )1191 self.assertEqual(1192 loader._load_oid_token("oidc_invalid_padding_length"),1193 None,1194 )1195 1196 def test_azure_no_refresh(self):1197 loader = KubeConfigLoader(1198 config_dict=self.TEST_KUBE_CONFIG,1199 active_context="azure",1200 )