Implement SSL support for create_server and create_unix_server

tests/certs/ssl_cert.pem

@@ -0,0 +1,15 @@
+-----BEGIN CERTIFICATE-----
+MIICVDCCAb2gAwIBAgIJANfHOBkZr8JOMA0GCSqGSIb3DQEBBQUAMF8xCzAJBgNV
+BAYTAlhZMRcwFQYDVQQHEw5DYXN0bGUgQW50aHJheDEjMCEGA1UEChMaUHl0aG9u
+IFNvZnR3YXJlIEZvdW5kYXRpb24xEjAQBgNVBAMTCWxvY2FsaG9zdDAeFw0xMDEw
+MDgyMzAxNTZaFw0yMDEwMDUyMzAxNTZaMF8xCzAJBgNVBAYTAlhZMRcwFQYDVQQH
+Ew5DYXN0bGUgQW50aHJheDEjMCEGA1UEChMaUHl0aG9uIFNvZnR3YXJlIEZvdW5k
+YXRpb24xEjAQBgNVBAMTCWxvY2FsaG9zdDCBnzANBgkqhkiG9w0BAQEFAAOBjQAw
+gYkCgYEA21vT5isq7F68amYuuNpSFlKDPrMUCa4YWYqZRt2OZ+/3NKaZ2xAiSwr7
+6MrQF70t5nLbSPpqE5+5VrS58SY+g/sXLiFd6AplH1wJZwh78DofbFYXUggktFMt
+pTyiX8jtP66bkcPkDADA089RI1TQR6Ca+n7HFa7c1fabVV6i3zkCAwEAAaMYMBYw
+FAYDVR0RBA0wC4IJbG9jYWxob3N0MA0GCSqGSIb3DQEBBQUAA4GBAHPctQBEQ4wd
+BJ6+JcpIraopLn8BGhbjNWj40mmRqWB/NAWF6M5ne7KpGAu7tLeG4hb1zLaldK8G
+lxy2GPSRF6LFS48dpEj2HbMv2nvv6xxalDMJ9+DicWgAKTQ6bcX2j3GUkCR0g/T1
+CRlNBAAlvhKzO7Clpf9l0YKBEfraJByX
+-----END CERTIFICATE-----

tests/certs/ssl_key.pem

@@ -0,0 +1,16 @@
+-----BEGIN PRIVATE KEY-----
+MIICdwIBADANBgkqhkiG9w0BAQEFAASCAmEwggJdAgEAAoGBANtb0+YrKuxevGpm
+LrjaUhZSgz6zFAmuGFmKmUbdjmfv9zSmmdsQIksK++jK0Be9LeZy20j6ahOfuVa0
+ufEmPoP7Fy4hXegKZR9cCWcIe/A6H2xWF1IIJLRTLaU8ol/I7T+um5HD5AwAwNPP
+USNU0Eegmvp+xxWu3NX2m1Veot85AgMBAAECgYA3ZdZ673X0oexFlq7AAmrutkHt
+CL7LvwrpOiaBjhyTxTeSNWzvtQBkIU8DOI0bIazA4UreAFffwtvEuPmonDb3F+Iq
+SMAu42XcGyVZEl+gHlTPU9XRX7nTOXVt+MlRRRxL6t9GkGfUAXI3XxJDXW3c0vBK
+UL9xqD8cORXOfE06rQJBAP8mEX1ERkR64Ptsoe4281vjTlNfIbs7NMPkUnrn9N/Y
+BLhjNIfQ3HFZG8BTMLfX7kCS9D593DW5tV4Z9BP/c6cCQQDcFzCcVArNh2JSywOQ
+ZfTfRbJg/Z5Lt9Fkngv1meeGNPgIMLN8Sg679pAOOWmzdMO3V706rNPzSVMME7E5
+oPIfAkEA8pDddarP5tCvTTgUpmTFbakm0KoTZm2+FzHcnA4jRh+XNTjTOv98Y6Ik
+eO5d1ZnKXseWvkZncQgxfdnMqqpj5wJAcNq/RVne1DbYlwWchT2Si65MYmmJ8t+F
+0mcsULqjOnEMwf5e+ptq5LzwbyrHZYq5FNk7ocufPv/ZQrcSSC+cFwJBAKvOJByS
+x56qyGeZLOQlWS2JS3KJo59XuLFGqcbgN9Om9xFa41Yb4N9NvplFivsvZdw3m1Q/
+SPIXQuT8RMPDVNQ=
+-----END PRIVATE KEY-----

tests/test_tcp.py

@@ -1,25 +1,31 @@
 import asyncio
+import logging
 import socket
 import uvloop
+import ssl
+import warnings
 
 from uvloop import _testbase as tb
 
 
 class _TestTCP:
     def test_create_server_1(self):
         CNT = 0           # number of clients that were successful
-        TOTAL_CNT = 100   # total number of clients that test will create
+        TOTAL_CNT = 25    # total number of clients that test will create
         TIMEOUT = 5.0     # timeout for this test
 
+        A_DATA = b'A' * 1024 * 1024
+        B_DATA = b'B' * 1024 * 1024
+
         async def handle_client(reader, writer):
             nonlocal CNT
 
-            data = await reader.readexactly(4)
-            self.assertEqual(data, b'AAAA')
+            data = await reader.readexactly(len(A_DATA))
+            self.assertEqual(data, A_DATA)
             writer.write(b'OK')
 
-            data = await reader.readexactly(4)
-            self.assertEqual(data, b'BBBB')
+            data = await reader.readexactly(len(B_DATA))
+            self.assertEqual(data, B_DATA)
             writer.writelines([b'SP', bytearray(b'A'), memoryview(b'M')])
 
             await writer.drain()
@@ -33,54 +39,43 @@ async def test_client(addr):
                 sock.setblocking(False)
                 await self.loop.sock_connect(sock, addr)
 
-                await self.loop.sock_sendall(sock, b'AAAA')
+                await self.loop.sock_sendall(sock, A_DATA)
                 data = await self.loop.sock_recv(sock, 2)
                 self.assertEqual(data, b'OK')
 
-                await self.loop.sock_sendall(sock, b'BBBB')
+                await self.loop.sock_sendall(sock, B_DATA)
                 data = await self.loop.sock_recv(sock, 4)
                 self.assertEqual(data, b'SPAM')
 
         async def start_server():
             nonlocal CNT
             CNT = 0
 
-            try:
-                srv = await asyncio.start_server(
-                    handle_client,
-                    ('127.0.0.1', 'localhost'), 0,
-                    family=socket.AF_INET,
-                    loop=self.loop)
-
-                try:
-                    srv_socks = srv.sockets
-                    self.assertTrue(srv_socks)
+            srv = await asyncio.start_server(
+                handle_client,
+                ('127.0.0.1', 'localhost'), 0,
+                family=socket.AF_INET,
+                loop=self.loop)
 
-                    addr = srv_socks[0].getsockname()
+            srv_socks = srv.sockets
+            self.assertTrue(srv_socks)
 
-                    tasks = []
-                    for _ in range(TOTAL_CNT):
-                        tasks.append(test_client(addr))
+            addr = srv_socks[0].getsockname()
 
-                    try:
-                        await asyncio.wait_for(
-                            asyncio.gather(*tasks, loop=self.loop),
-                            TIMEOUT, loop=self.loop)
-                    finally:
-                        self.loop.stop()
+            tasks = []
+            for _ in range(TOTAL_CNT):
+                tasks.append(test_client(addr))
 
-                finally:
-                    self.loop.call_soon(srv.close)
-                    await srv.wait_closed()
+            await asyncio.wait_for(
+                asyncio.gather(*tasks, loop=self.loop),
+                TIMEOUT, loop=self.loop)
 
-                    # Check that the server cleaned-up proxy-sockets
-                    for srv_sock in srv_socks:
-                        self.assertEqual(srv_sock.fileno(), -1)
+            self.loop.call_soon(srv.close)
+            await srv.wait_closed()
 
-            except:
-                self.loop.stop()  # We don't want this test to stuck when
-                                  # it fails.
-                raise
+            # Check that the server cleaned-up proxy-sockets
+            for srv_sock in srv_socks:
+                self.assertEqual(srv_sock.fileno(), -1)
 
         async def start_server_sock():
             nonlocal CNT
@@ -89,47 +84,35 @@ async def start_server_sock():
             sock = socket.socket()
             sock.bind(('127.0.0.1', 0))
             addr = sock.getsockname()
-            try:
-                srv = await asyncio.start_server(
-                    handle_client,
-                    None, None,
-                    family=socket.AF_INET,
-                    loop=self.loop,
-                    sock=sock)
 
-                try:
-                    srv_socks = srv.sockets
-                    self.assertTrue(srv_socks)
-
-                    tasks = []
-                    for _ in range(TOTAL_CNT):
-                        tasks.append(test_client(addr))
-
-                    try:
-                        await asyncio.wait_for(
-                            asyncio.gather(*tasks, loop=self.loop),
-                            TIMEOUT, loop=self.loop)
-                    finally:
-                        self.loop.stop()
-
-                finally:
-                    srv.close()
-
-                    # Check that the server cleaned-up proxy-sockets
-                    for srv_sock in srv_socks:
-                        self.assertEqual(srv_sock.fileno(), -1)
-
-            except:
-                self.loop.stop()  # We don't want this test to stuck when
-                                  # it fails.
-                raise
-
-        self.loop.create_task(start_server())
-        self.loop.run_forever()
+            srv = await asyncio.start_server(
+                handle_client,
+                None, None,
+                family=socket.AF_INET,
+                loop=self.loop,
+                sock=sock)
+
+            srv_socks = srv.sockets
+            self.assertTrue(srv_socks)
+
+            tasks = []
+            for _ in range(TOTAL_CNT):
+                tasks.append(test_client(addr))
+
+            await asyncio.wait_for(
+                asyncio.gather(*tasks, loop=self.loop),
+                TIMEOUT, loop=self.loop)
+
+            srv.close()
+
+            # Check that the server cleaned-up proxy-sockets
+            for srv_sock in srv_socks:
+                self.assertEqual(srv_sock.fileno(), -1)
+
+        self.loop.run_until_complete(start_server())
         self.assertEqual(CNT, TOTAL_CNT)
 
-        self.loop.create_task(start_server_sock())
-        self.loop.run_forever()
+        self.loop.run_until_complete(start_server_sock())
         self.assertEqual(CNT, TOTAL_CNT)
 
     def test_create_connection_1(self):
@@ -185,8 +168,7 @@ def run(coro):
 
             srv = tb.tcp_server(server,
                                 max_clients=TOTAL_CNT,
-                                backlog=TOTAL_CNT,
-                                timeout=5)
+                                backlog=TOTAL_CNT)
             srv.start()
 
             tasks = []
@@ -248,8 +230,7 @@ def run(coro):
 
             srv = tb.tcp_server(server,
                                 max_clients=TOTAL_CNT,
-                                backlog=TOTAL_CNT,
-                                timeout=5)
+                                backlog=TOTAL_CNT)
             srv.start()
 
             tasks = []
@@ -403,3 +384,106 @@ class Test_UV_TCP(_TestTCP, tb.UVTestCase):
 
 class Test_AIO_TCP(_TestTCP, tb.AIOTestCase):
     pass
+
+
+class _TestSSL(tb.SSLTestCase):
+
+    def test_create_server_ssl_1(self):
+        CNT = 0           # number of clients that were successful
+        TOTAL_CNT = 25    # total number of clients that test will create
+        TIMEOUT = 5.0     # timeout for this test
+
+        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()
+
+        clients = []
+
+        async def handle_client(reader, writer):
+            nonlocal CNT
+
+            data = await reader.readexactly(len(A_DATA))
+            self.assertEqual(data, A_DATA)
+            writer.write(b'OK')
+
+            data = await reader.readexactly(len(B_DATA))
+            self.assertEqual(data, B_DATA)
+            writer.writelines([b'SP', bytearray(b'A'), memoryview(b'M')])
+
+            await writer.drain()
+            writer.close()
+
+            CNT += 1
+
+        async def test_client(addr):
+            fut = asyncio.Future(loop=self.loop)
+
+            def prog():
+                try:
+                    yield tb.starttls(client_sslctx)
+                    yield tb.connect(addr)
+                    yield tb.write(A_DATA)
+
+                    data = yield tb.read(2)
+                    self.assertEqual(data, b'OK')
+
+                    yield tb.write(B_DATA)
+                    data = yield tb.read(4)
+                    self.assertEqual(data, b'SPAM')
+
+                    yield tb.close()
+
+                except Exception as ex:
+                    self.loop.call_soon_threadsafe(fut.set_exception, ex)
+                else:
+                    self.loop.call_soon_threadsafe(fut.set_result, None)
+
+            client = tb.tcp_client(prog)
+            client.start()
+            clients.append(client)
+
+            await fut
+
+        async def start_server():
+            srv = await asyncio.start_server(
+                handle_client,
+                '127.0.0.1', 0,
+                family=socket.AF_INET,
+                ssl=sslctx,
+                loop=self.loop)
+
+            try:
+                srv_socks = srv.sockets
+                self.assertTrue(srv_socks)
+
+                addr = srv_socks[0].getsockname()
+
+                tasks = []
+                for _ in range(TOTAL_CNT):
+                    tasks.append(test_client(addr))
+
+                await asyncio.wait_for(
+                    asyncio.gather(*tasks, loop=self.loop),
+                    TIMEOUT, loop=self.loop)
+
+            finally:
+                self.loop.call_soon(srv.close)
+                await srv.wait_closed()
+
+        with self._silence_eof_received_warning():
+            self.loop.run_until_complete(start_server())
+
+        self.assertEqual(CNT, TOTAL_CNT)
+
+        for client in clients:
+            client.stop()
+
+
+class Test_UV_TCPSSL(_TestSSL, tb.UVTestCase):
+    pass
+
+
+class Test_AIO_TCPSSL(_TestSSL, tb.AIOTestCase):
+    pass