Add ssl_handshake_timeout parameter to loop connection/server APIs

As part of this commit, fix error propagation in SSLProtocol
implementation.

Author: 1st1

tests/test_tcp.py

@@ -318,6 +318,16 @@ async def test():
 
         self.loop.run_until_complete(test())
 
+    def test_create_server_8(self):
+        if self.implementation == 'asyncio' and not self.PY37:
+            raise unittest.SkipTest()
+
+        with self.assertRaisesRegex(
+                ValueError, 'ssl_handshake_timeout is only meaningful'):
+            self.loop.run_until_complete(
+                self.loop.create_server(
+                    lambda: None, host='::', port=0, ssl_handshake_timeout=10))
+
     def test_create_connection_open_con_addr(self):
         async def client(addr):
             reader, writer = await asyncio.open_connection(
@@ -501,6 +511,16 @@ async def client(addr):
                              backlog=1) as srv:
             self.loop.run_until_complete(client(srv.addr))
 
+    def test_create_connection_6(self):
+        if self.implementation == 'asyncio' and not self.PY37:
+            raise unittest.SkipTest()
+
+        with self.assertRaisesRegex(
+                ValueError, 'ssl_handshake_timeout is only meaningful'):
+            self.loop.run_until_complete(
+                self.loop.create_connection(
+                    lambda: None, host='::', port=0, ssl_handshake_timeout=10))
+
     def test_transport_shutdown(self):
         CNT = 0           # number of clients that were successful
         TOTAL_CNT = 100   # total number of clients that test will create
@@ -855,6 +875,17 @@ async def runner():
         srv.close()
         self.loop.run_until_complete(srv.wait_closed())
 
+    def test_connect_accepted_socket_ssl_args(self):
+        if self.implementation == 'asyncio' and not self.PY37:
+            raise unittest.SkipTest()
+
+        with self.assertRaisesRegex(
+                ValueError, 'ssl_handshake_timeout is only meaningful'):
+            with socket.socket() as s:
+                self.loop.run_until_complete(
+                    self.loop.connect_accepted_socket(
+                        (lambda: None), s, ssl_handshake_timeout=10.0))
+
     def test_connect_accepted_socket(self, server_ssl=None, client_ssl=None):
         loop = self.loop
 
@@ -898,9 +929,15 @@ def client():
         conn, _ = lsock.accept()
         proto = MyProto(loop=loop)
         proto.loop = loop
+
+        extras = {}
+        if server_ssl and (self.implementation != 'asyncio' or self.PY37):
+            extras = dict(ssl_handshake_timeout=10.0)
+
         f = loop.create_task(
             loop.connect_accepted_socket(
-                (lambda: proto), conn, ssl=server_ssl))
+                (lambda: proto), conn, ssl=server_ssl,
+                **extras))
         loop.run_forever()
         conn.close()
         lsock.close()
@@ -1017,12 +1054,17 @@ def prog(sock):
             await fut
 
         async def start_server():
+            extras = {}
+            if self.implementation != 'asyncio' or self.PY37:
+                extras = dict(ssl_handshake_timeout=10.0)
+
             srv = await asyncio.start_server(
                 handle_client,
                 '127.0.0.1', 0,
                 family=socket.AF_INET,
                 ssl=sslctx,
-                loop=self.loop)
+                loop=self.loop,
+                **extras)
 
             try:
                 srv_socks = srv.sockets
@@ -1080,11 +1122,16 @@ def server(sock):
             sock.close()
 
         async def client(addr):
+            extras = {}
+            if self.implementation != 'asyncio' or self.PY37:
+                extras = dict(ssl_handshake_timeout=10.0)
+
             reader, writer = await asyncio.open_connection(
                 *addr,
                 ssl=client_sslctx,
                 server_hostname='',
-                loop=self.loop)
+                loop=self.loop,
+                **extras)
 
             writer.write(A_DATA)
             self.assertEqual(await reader.readexactly(2), b'OK')
@@ -1140,6 +1187,77 @@ def run(coro):
         with self._silence_eof_received_warning():
             run(client_sock)
 
+    def test_create_connection_ssl_slow_handshake(self):
+        if self.implementation == 'asyncio':
+            raise unittest.SkipTest()
+
+        client_sslctx = self._create_client_ssl_context()
+
+        # silence error logger
+        self.loop.set_exception_handler(lambda *args: None)
+
+        def server(sock):
+            try:
+                sock.recv_all(1024 * 1024)
+            except ConnectionAbortedError:
+                pass
+            finally:
+                sock.close()
+
+        async def client(addr):
+            reader, writer = await asyncio.open_connection(
+                *addr,
+                ssl=client_sslctx,
+                server_hostname='',
+                loop=self.loop,
+                ssl_handshake_timeout=1.0)
+
+        with self.tcp_server(server,
+                             max_clients=1,
+                             backlog=1) as srv:
+
+            with self.assertRaisesRegex(
+                    ConnectionAbortedError,
+                    r'SSL handshake.*is taking longer'):
+
+                self.loop.run_until_complete(client(srv.addr))
+
+    def test_create_connection_ssl_failed_certificate(self):
+        if self.implementation == 'asyncio':
+            raise unittest.SkipTest()
+
+        # silence error logger
+        self.loop.set_exception_handler(lambda *args: None)
+
+        sslctx = self._create_server_ssl_context(self.ONLYCERT, self.ONLYKEY)
+        client_sslctx = self._create_client_ssl_context(disable_verify=False)
+
+        def server(sock):
+            try:
+                sock.starttls(
+                    sslctx,
+                    server_side=True)
+                sock.connect()
+            except ssl.SSLError:
+                pass
+            finally:
+                sock.close()
+
+        async def client(addr):
+            reader, writer = await asyncio.open_connection(
+                *addr,
+                ssl=client_sslctx,
+                server_hostname='',
+                loop=self.loop,
+                ssl_handshake_timeout=1.0)
+
+        with self.tcp_server(server,
+                             max_clients=1,
+                             backlog=1) as srv:
+
+            with self.assertRaises(ssl.SSLCertVerificationError):
+                self.loop.run_until_complete(client(srv.addr))
+
     def test_ssl_connect_accepted_socket(self):
         server_context = ssl.SSLContext(ssl.PROTOCOL_SSLv23)
         server_context.load_cert_chain(self.ONLYCERT, self.ONLYKEY)

tests/test_unix.py

@@ -155,6 +155,16 @@ def test_create_unix_server_2(self):
                 self.loop.run_until_complete(
                     self.loop.create_unix_server(object, sock_name))
 
+    def test_create_unix_server_3(self):
+        if self.implementation == 'asyncio' and not self.PY37:
+            raise unittest.SkipTest()
+
+        with self.assertRaisesRegex(
+                ValueError, 'ssl_handshake_timeout is only meaningful'):
+            self.loop.run_until_complete(
+                self.loop.create_unix_server(
+                    lambda: None, path='/tmp/a', ssl_handshake_timeout=10))
+
     def test_create_unix_server_existing_path_sock(self):
         with self.unix_sock_name() as path:
             sock = socket.socket(socket.AF_UNIX)
@@ -363,7 +373,18 @@ async def client():
         self.assertIn(excs[0].__class__,
                       (BrokenPipeError, ConnectionResetError))
 
-    @unittest.skipUnless(sys.version_info < (3, 7), 'Python version must be < 3.7')
+    def test_create_unix_connection_6(self):
+        if self.implementation == 'asyncio' and not self.PY37:
+            raise unittest.SkipTest()
+
+        with self.assertRaisesRegex(
+                ValueError, 'ssl_handshake_timeout is only meaningful'):
+            self.loop.run_until_complete(
+                self.loop.create_unix_connection(
+                    lambda: None, path='/tmp/a', ssl_handshake_timeout=10))
+
+    @unittest.skipUnless(sys.version_info < (3, 7),
+                         'Python version must be < 3.7')
     def test_transport_unclosed_warning(self):
         async def test(sock):
             return await self.loop.create_unix_connection(
@@ -523,14 +544,19 @@ def prog(sock):
             await fut
 
         async def start_server():
+            extras = {}
+            if self.implementation != 'asyncio' or self.PY37:
+                extras = dict(ssl_handshake_timeout=10.0)
+
             with tempfile.TemporaryDirectory() as td:
                 sock_name = os.path.join(td, 'sock')
 
                 srv = await asyncio.start_unix_server(
                     handle_client,
                     sock_name,
                     ssl=sslctx,
-                    loop=self.loop)
+                    loop=self.loop,
+                    **extras)
 
                 try:
                     tasks = []
@@ -584,11 +610,16 @@ def server(sock):
             sock.close()
 
         async def client(addr):
+            extras = {}
+            if self.implementation != 'asyncio' or self.PY37:
+                extras = dict(ssl_handshake_timeout=10.0)
+
             reader, writer = await asyncio.open_unix_connection(
                 addr,
                 ssl=client_sslctx,
                 server_hostname='',
-                loop=self.loop)
+                loop=self.loop,
+                **extras)
 
             writer.write(A_DATA)
             self.assertEqual(await reader.readexactly(2), b'OK')

uvloop/_testbase.py

@@ -13,6 +13,7 @@
 import select
 import socket
 import ssl
+import sys
 import tempfile
 import threading
 import time
@@ -89,6 +90,9 @@ def setUp(self):
             self._get_running_loop = asyncio.events._get_running_loop
             asyncio.events._get_running_loop = lambda: None
 
+        self.PY37 = sys.version_info[:2] >= (3, 7)
+        self.PY36 = sys.version_info[:2] >= (3, 6)
+
     def tearDown(self):
         self.loop.close()
 
@@ -268,10 +272,11 @@ def _create_server_ssl_context(self, certfile, keyfile=None):
         sslcontext.load_cert_chain(certfile, keyfile)
         return sslcontext
 
-    def _create_client_ssl_context(self):
+    def _create_client_ssl_context(self, *, disable_verify=True):
         sslcontext = ssl.create_default_context()
         sslcontext.check_hostname = False
-        sslcontext.verify_mode = ssl.CERT_NONE
+        if disable_verify:
+            sslcontext.verify_mode = ssl.CERT_NONE
         return sslcontext
 
     @contextlib.contextmanager

uvloop/handles/pipe.pxd

@@ -4,7 +4,7 @@ cdef class UnixServer(UVStreamServer):
 
     @staticmethod
     cdef UnixServer new(Loop loop, object protocol_factory, Server server,
-                        object ssl)
+                        object ssl, object ssl_handshake_timeout)
 
 
 cdef class UnixTransport(UVStream):

uvloop/handles/pipe.pyx

@@ -39,11 +39,12 @@ cdef class UnixServer(UVStreamServer):
 
     @staticmethod
     cdef UnixServer new(Loop loop, object protocol_factory, Server server,
-                          object ssl):
+                        object ssl, object ssl_handshake_timeout):
 
         cdef UnixServer handle
         handle = UnixServer.__new__(UnixServer)
-        handle._init(loop, protocol_factory, server, ssl)
+        handle._init(loop, protocol_factory, server,
+                     ssl, ssl_handshake_timeout)
         __pipe_init_uv_handle(<UVStream>handle, loop)
         return handle
 

uvloop/handles/streamserver.pxd

@@ -1,14 +1,15 @@
 cdef class UVStreamServer(UVSocketHandle):
     cdef:
         object ssl
+        object ssl_handshake_timeout
         object protocol_factory
         bint opened
         Server _server
 
     # All "inline" methods are final
 
     cdef inline _init(self, Loop loop, object protocol_factory,
-                      Server server, object ssl)
+                      Server server, object ssl, object ssl_handshake_timeout)
 
     cdef inline _mark_as_open(self)
 

uvloop/handles/streamserver.pyx

@@ -5,16 +5,24 @@ cdef class UVStreamServer(UVSocketHandle):
         self.opened = 0
         self._server = None
         self.ssl = None
+        self.ssl_handshake_timeout = None
         self.protocol_factory = None
 
     cdef inline _init(self, Loop loop, object protocol_factory,
-                      Server server, object ssl):
+                      Server server, object ssl, object ssl_handshake_timeout):
+
+        if ssl is not None:
+            if not isinstance(ssl, ssl_SSLContext):
+                raise TypeError(
+                    'ssl is expected to be None or an instance of '
+                    'ssl.SSLContext, got {!r}'.format(ssl))
+        else:
+            if ssl_handshake_timeout is not None:
+                raise ValueError(
+                    'ssl_handshake_timeout is only meaningful with ssl')
 
-        if ssl is not None and not isinstance(ssl, ssl_SSLContext):
-            raise TypeError(
-                'ssl is expected to be None or an instance of '
-                'ssl.SSLContext, got {!r}'.format(ssl))
         self.ssl = ssl
+        self.ssl_handshake_timeout = ssl_handshake_timeout
 
         self._start_init(loop)
         self.protocol_factory = protocol_factory
@@ -57,8 +65,9 @@ cdef class UVStreamServer(UVSocketHandle):
             ssl_protocol = SSLProtocol(
                 self._loop, protocol, self.ssl,
                 waiter,
-                True,  # server_side
-                None)  # server_hostname
+                server_side=True,
+                server_hostname=None,
+                ssl_handshake_timeout=self.ssl_handshake_timeout)
 
             client = self._make_new_transport(ssl_protocol, None)
 

uvloop/handles/tcp.pxd

@@ -3,7 +3,8 @@ cdef class TCPServer(UVStreamServer):
 
     @staticmethod
     cdef TCPServer new(Loop loop, object protocol_factory, Server server,
-                       object ssl, unsigned int flags)
+                       object ssl, unsigned int flags,
+                       object ssl_handshake_timeout)
 
 
 cdef class TCPTransport(UVStream):