Add support for TLS 1.3 to all HTTPSConnection implementations

CHANGES.rst

@@ -6,6 +6,8 @@ dev (master)
 
 * Implemented a more efficient ``HTTPResponse.__iter__()`` method (Issue #1483)
 
+* Added support for TLSv1.3 support for all ``HTTPSConnection`` implementations. (PR #1496)
+
 * ... [Short description of non-trivial change.] (Issue #)
 
 

dummyserver/testcase.py

@@ -115,13 +115,16 @@ class HTTPDummyServerTestCase(unittest.TestCase):
     scheme = 'http'
     host = 'localhost'
     host_alt = '127.0.0.1'  # Some tests need two hosts
-    certs = DEFAULT_CERTS
+
+    @classmethod
+    def certs(cls):
+        return DEFAULT_CERTS
 
     @classmethod
     def _start_server(cls):
         cls.io_loop = ioloop.IOLoop.current()
         app = web.Application([(r".*", TestingApp)])
-        cls.server, cls.port = run_tornado_app(app, cls.io_loop, cls.certs,
+        cls.server, cls.port = run_tornado_app(app, cls.io_loop, cls.certs(),
                                                cls.scheme, cls.host)
         cls.server_thread = run_loop_in_thread(cls.io_loop)
 
@@ -143,7 +146,10 @@ def tearDownClass(cls):
 class HTTPSDummyServerTestCase(HTTPDummyServerTestCase):
     scheme = 'https'
     host = 'localhost'
-    certs = DEFAULT_CERTS
+
+    @classmethod
+    def certs(cls):
+        return DEFAULT_CERTS
 
 
 @pytest.mark.skipif(not HAS_IPV6, reason='IPv6 not available')

src/urllib3/contrib/_securetransport/bindings.py

@@ -516,6 +516,8 @@ class SecurityConst(object):
     kTLSProtocol1 = 4
     kTLSProtocol11 = 7
     kTLSProtocol12 = 8
+    kTLSProtocol13 = 10
+    kTLSProtocolMaxSupported = 999
 
     kSSLClientSide = 1
     kSSLStreamType = 0

src/urllib3/contrib/pyopenssl.py

@@ -360,6 +360,9 @@ def getpeercert(self, binary_form=False):
             'subjectAltName': get_subj_alt_name(x509)
         }
 
+    def version(self):
+        return self.connection.get_protocol_version_name()
+
     def _reuse(self):
         self._makefile_refs += 1
 

src/urllib3/contrib/securetransport.py

@@ -124,7 +124,7 @@
 # Basically this is simple: for PROTOCOL_SSLv23 we turn it into a low of
 # TLSv1 and a high of TLSv1.2. For everything else, we pin to that version.
 _protocol_to_min_max = {
-    ssl.PROTOCOL_SSLv23: (SecurityConst.kTLSProtocol1, SecurityConst.kTLSProtocol12),
+    ssl.PROTOCOL_SSLv23: (SecurityConst.kTLSProtocol1, SecurityConst.kTLSProtocolMaxSupported),
 }
 
 if hasattr(ssl, "PROTOCOL_SSLv2"):
@@ -147,6 +147,10 @@
     _protocol_to_min_max[ssl.PROTOCOL_TLSv1_2] = (
         SecurityConst.kTLSProtocol12, SecurityConst.kTLSProtocol12
     )
+if hasattr(ssl, "PROTOCOL_TLSv1_3"):
+    _protocol_to_min_max[ssl.PROTOCOL_TLSv1_3] = (
+        SecurityConst.kTLSProtocol13, SecurityConst.kTLSProtocol13
+    )
 if hasattr(ssl, "PROTOCOL_TLS"):
     _protocol_to_min_max[ssl.PROTOCOL_TLS] = _protocol_to_min_max[ssl.PROTOCOL_SSLv23]
 
@@ -667,6 +671,25 @@ def getpeercert(self, binary_form=False):
 
         return der_bytes
 
+    def version(self):
+        protocol = Security.SSLProtocol()
+        result = Security.SSLGetNegotiatedProtocolVersion(self.context, ctypes.byref(protocol))
+        _assert_no_error(result)
+        if protocol == SecurityConst.kTLSProtocol13:
+            return 'TLSv1.3'
+        elif protocol == SecurityConst.kTLSProtocol12:
+            return 'TLSv1.2'
+        elif protocol == SecurityConst.kTLSProtocol11:
+            return 'TLSv1.1'
+        elif protocol == SecurityConst.kTLSProtocol1:
+            return 'TLSv1'
+        elif protocol == SecurityConst.kSSLProtocol3:
+            return 'SSLv3'
+        elif protocol == SecurityConst.kSSLProtocol2:
+            return 'SSLv2'
+        else:
+            raise ssl.SSLError('Unknown TLS version: %r' % protocol)
+
     def _reuse(self):
         self._makefile_refs += 1
 

test/contrib/test_pyopenssl.py

@@ -31,7 +31,11 @@ def teardown_module():
         pass
 
 
-from ..with_dummyserver.test_https import TestHTTPS, TestHTTPS_TLSv1  # noqa: F401
+from ..with_dummyserver.test_https import (  # noqa: F401
+    TestHTTPS, TestHTTPS_TLSv1, TestHTTPS_TLSv1_1,
+    TestHTTPS_TLSv1_2, TestHTTPS_TLSv1_3, TestHTTPS_IPSAN,
+    TestHTTPS_IPv6Addr, TestHTTPS_NoSAN
+)
 from ..with_dummyserver.test_socketlevel import (  # noqa: F401
     TestSNI, TestSocketClosing, TestClientCerts
 )

test/contrib/test_securetransport.py

@@ -27,7 +27,10 @@ def teardown_module():
         pass
 
 
-from ..with_dummyserver.test_https import TestHTTPS, TestHTTPS_TLSv1  # noqa: F401
+from ..with_dummyserver.test_https import (  # noqa: F401
+    TestHTTPS, TestHTTPS_TLSv1, TestHTTPS_TLSv1_1,
+    TestHTTPS_TLSv1_2, TestHTTPS_TLSv1_3
+)
 from ..with_dummyserver.test_socketlevel import (  # noqa: F401
     TestSNI, TestSocketClosing, TestClientCerts
 )

test/with_dummyserver/test_https.py

@@ -72,11 +72,6 @@ def test_dotted_fqdn(self):
         r = pool.request('GET', '/')
         self.assertEqual(r.status, 200, r.data)
 
-    def test_set_ssl_version_to_tlsv1(self):
-        self._pool.ssl_version = ssl.PROTOCOL_TLSv1
-        r = self._pool.request('GET', '/')
-        self.assertEqual(r.status, 200, r.data)
-
     def test_client_intermediate(self):
         client_cert, client_key = (
             DEFAULT_CLIENT_CERTS['certfile'],
@@ -559,30 +554,76 @@ def _request_without_resource_warnings(self, method, url):
         return [x for x in w if not isinstance(x.message, ResourceWarning)]
 
 
-class TestHTTPS_TLSv1(HTTPSDummyServerTestCase):
-    certs = DEFAULT_CERTS.copy()
-    certs['ssl_version'] = ssl.PROTOCOL_TLSv1
+class TestHTTPS_TLSVersion(TestHTTPS):
+    tls_protocol_name = None
 
-    def setUp(self):
-        self._pool = HTTPSConnectionPool(self.host, self.port)
-        self.addCleanup(self._pool.close)
+    @classmethod
+    def certs(cls):
+        return pytest.skip('Don\'t run parent version.')
 
-    def test_discards_connection_on_sslerror(self):
-        self._pool.cert_reqs = 'CERT_REQUIRED'
-        with self.assertRaises(MaxRetryError) as cm:
-            self._pool.request('GET', '/', retries=0)
-        self.assertIsInstance(cm.exception.reason, SSLError)
-        self._pool.ca_certs = DEFAULT_CA
-        self._pool.request('GET', '/')
+    def test_set_ssl_version_to_tls_version(self):
+        self._pool.ssl_version = self.certs()['ssl_version']
+        r = self._pool.request('GET', '/')
+        self.assertEqual(r.status, 200, r.data)
 
-    def test_set_cert_default_cert_required(self):
-        conn = VerifiedHTTPSConnection(self.host, self.port)
-        conn.set_cert(ca_certs=DEFAULT_CA)
-        self.assertEqual(conn.cert_reqs, 'CERT_REQUIRED')
+    def test_tls_protocol_name_of_socket(self):
+        conn = self._pool._get_conn()
+        conn.connect()
+
+        if not hasattr(conn.sock, 'version'):
+            pytest.skip('SSLSocket.version() not available')
+
+        self.assertEqual(conn.sock.version(), self.tls_protocol_name)
+
+
+@pytest.mark.skipif(not hasattr(ssl, "PROTOCOL_TLSv1"), reason="Requires TLSv1 support")
+class TestHTTPS_TLSv1(TestHTTPS_TLSVersion):
+    tls_protocol_name = 'TLSv1'
+
+    @classmethod
+    def certs(cls):
+        certs = DEFAULT_CERTS.copy()
+        certs['ssl_version'] = ssl.PROTOCOL_TLSv1
+        return certs
+
+
+@pytest.mark.skipif(not hasattr(ssl, "PROTOCOL_TLSv1_1"), reason="Requires TLSv1.1 support")
+class TestHTTPS_TLSv1_1(TestHTTPS_TLSVersion):
+    tls_protocol_name = 'TLSv1.1'
+
+    @classmethod
+    def certs(cls):
+        certs = DEFAULT_CERTS.copy()
+        certs['ssl_version'] = ssl.PROTOCOL_TLSv1_1
+        return certs
+
+
+@pytest.mark.skipif(not hasattr(ssl, "PROTOCOL_TLSv1_2"), reason="Requires TLSv1.2 support")
+class TestHTTPS_TLSv1_2(TestHTTPS_TLSVersion):
+    tls_protocol_name = 'TLSv1.2'
+
+    @classmethod
+    def certs(cls):
+        certs = DEFAULT_CERTS.copy()
+        certs['ssl_version'] = ssl.PROTOCOL_TLSv1_2
+        return certs
+
+
+@pytest.mark.skipif(not hasattr(ssl, "PROTOCOL_TLSv1_3"), reason="Requires TLSv1.3 support")
+class TestHTTPS_TLSv1_3(TestHTTPS_TLSVersion):
+    tls_protocol_name = 'TLSv1.3'
+
+    @classmethod
+    def certs(cls):
+        certs = DEFAULT_CERTS.copy()
+        certs['ssl_version'] = ssl.PROTOCOL_TLSv1_3
+        return certs
 
 
 class TestHTTPS_NoSAN(HTTPSDummyServerTestCase):
-    certs = NO_SAN_CERTS
+    @classmethod
+    def certs(cls):
+        return NO_SAN_CERTS
 
     def test_warning_for_certs_without_a_san(self):
         """Ensure that a warning is raised when the cert from the server has
@@ -598,7 +639,9 @@ def test_warning_for_certs_without_a_san(self):
 
 
 class TestHTTPS_IPSAN(HTTPSDummyServerTestCase):
-    certs = IP_SAN_CERTS
+    @classmethod
+    def certs(cls):
+        return IP_SAN_CERTS
 
     def test_can_validate_ip_san(self):
         """Ensure that urllib3 can validate SANs with IP addresses in them."""
@@ -616,7 +659,9 @@ def test_can_validate_ip_san(self):
 
 
 class TestHTTPS_IPv6Addr(IPV6HTTPSDummyServerTestCase):
-    certs = IPV6_ADDR_CERTS
+    @classmethod
+    def certs(cls):
+        return IPV6_ADDR_CERTS
 
     @pytest.mark.skipif(not HAS_IPV6, reason='Only runs on IPv6 systems')
     def test_strip_square_brackets_before_validating(self):

test/with_dummyserver/test_socketlevel.py

@@ -626,7 +626,7 @@ def socket_handler(listener):
             # First request should fail.
             response = pool.urlopen('GET', '/', retries=0,
                                     preload_content=False,
-                                    timeout=Timeout(connect=1, read=0.001))
+                                    timeout=Timeout(connect=1, read=0.1))
             try:
                 self.assertRaises(ReadTimeoutError, response.read)
             finally: