Attach the passed socket before awaiting connection_made

tests/test_tcp.py

@@ -6,6 +6,7 @@
 import socket
 import unittest.mock
 import ssl
+import struct
 import sys
 import threading
 import time
@@ -861,6 +862,37 @@ async def test():
 
         self.loop.run_until_complete(test())
 
+    def test_create_connection_sock_abort_detaches(self):
+        # If the protocol aborts the transport in connection_made(),
+        # connection_lost() closes the transport before
+        # create_connection() resumes. The passed socket must still be
+        # detached, so that its __del__ won't close a recycled fd.
+        class AbortProto(asyncio.Protocol):
+            def connection_made(self, transport):
+                transport.abort()
+
+        async def test():
+            srv = await asyncio.start_server(
+                lambda r, w: w.close(),
+                '127.0.0.1', 0,
+                family=socket.AF_INET)
+            addr = srv.sockets[0].getsockname()
+
+            sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
+            sock.setblocking(False)
+            await self.loop.sock_connect(sock, addr)
+
+            tr, _ = await self.loop.create_connection(AbortProto, sock=sock)
+            self.assertTrue(tr.is_closing())
+            await asyncio.sleep(0)
+
+            self.assertEqual(sock.fileno(), -1)
+
+            srv.close()
+            await srv.wait_closed()
+
+        self.loop.run_until_complete(test())
+
 
 class Test_UV_TCP(_TestTCP, tb.UVTestCase):
 
@@ -3309,6 +3341,48 @@ async def run_main():
 
         self.loop.run_until_complete(run_main())
 
+    def test_create_connection_ssl_sock_reset_detaches(self):
+        # If the peer resets the connection before the ClientHello is
+        # written, connection_lost() closes the transport before
+        # create_connection() resumes. The passed socket must still be
+        # detached, so that its __del__ won't close a recycled fd.
+        client_sslctx = self._create_client_ssl_context()
+
+        # silence error logger
+        self.loop.set_exception_handler(lambda *args: None)
+
+        def reset(reader, writer):
+            writer.get_extra_info('socket').setsockopt(
+                socket.SOL_SOCKET, socket.SO_LINGER, struct.pack('ii', 1, 0))
+            writer.transport.abort()
+
+        async def test():
+            srv = await asyncio.start_server(
+                reset, '127.0.0.1', 0, family=socket.AF_INET)
+            addr = srv.sockets[0].getsockname()
+
+            sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
+            sock.setblocking(False)
+            await self.loop.sock_connect(sock, addr)
+
+            # Wait for the RST to arrive.
+            for _ in range(500):
+                if select.select([sock], [], [], 0)[0]:
+                    break
+                await asyncio.sleep(0.01)
+
+            with self.assertRaises(OSError):
+                await self.loop.create_connection(
+                    asyncio.Protocol, sock=sock,
+                    ssl=client_sslctx, server_hostname='')
+
+            self.assertEqual(sock.fileno(), -1)
+
+            srv.close()
+            await srv.wait_closed()
+
+        self.loop.run_until_complete(test())
+
 
 class Test_UV_TCPSSL(_TestSSL, tb.UVTestCase):
     pass

tests/test_unix.py

@@ -1,6 +1,7 @@
 import asyncio
 import os
 import pathlib
+import select
 import socket
 import tempfile
 import time
@@ -515,6 +516,36 @@ async def test():
 
         self.loop.run_until_complete(test())
 
+    def test_create_unix_connection_sock_abort_detaches(self):
+        # Same as test_create_connection_sock_abort_detaches but for
+        # the create_unix_connection(sock=) path.
+        class AbortProto(asyncio.Protocol):
+            def connection_made(self, transport):
+                transport.abort()
+
+        async def test():
+            srv_path = os.path.join(tempfile.mkdtemp(), 'test.sock')
+            srv = await asyncio.start_unix_server(
+                lambda r, w: w.close(), path=srv_path)
+
+            sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
+            sock.setblocking(False)
+            await self.loop.sock_connect(sock, srv_path)
+
+            tr, _ = await self.loop.create_unix_connection(
+                AbortProto, sock=sock)
+            self.assertTrue(tr.is_closing())
+            await asyncio.sleep(0)
+
+            self.assertEqual(sock.fileno(), -1)
+
+            srv.close()
+            await srv.wait_closed()
+            if os.path.exists(srv_path):
+                os.unlink(srv_path)
+
+        self.loop.run_until_complete(test())
+
 
 class Test_UV_Unix(_TestUnix, tb.UVTestCase):
 
@@ -808,6 +839,44 @@ def run(coro):
         with self._silence_eof_received_warning():
             run(client)
 
+    def test_create_unix_connection_ssl_sock_closed_detaches(self):
+        # Same as test_create_connection_ssl_sock_reset_detaches but for
+        # the create_unix_connection(sock=) path: writing the ClientHello
+        # to a socket whose peer is closed fails with EPIPE.
+        client_sslctx = self._create_client_ssl_context()
+
+        # silence error logger
+        self.loop.set_exception_handler(lambda *args: None)
+
+        async def test():
+            srv_path = os.path.join(tempfile.mkdtemp(), 'test.sock')
+            srv = await asyncio.start_unix_server(
+                lambda r, w: w.transport.abort(), path=srv_path)
+
+            sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
+            sock.setblocking(False)
+            await self.loop.sock_connect(sock, srv_path)
+
+            # Wait for the peer to close the connection.
+            for _ in range(500):
+                if select.select([sock], [], [], 0)[0]:
+                    break
+                await asyncio.sleep(0.01)
+
+            with self.assertRaises(OSError):
+                await self.loop.create_unix_connection(
+                    asyncio.Protocol, sock=sock,
+                    ssl=client_sslctx, server_hostname='')
+
+            self.assertEqual(sock.fileno(), -1)
+
+            srv.close()
+            await srv.wait_closed()
+            if os.path.exists(srv_path):
+                os.unlink(srv_path)
+
+        self.loop.run_until_complete(test())
+
 
 class Test_UV_UnixSSL(_TestSSL, tb.UVTestCase):
     pass

uvloop/loop.pyx

@@ -2061,6 +2061,7 @@ cdef class Loop:
                 # FD here because of:
                 # https://github.com/python/asyncio/pull/449
                 tr._open(sock.fileno())
+                tr._attach_fileobj(sock)
                 tr._init_protocol()
                 await waiter
             except (KeyboardInterrupt, SystemExit):
@@ -2083,8 +2084,6 @@ cdef class Loop:
                 sock.detach()
                 raise
 
-            tr._attach_fileobj(sock)
-
         if ssl:
             app_transport = protocol._get_app_transport(context)
             try:
@@ -2317,6 +2316,7 @@ cdef class Loop:
             tr = UnixTransport.new(self, protocol, None, waiter, context)
             try:
                 tr._open(sock.fileno())
+                tr._attach_fileobj(sock)
                 tr._init_protocol()
                 await waiter
             except (KeyboardInterrupt, SystemExit):
@@ -2327,8 +2327,6 @@ cdef class Loop:
                 sock.detach()
                 raise
 
-            tr._attach_fileobj(sock)
-
         if ssl:
             app_transport = protocol._get_app_transport(Context_CopyCurrent())
             try: