diff --git a/tests/test_tcp.py b/tests/test_tcp.py index b933166b..f60fb2e9 100644 --- a/tests/test_tcp.py +++ b/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 diff --git a/tests/test_unix.py b/tests/test_unix.py index d66dc708..54dd65e2 100644 --- a/tests/test_unix.py +++ b/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 diff --git a/uvloop/loop.pyx b/uvloop/loop.pyx index 802cdb69..7d1b23b1 100644 --- a/uvloop/loop.pyx +++ b/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: