Load system CA certificates by default on empty HTTPSConnection
MODIFY
CHANGES.rst
CHANGES.rst
@@ -4,6 +4,10 @@ Changes
dev (master)
------------
+* Change ``HTTPSConnection`` to load system CA certificates
+ when ``ca_certs``, ``ca_cert_dir``, and ``ssl_context`` are
+ unspecified. (Pull #1608, Issue #1603)
+
* Upgrade bundled rfc3986 to 1.3.2. (Pull #1609, Issue #1605)

MODIFY
src/urllib3/connection.py
src/urllib3/connection.py
@@ -249,12 +249,23 @@ class HTTPSConnection(HTTPConnection):
conn = self._new_conn()
self._prepare_conn(conn)
+ # Wrap socket using verification with the root certs in
+ # trusted_root_certs
+ default_ssl_context = False
if self.ssl_context is None:
+ default_ssl_context = True
self.ssl_context = create_urllib3_context(
- ssl_version=resolve_ssl_version(None),
- cert_reqs=resolve_cert_reqs(None),
+ ssl_version=resolve_ssl_version(self.ssl_version),
+ cert_reqs=resolve_cert_reqs(self.cert_reqs),
)
+ # Try to load OS default certs if none are given.
+ # Works well on Windows (requires Python3.4+)
+ context = self.ssl_context
+ if (not self.ca_certs and not self.ca_cert_dir and default_ssl_context
+ and hasattr(context, 'load_default_certs')):
+ context.load_default_certs()
+
self.sock = ssl_wrap_socket(
sock=conn,
keyfile=self.key_file,
@@ -331,7 +342,9 @@ class VerifiedHTTPSConnection(HTTPSConnection):
# Wrap socket using verification with the root certs in
# trusted_root_certs
+ default_ssl_context = False
if self.ssl_context is None:
+ default_ssl_context = True
self.ssl_context = create_urllib3_context(
ssl_version=resolve_ssl_version(self.ssl_version),
cert_reqs=resolve_cert_reqs(self.cert_reqs),
@@ -339,6 +352,13 @@ class VerifiedHTTPSConnection(HTTPSConnection):
context = self.ssl_context
context.verify_mode = resolve_cert_reqs(self.cert_reqs)
+
+ # Try to load OS default certs if none are given.
+ # Works well on Windows (requires Python3.4+)
+ if (not self.ca_certs and not self.ca_cert_dir and default_ssl_context
+ and hasattr(context, 'load_default_certs')):
+ context.load_default_certs()
+
self.sock = ssl_wrap_socket(
sock=conn,
keyfile=self.key_file,

MODIFY
test/with_dummyserver/test_socketlevel.py
test/with_dummyserver/test_socketlevel.py
@@ -35,6 +35,7 @@ from threading import Event
import select
import socket
import ssl
+import mock
import pytest
@@ -1184,6 +1185,83 @@ class TestSSL(SocketDummyServerTestCase):
response = pool.urlopen('GET', '/', retries=1)
self.assertEqual(response.data, b'Success')
+ def test_ssl_load_default_certs_when_empty(self):
+ def socket_handler(listener):
+ sock = listener.accept()[0]
+ ssl_sock = ssl.wrap_socket(sock,
+ server_side=True,
+ keyfile=DEFAULT_CERTS['keyfile'],
+ certfile=DEFAULT_CERTS['certfile'],
+ ca_certs=DEFAULT_CA)
+
+ buf = b''
+ while not buf.endswith(b'\r\n\r\n'):
+ buf += ssl_sock.recv(65536)
+
+ ssl_sock.send(b'HTTP/1.1 200 OK\r\n'
+ b'Content-Type: text/plain\r\n'
+ b'Content-Length: 5\r\n\r\n'
+ b'Hello')
+
+ ssl_sock.close()
+ sock.close()
+
+ context = mock.create_autospec(ssl_.SSLContext)
+ context.load_default_certs = mock.Mock()
+ context.options = 0
+
+ with mock.patch("urllib3.util.ssl_.SSLContext", lambda *_, **__: context):
+
+ self._start_server(socket_handler)
+ pool = HTTPSConnectionPool(self.host, self.port)
+ self.addCleanup(pool.close)
+
+ with self.assertRaises(MaxRetryError):
+ pool.request("GET", "/", timeout=0.01)
+
+ context.load_default_certs.assert_called_with()
+
+ def test_ssl_dont_load_default_certs_when_given(self):
+ def socket_handler(listener):
+ sock = listener.accept()[0]
+ ssl_sock = ssl.wrap_socket(sock,
+ server_side=True,
+ keyfile=DEFAULT_CERTS['keyfile'],
+ certfile=DEFAULT_CERTS['certfile'],
+ ca_certs=DEFAULT_CA)
+
+ buf = b''
+ while not buf.endswith(b'\r\n\r\n'):
+ buf += ssl_sock.recv(65536)
+
+ ssl_sock.send(b'HTTP/1.1 200 OK\r\n'
+ b'Content-Type: text/plain\r\n'
+ b'Content-Length: 5\r\n\r\n'
+ b'Hello')
+
+ ssl_sock.close()
+ sock.close()
+
+ context = mock.create_autospec(ssl_.SSLContext)
+ context.load_default_certs = mock.Mock()
+ context.options = 0
+
+ with mock.patch("urllib3.util.ssl_.SSLContext", lambda *_, **__: context):
+ for kwargs in [{"ca_certs": "/a"},
+ {"ca_cert_dir": "/a"},
+ {"ca_certs": "a", "ca_cert_dir": "a"},
+ {"ssl_context": context}]:
+
+ self._start_server(socket_handler)
+
+ pool = HTTPSConnectionPool(self.host, self.port, **kwargs)
+ self.addCleanup(pool.close)
+
+ with self.assertRaises(MaxRetryError):
+ pool.request("GET", "/", timeout=0.01)
+
+ context.load_default_certs.assert_not_called()
+
class TestErrorWrapping(SocketDummyServerTestCase):

Load system CA certificates by default on empty HTTPSConnection (#1608)