Add option to disable strict X509 verification

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.

kubernetes/client/rest.py

@@ -124,13 +124,33 @@ 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,
             "cert_file": configuration.cert_file,
             "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

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.