fix: set minimum TLS version to v1.2 (#139)

This commit introduces a custom HTTP adapter that wraps the original one
and adds an SSL context that sets the minimum TLS version required to v1.2.

Author: pyrooka

ibm_cloud_sdk_core/base_service.py

@@ -24,15 +24,15 @@
 from urllib3.util.retry import Retry
 
 import requests
-from requests.adapters import HTTPAdapter
 from requests.structures import CaseInsensitiveDict
 
 from ibm_cloud_sdk_core.authenticators import Authenticator
 from .api_exception import ApiException
 from .detailed_response import DetailedResponse
 from .token_managers.token_manager import TokenManager
 from .utils import (has_bad_first_or_last_char, remove_null_values,
-                    cleanup_values, read_external_sources, strip_extra_slashes)
+                    cleanup_values, read_external_sources, strip_extra_slashes,
+                    SSLHTTPAdapter)
 from .version import __version__
 
 # Uncomment this to enable http debugging
@@ -94,12 +94,15 @@ def __init__(self,
         self.enable_gzip_compression = enable_gzip_compression
         self._set_user_agent_header(self._build_user_agent())
         self.retry_config = None
-        self.http_adapter = HTTPAdapter()
+        self.http_adapter = SSLHTTPAdapter()
         if not self.authenticator:
             raise ValueError('authenticator must be provided')
         if not isinstance(self.authenticator, Authenticator):
             raise ValueError('authenticator should be of type Authenticator')
 
+        self.http_client.mount('http://', self.http_adapter)
+        self.http_client.mount('https://', self.http_adapter)
+
     def enable_retries(self, max_retries: int = 4, retry_interval: float = 1.0) -> None:
         """Enable automatic retries on the underlying http client used by the BaseService instance.
 
@@ -121,14 +124,14 @@ def enable_retries(self, max_retries: int = 4, retry_interval: float = 1.0) -> N
             allowed_methods=['HEAD', 'GET', 'PUT',
                              'DELETE', 'OPTIONS', 'TRACE', 'POST']
         )
-        self.http_adapter = HTTPAdapter(max_retries=self.retry_config)
+        self.http_adapter = SSLHTTPAdapter(max_retries=self.retry_config)
         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 = HTTPAdapter()
+        self.http_adapter = SSLHTTPAdapter()
         self.http_client.mount('http://', self.http_adapter)
         self.http_client.mount('https://', self.http_adapter)
 

ibm_cloud_sdk_core/utils.py

@@ -16,14 +16,33 @@
 # from ibm_cloud_sdk_core.authenticators import Authenticator
 import datetime
 import json as json_import
+import ssl
 from os import getenv, environ, getcwd
 from os.path import isfile, join, expanduser
 from typing import List, Union
 from urllib.parse import urlparse, parse_qs
 
+from requests.adapters import HTTPAdapter
+from urllib3.util.ssl_ import create_urllib3_context
+
 import dateutil.parser as date_parser
 
 
+class SSLHTTPAdapter(HTTPAdapter):
+    """Wraps the original HTTP adapter and adds additional SSL context.
+    """
+    def __init__(self, *args, **kwargs):
+        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.
+        """
+        ssl_context = create_urllib3_context()
+        ssl_context.minimum_version = ssl.TLSVersion.TLSv1_2
+        super().init_poolmanager(connections, maxsize, block, ssl_context=ssl_context)
+
+
 def has_bad_first_or_last_char(val: str) -> bool:
     """Returns true if a string starts with any of: {," ; or ends with any of: },".
 

test/test_base_service.py

@@ -3,6 +3,7 @@
 import gzip
 import json
 import os
+import ssl
 import tempfile
 import time
 from shutil import copyfile
@@ -923,3 +924,11 @@ def test_configure_service_error():
     with pytest.raises(ValueError) as err:
         service.configure_service(None)
     assert str(err.value) == 'Service_name must be of type string.'
+
+
+def test_min_ssl_version():
+    service = AnyServiceV1('2022-03-08', authenticator=NoAuthAuthenticator())
+    adapter = service.http_client.get_adapter('https://')
+    ssl_context = adapter.poolmanager.connection_pool_kw.get('ssl_context', None)
+    assert ssl_context is not None
+    assert ssl_context.minimum_version == ssl.TLSVersion.TLSv1_2