Implement SSL for create_connection and create_unix_connection

Author: 1st1

tests/test_tcp.py

@@ -480,6 +480,70 @@ async def start_server():
         for client in clients:
             client.stop()
 
+    def test_create_connection_ssl_1(self):
+        CNT = 0
+        TOTAL_CNT = 25
+
+        A_DATA = b'A' * 1024 * 1024
+        B_DATA = b'B' * 1024 * 1024
+
+        sslctx = self._create_server_ssl_context(self.ONLYCERT, self.ONLYKEY)
+        client_sslctx = self._create_client_ssl_context()
+
+        def server():
+            yield tb.starttls(
+                sslctx,
+                server_side=True)
+
+            data = yield tb.read(len(A_DATA))
+            self.assertEqual(data, A_DATA)
+            yield tb.write(b'OK')
+
+            data = yield tb.read(len(B_DATA))
+            self.assertEqual(data, B_DATA)
+            yield tb.write(b'SPAM')
+
+            yield tb.close()
+
+        async def client(addr):
+            reader, writer = await asyncio.open_connection(
+                *addr,
+                ssl=client_sslctx,
+                server_hostname='',
+                loop=self.loop)
+
+            writer.write(A_DATA)
+            self.assertEqual(await reader.readexactly(2), b'OK')
+
+            writer.write(B_DATA)
+            self.assertEqual(await reader.readexactly(4), b'SPAM')
+
+            nonlocal CNT
+            CNT += 1
+
+            writer.close()
+
+        def run(coro):
+            nonlocal CNT
+            CNT = 0
+
+            srv = tb.tcp_server(server,
+                                max_clients=TOTAL_CNT,
+                                backlog=TOTAL_CNT)
+            srv.start()
+
+            tasks = []
+            for _ in range(TOTAL_CNT):
+                tasks.append(coro(srv.addr))
+
+            self.loop.run_until_complete(
+                asyncio.gather(*tasks, loop=self.loop))
+            srv.join()
+            self.assertEqual(CNT, TOTAL_CNT)
+
+        with self._silence_eof_received_warning():
+            run(client)
+
 
 class Test_UV_TCPSSL(_TestSSL, tb.UVTestCase):
     pass

tests/test_unix.py

@@ -413,6 +413,71 @@ async def start_server():
         for client in clients:
             client.stop()
 
+    def test_create_unix_connection_ssl_1(self):
+        CNT = 0
+        TOTAL_CNT = 25
+
+        A_DATA = b'A' * 1024 * 1024
+        B_DATA = b'B' * 1024 * 1024
+
+        sslctx = self._create_server_ssl_context(self.ONLYCERT, self.ONLYKEY)
+        client_sslctx = self._create_client_ssl_context()
+
+        def server():
+            yield tb.starttls(
+                sslctx,
+                server_side=True)
+
+            data = yield tb.read(len(A_DATA))
+            self.assertEqual(data, A_DATA)
+            yield tb.write(b'OK')
+
+            data = yield tb.read(len(B_DATA))
+            self.assertEqual(data, B_DATA)
+            yield tb.write(b'SPAM')
+
+            yield tb.close()
+
+        async def client(addr):
+            reader, writer = await asyncio.open_unix_connection(
+                addr,
+                ssl=client_sslctx,
+                server_hostname='',
+                loop=self.loop)
+
+            writer.write(A_DATA)
+            self.assertEqual(await reader.readexactly(2), b'OK')
+
+            writer.write(B_DATA)
+            self.assertEqual(await reader.readexactly(4), b'SPAM')
+
+            nonlocal CNT
+            CNT += 1
+
+            writer.close()
+
+        def run(coro):
+            nonlocal CNT
+            CNT = 0
+
+            srv = tb.tcp_server(server,
+                                family=socket.AF_UNIX,
+                                max_clients=TOTAL_CNT,
+                                backlog=TOTAL_CNT)
+            srv.start()
+
+            tasks = []
+            for _ in range(TOTAL_CNT):
+                tasks.append(coro(srv.addr))
+
+            self.loop.run_until_complete(
+                asyncio.gather(*tasks, loop=self.loop))
+            srv.join()
+            self.assertEqual(CNT, TOTAL_CNT)
+
+        with self._silence_eof_received_warning():
+            run(client)
+
 
 class Test_UV_UnixSSL(_TestSSL, tb.UVTestCase):
     pass

uvloop/loop.pyx

@@ -1041,8 +1041,27 @@ cdef class Loop:
             system.addrinfo *lai
             UVTCPTransport tr
 
-        if ssl is not None:
-            raise NotImplementedError('SSL is not yet supported')
+            object app_protocol
+            object protocol
+            object ssl_waiter
+
+        app_protocol = protocol = protocol_factory()
+        ssl_waiter = None
+        if ssl:
+            if server_hostname is None:
+                if not host:
+                    raise ValueError('You must set server_hostname '
+                                     'when using ssl without a host')
+                server_hostname = host
+
+            ssl_waiter = aio_Future(loop=self)
+            sslcontext = None if isinstance(ssl, bool) else ssl
+            protocol = aio_SSLProtocol(
+                self, app_protocol, sslcontext, ssl_waiter,
+                False, server_hostname)
+        else:
+            if server_hostname is not None:
+                raise ValueError('server_hostname is only meaningful with ssl')
 
         if host is not None or port is not None:
             f1 = self._getaddrinfo(host, port, family,
@@ -1075,8 +1094,6 @@ cdef class Loop:
                     raise OSError(
                         'getaddrinfo() returned empty list for local_addr')
 
-            protocol = protocol_factory()
-
             exceptions = []
             rai = ai_remote.data
             while rai is not NULL:
@@ -1146,7 +1163,11 @@ cdef class Loop:
                 tr._close()
                 raise
 
-        return tr, protocol
+        if ssl:
+            await ssl_waiter
+            return protocol._app_transport, app_protocol
+        else:
+            return tr, protocol
 
     @aio_coroutine
     async def create_unix_server(self, protocol_factory, str path=None,
@@ -1204,10 +1225,27 @@ cdef class Loop:
                                      ssl=None, sock=None,
                                      server_hostname=None):
 
-        cdef UVPipeTransport tr
-
-        if ssl is not None:
-            raise NotImplementedError('SSL is not yet supported')
+        cdef:
+            UVPipeTransport tr
+            object app_protocol
+            object protocol
+            object ssl_waiter
+
+        app_protocol = protocol = protocol_factory()
+        ssl_waiter = None
+        if ssl:
+            if server_hostname is None:
+                raise ValueError('You must set server_hostname '
+                                 'when using ssl without a host')
+
+            ssl_waiter = aio_Future(loop=self)
+            sslcontext = None if isinstance(ssl, bool) else ssl
+            protocol = aio_SSLProtocol(
+                self, app_protocol, sslcontext, ssl_waiter,
+                False, server_hostname)
+        else:
+            if server_hostname is not None:
+                raise ValueError('server_hostname is only meaningful with ssl')
 
         if path is not None:
             if isinstance(path, str):
@@ -1217,7 +1255,6 @@ cdef class Loop:
                 raise ValueError(
                     'path and sock can not be specified at the same time')
 
-            protocol = protocol_factory()
             waiter = aio_Future(loop=self)
             tr = UVPipeTransport.new(self, protocol, None, waiter)
             tr.connect(path)
@@ -1236,7 +1273,6 @@ cdef class Loop:
                     'A UNIX Domain Socket was expected, got {!r}'.format(sock))
 
             waiter = aio_Future(loop=self)
-            protocol = protocol_factory()
             tr = UVPipeTransport.new(self, protocol, None, waiter)
             try:
                 # libuv will make socket non-blocking
@@ -1249,7 +1285,11 @@ cdef class Loop:
                 tr._close()
                 raise
 
-        return tr, protocol
+        if ssl:
+            await ssl_waiter
+            return protocol._app_transport, app_protocol
+        else:
+            return tr, protocol
 
     def default_exception_handler(self, context):
         message = context.get('message')