From 9601fbd5a8055484d8cdf39a96a7808700127f0c Mon Sep 17 00:00:00 2001 From: Vishwa Date: Fri, 11 Sep 2026 21:55:27 +0530 Subject: [PATCH] Add option to disable strict X509 verification --- kubernetes/client/configuration.py | 1 + kubernetes/client/rest.py | 8 ++++++++ kubernetes/test/test_api_client.py | 15 +++++++++++++++ 3 files changed, 24 insertions(+) diff --git a/kubernetes/client/configuration.py b/kubernetes/client/configuration.py index d3a0009cf0..ffa81b0ae2 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 1a1015d75e..697a5fa0f7 100644 --- a/kubernetes/client/rest.py +++ b/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 diff --git a/kubernetes/test/test_api_client.py b/kubernetes/test/test_api_client.py index 91cbe45bbc..d83b43e9e2 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 @@ -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)