Allow the CA certificate data to be passed as a string.

CONTRIBUTORS.txt

@@ -282,5 +282,8 @@ In chronological order:
   * Remove a spurious TypeError from the exception chain inside
     HTTPConnectionPool._make_request(), also for BaseExceptions.
 
+* Benno Rice <benno@jeamland.net>
+  * Allow cadata parameter to be passed to underlying ``SSLContext.load_verify_locations()``.
+
 * [Your name or handle] <[email or website]>
   * [Brief summary of your changes]

src/urllib3/connection.py

@@ -275,6 +275,7 @@ class VerifiedHTTPSConnection(HTTPSConnection):
     cert_reqs = None
     ca_certs = None
     ca_cert_dir = None
+    ca_cert_data = None
     ssl_version = None
     assert_fingerprint = None
 
@@ -288,6 +289,7 @@ def set_cert(
         assert_hostname=None,
         assert_fingerprint=None,
         ca_cert_dir=None,
+        ca_cert_data=None,
     ):
         """
         This method should only be called once, before the connection is used.
@@ -308,6 +310,7 @@ def set_cert(
         self.assert_fingerprint = assert_fingerprint
         self.ca_certs = ca_certs and os.path.expanduser(ca_certs)
         self.ca_cert_dir = ca_cert_dir and os.path.expanduser(ca_cert_dir)
+        self.ca_cert_data = ca_cert_data
 
     def connect(self):
         # Add certificate verification
@@ -358,6 +361,7 @@ def connect(self):
         if (
             not self.ca_certs
             and not self.ca_cert_dir
+            and not self.ca_cert_data
             and default_ssl_context
             and hasattr(context, "load_default_certs")
         ):
@@ -370,6 +374,7 @@ def connect(self):
             key_password=self.key_password,
             ca_certs=self.ca_certs,
             ca_cert_dir=self.ca_cert_dir,
+            ca_cert_data=self.ca_cert_data,
             server_hostname=server_hostname,
             ssl_context=context,
         )

src/urllib3/util/ssl_.py

@@ -119,12 +119,15 @@ def load_cert_chain(self, certfile, keyfile):
             self.certfile = certfile
             self.keyfile = keyfile
 
-        def load_verify_locations(self, cafile=None, capath=None):
+        def load_verify_locations(self, cafile=None, capath=None, cadata=None):
             self.ca_certs = cafile
 
             if capath is not None:
                 raise SSLError("CA directories not supported in older Pythons")
 
+            if cadata is not None:
+                raise SSLError("CA data not supported in older Pythons")
+
         def set_ciphers(self, cipher_suite):
             self.ciphers = cipher_suite
 
@@ -305,6 +308,7 @@ def ssl_wrap_socket(
     ssl_context=None,
     ca_cert_dir=None,
     key_password=None,
+    ca_cert_data=None,
 ):
     """
     All arguments except for server_hostname, ssl_context, and ca_cert_dir have
@@ -323,6 +327,9 @@ def ssl_wrap_socket(
         SSLContext.load_verify_locations().
     :param key_password:
         Optional password if the keyfile is encrypted.
+    :param ca_cert_data:
+        Optional string containing CA certificates in PEM format suitable for
+        passing as the cadata parameter to SSLContext.load_verify_locations()
     """
     context = ssl_context
     if context is None:
@@ -331,9 +338,9 @@ def ssl_wrap_socket(
         # this code.
         context = create_urllib3_context(ssl_version, cert_reqs, ciphers=ciphers)
 
-    if ca_certs or ca_cert_dir:
+    if ca_certs or ca_cert_dir or ca_cert_data:
         try:
-            context.load_verify_locations(ca_certs, ca_cert_dir)
+            context.load_verify_locations(ca_certs, ca_cert_dir, ca_cert_data)
         except IOError as e:  # Platform-specific: Python 2.7
             raise SSLError(e)
         # Py33 raises FileNotFoundError which subclasses OSError

test/test_ssl.py

@@ -138,7 +138,7 @@ def test_wrap_socket_given_ca_certs_no_load_default_certs(monkeypatch):
     ssl_.ssl_wrap_socket(sock, ca_certs="/tmp/fake-file")
 
     context.load_default_certs.assert_not_called()
-    context.load_verify_locations.assert_called_with("/tmp/fake-file", None)
+    context.load_verify_locations.assert_called_with("/tmp/fake-file", None, None)
 
 
 def test_wrap_socket_default_loads_default_certs(monkeypatch):

test/test_util.py

@@ -732,7 +732,9 @@ def test_ssl_wrap_socket_loads_verify_locations(self):
         socket = object()
         mock_context = Mock()
         ssl_wrap_socket(ssl_context=mock_context, ca_certs="/path/to/pem", sock=socket)
-        mock_context.load_verify_locations.assert_called_once_with("/path/to/pem", None)
+        mock_context.load_verify_locations.assert_called_once_with(
+            "/path/to/pem", None, None
+        )
 
     def test_ssl_wrap_socket_loads_certificate_directories(self):
         socket = object()
@@ -741,7 +743,17 @@ def test_ssl_wrap_socket_loads_certificate_directories(self):
             ssl_context=mock_context, ca_cert_dir="/path/to/pems", sock=socket
         )
         mock_context.load_verify_locations.assert_called_once_with(
-            None, "/path/to/pems"
+            None, "/path/to/pems", None
+        )
+
+    def test_ssl_wrap_socket_loads_certificate_data(self):
+        socket = object()
+        mock_context = Mock()
+        ssl_wrap_socket(
+            ssl_context=mock_context, ca_cert_data="TOTALLY PEM DATA", sock=socket
+        )
+        mock_context.load_verify_locations.assert_called_once_with(
+            None, None, "TOTALLY PEM DATA"
         )
 
     def test_ssl_wrap_socket_with_no_sni_warns(self):