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

@@ -139,6 +139,11 @@ def __init__(self, configuration) -> None:
             cert_reqs = ssl.CERT_REQUIRED
         else:
             cert_reqs = ssl.CERT_NONE
+        if configuration.disable_strict_ssl_verification:
+            ssl_context = ssl.create_default_context()
+            ssl_context.verify_flags &= ~ssl.VERIFY_X509_STRICT
+        else:
+            ssl_context = None
 
         pool_args = {
             "cert_reqs": cert_reqs,
@@ -147,6 +152,9 @@ 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

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
@@ -13,6 +14,20 @@
 
 
 class TestApiClient(unittest.TestCase):
+    def test_disable_strict_ssl_verification(self):
+        config = Configuration(proxy='', no_proxy='')
+        config.disable_strict_ssl_verification = True
+
+        rest_client = RESTClientObject(config)
+
+        ssl_context = (
+            rest_client.pool_manager.connection_pool_kw["ssl_context"]
+        )
+
+        self.assertIsNotNone(ssl_context)
+        self.assertFalse(
+            ssl_context.verify_flags & ssl.VERIFY_X509_STRICT
+        )
     def test_context_manager_closes_threadpool(self):
         with kubernetes.client.ApiClient() as client:
             pool = weakref.ref(client.pool)