diff --git a/kubernetes/client/configuration.py b/kubernetes/client/configuration.py index 8d532264c7..5d20443330 100644 --- a/kubernetes/client/configuration.py +++ b/kubernetes/client/configuration.py @@ -306,6 +306,7 @@ def __init__( """ self.verify_ssl = verify_ssl + self.disable_strict_ssl_verification = False """SSL/TLS verification Set this to false to skip verifying SSL certificate when calling API from https server. diff --git a/kubernetes/client/rest.py b/kubernetes/client/rest.py index 36d471f6bf..b9745493f6 100644 --- a/kubernetes/client/rest.py +++ b/kubernetes/client/rest.py @@ -124,6 +124,24 @@ def __init__(self, configuration) -> None: else: cert_reqs = ssl.CERT_NONE + ssl_context = None + if configuration.disable_strict_ssl_verification: + ssl_context = ssl.create_default_context( + cafile=configuration.ssl_ca_cert, + cadata=configuration.ca_cert_data, + ) + + ssl_context.verify_flags &= ~ssl.VERIFY_X509_STRICT + + if configuration.cert_file: + ssl_context.load_cert_chain( + configuration.cert_file, keyfile=configuration.key_file + ) + + if not configuration.verify_ssl: + ssl_context.check_hostname = False + ssl_context.verify_mode = ssl.CERT_NONE + pool_args = { "cert_reqs": cert_reqs, "ca_certs": configuration.ssl_ca_cert, @@ -131,6 +149,8 @@ def __init__(self, configuration) -> None: "key_file": configuration.key_file, "ca_cert_data": configuration.ca_cert_data, } + if ssl_context is not None: + pool_args["ssl_context"] = ssl_context if configuration.assert_hostname is not None: pool_args['assert_hostname'] = ( configuration.assert_hostname diff --git a/kubernetes/test/test_api_client.py b/kubernetes/test/test_api_client.py index 91cbe45bbc..43fd30e291 100644 --- a/kubernetes/test/test_api_client.py +++ b/kubernetes/test/test_api_client.py @@ -4,6 +4,7 @@ import unittest from unittest import mock import weakref +import ssl import kubernetes from kubernetes.aio.client.configuration import Configuration as AsyncConfiguration @@ -86,6 +87,20 @@ def test_client_go_read_retries_do_not_override_write_retries(self): self.assertNotIn('retries', rest_client.pool_manager.request.call_args.kwargs) + def test_disable_strict_ssl_verification(self): + config = Configuration() + config.disable_strict_ssl_verification = True + + rest_client = RESTClientObject(config) + + self.assertIsNotNone( + rest_client.pool_manager.connection_pool_kw["ssl_context"] + ) + self.assertFalse( + rest_client.pool_manager.connection_pool_kw["ssl_context"].verify_flags + & ssl.VERIFY_X509_STRICT + ) + class TestConfigurationAuthSettings(unittest.TestCase): """Regression tests for Configuration.auth_settings() bearer-token lookup.