fix: use the correct SSL config if cert verification is disabled

Signed-off-by: Norbert Biczo <[email protected]>

Author: pyrooka

ibm_cloud_sdk_core/base_service.py

@@ -108,7 +108,7 @@ def __init__(
         self.enable_gzip_compression = enable_gzip_compression
         self._set_user_agent_header(self._build_user_agent())
         self.retry_config = None
-        self.http_adapter = SSLHTTPAdapter()
+        self.http_adapter = SSLHTTPAdapter(_disable_ssl_verification=self.disable_ssl_verification)
         if not self.authenticator:
             raise ValueError('authenticator must be provided')
         if not isinstance(self.authenticator, Authenticator):
@@ -138,14 +138,16 @@ def enable_retries(self, max_retries: int = 4, retry_interval: float = 30.0) ->
             # Omitting this will default to all methods except POST
             allowed_methods=['HEAD', 'GET', 'PUT', 'DELETE', 'OPTIONS', 'TRACE', 'POST'],
         )
-        self.http_adapter = SSLHTTPAdapter(max_retries=self.retry_config)
+        self.http_adapter = SSLHTTPAdapter(
+            max_retries=self.retry_config, _disable_ssl_verification=self.disable_ssl_verification
+        )
         self.http_client.mount('http://', self.http_adapter)
         self.http_client.mount('https://', self.http_adapter)
 
     def disable_retries(self):
         """Remove retry config from http_adapter"""
         self.retry_config = None
-        self.http_adapter = SSLHTTPAdapter()
+        self.http_adapter = SSLHTTPAdapter(_disable_ssl_verification=self.disable_ssl_verification)
         self.http_client.mount('http://', self.http_adapter)
         self.http_client.mount('https://', self.http_adapter)
 
@@ -223,8 +225,18 @@ def set_disable_ssl_verification(self, status: bool = False) -> None:
         Keyword Arguments:
             status: set to true to disable ssl verification (default: {False})
         """
+        if self.disable_ssl_verification == status:
+            # Do nothing if the state doesn't change.
+            return
+
         self.disable_ssl_verification = status
 
+        self.http_adapter = SSLHTTPAdapter(
+            max_retries=self.retry_config, _disable_ssl_verification=self.disable_ssl_verification
+        )
+        self.http_client.mount('http://', self.http_adapter)
+        self.http_client.mount('https://', self.http_adapter)
+
     def set_service_url(self, service_url: str) -> None:
         """Set the url the service will make HTTP requests too.
 

ibm_cloud_sdk_core/utils.py

@@ -25,7 +25,7 @@
 from typing import List, Union
 from urllib.parse import urlparse, parse_qs
 
-from requests.adapters import HTTPAdapter
+from requests.adapters import HTTPAdapter, DEFAULT_POOLBLOCK
 from urllib3.util.ssl_ import create_urllib3_context
 
 import dateutil.parser as date_parser
@@ -35,14 +35,21 @@ class SSLHTTPAdapter(HTTPAdapter):
     """Wraps the original HTTP adapter and adds additional SSL context."""
 
     def __init__(self, *args, **kwargs):
+        self._disable_ssl_verification = kwargs.pop('_disable_ssl_verification', None)
+
         super().__init__(*args, **kwargs)
 
-    # pylint: disable=arguments-differ
-    def init_poolmanager(self, connections, maxsize, block):
-        """Extends the parent's method by adding minimum SSL version to the args."""
+    def init_poolmanager(self, connections, maxsize, block=DEFAULT_POOLBLOCK, **pool_kwargs):
+        """Create and use custom SSL configuration."""
+
         ssl_context = create_urllib3_context()
         ssl_context.minimum_version = ssl.TLSVersion.TLSv1_2
-        super().init_poolmanager(connections, maxsize, block, ssl_context=ssl_context)
+
+        if self._disable_ssl_verification:
+            ssl_context.check_hostname = False
+            ssl_context.verify_mode = ssl.CERT_NONE
+
+        super().init_poolmanager(connections, maxsize, block, ssl_context=ssl_context, **pool_kwargs)
 
 
 class GzipStream(io.RawIOBase):
@@ -60,7 +67,7 @@ class GzipStream(io.RawIOBase):
                It can be a file-like object, bytes or string.
     """
 
-    def __init__(self, source: Union[io.IOBase, bytes, str]) -> 'GzipStream':
+    def __init__(self, source: Union[io.IOBase, bytes, str]):
         self.buffer = b''
 
         if isinstance(source, io.IOBase):

test_integration/test_no_ssl_verification.py

@@ -4,8 +4,8 @@
 from http.server import HTTPServer, SimpleHTTPRequestHandler
 from ssl import PROTOCOL_TLS_SERVER, SSLContext
 
-from ibm_cloud_sdk_core.authenticators import NoAuthAuthenticator
 from test.test_base_service import AnyServiceV1
+from ibm_cloud_sdk_core.authenticators import NoAuthAuthenticator
 
 
 def test_no_ssl_verification():
@@ -33,7 +33,7 @@ def test_no_ssl_verification():
     try:
         res = service.send(prepped)
         assert res is not None
-    except Exception:
+    except Exception:  # pylint: disable=try-except-raise
         raise
     finally:
         server.shutdown()