Allow the CA certificate data to be passed as a string
MODIFY
CONTRIBUTORS.txt
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]

MODIFY
src/urllib3/connection.py
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 @@ class VerifiedHTTPSConnection(HTTPSConnection):
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 @@ class VerifiedHTTPSConnection(HTTPSConnection):
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 @@ class VerifiedHTTPSConnection(HTTPSConnection):
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 @@ class VerifiedHTTPSConnection(HTTPSConnection):
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,
)

MODIFY
src/urllib3/util/ssl_.py
src/urllib3/util/ssl_.py
@@ -119,12 +119,15 @@ except ImportError:
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

MODIFY
test/test_ssl.py
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):

MODIFY
test/test_util.py
test/test_util.py
@@ -732,7 +732,9 @@ class TestUtil(object):
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 @@ class TestUtil(object):
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):

Allow the CA certificate data to be passed as a string (#1804)