@@ -775,6 +775,130 @@ async def test():
775775 with s1 , s2 :
776776 loop .run_until_complete (test ())
777777
778+ def test_create_connection_sock_cancel_detaches (self ):
779+ async def client (addr ):
780+ sock = socket .socket (socket .AF_INET , socket .SOCK_STREAM )
781+ sock .setblocking (False )
782+ try :
783+ sock .connect (addr )
784+ except BlockingIOError :
785+ pass
786+ await asyncio .sleep (0.01 )
787+
788+ task = asyncio .ensure_future (
789+ self .loop .create_connection (asyncio .Protocol , sock = sock ))
790+ await asyncio .sleep (0 )
791+ task .cancel ()
792+ with self .assertRaises (asyncio .CancelledError ):
793+ await task
794+
795+ # After cancellation the socket must be detached (fd == -1)
796+ # so that its __del__ won't close a recycled fd.
797+ self .assertEqual (sock .fileno (), - 1 )
798+
799+ def _recv_or_abort (sock ):
800+ try :
801+ sock .recv_all (1 )
802+ except ConnectionAbortedError :
803+ pass
804+
805+ with self .tcp_server (_recv_or_abort ,
806+ max_clients = 1 ,
807+ backlog = 1 ) as srv :
808+ self .loop .run_until_complete (client (srv .addr ))
809+
810+ def test_create_connection_sock_cancel_fd_leak (self ):
811+ # Regression test for https://github.com/MagicStack/uvloop/issues/645
812+ # and https://github.com/aio-libs/aiohttp/issues/10506
813+ #
814+ # When create_connection(sock=sock) is cancelled, the socket must
815+ # be detached so its close()/`__del__` won't double-close the fd.
816+ # Without the fix, libuv closes the fd but the socket object still
817+ # references it, enabling a chain of fd corruption and data leak:
818+ #
819+ # 1. cancel → libuv closes fd N
820+ # 2. New connection (victim) reuses fd N
821+ # 3. Stale sock.close() closes fd N → breaks the victim
822+ # 4. Another fd N is opened (new connection)
823+ # 5. Victim writev(N) → data goes to the wrong connection
824+
825+ async def test ():
826+ srv = await asyncio .start_server (
827+ lambda r , w : w .close (),
828+ '127.0.0.1' , 0 ,
829+ family = socket .AF_INET )
830+ addr = srv .sockets [0 ].getsockname ()
831+
832+ # --- Step 1: create_connection with sock= and cancel it ---
833+ sock = socket .socket (socket .AF_INET , socket .SOCK_STREAM )
834+ sock .setblocking (False )
835+ await self .loop .sock_connect (sock , addr )
836+ stale_fd = sock .fileno ()
837+
838+ task = self .loop .create_task (
839+ self .loop .create_connection (asyncio .Protocol , sock = sock )
840+ )
841+ await asyncio .sleep (0 )
842+ task .cancel ()
843+ with self .assertRaises (asyncio .CancelledError ):
844+ await task
845+
846+ # --- Step 2: a victim connection reuses the fd ---
847+ victim_tr , _ = await self .loop .create_connection (
848+ asyncio .Protocol , * addr )
849+ victim_fd = victim_tr .get_extra_info ('socket' ).fileno ()
850+ if victim_fd != stale_fd :
851+ victim_tr .close ()
852+ sock .close ()
853+ srv .close ()
854+ await srv .wait_closed ()
855+ raise unittest .SkipTest (
856+ f'fd not reused (got { victim_fd } { stale_fd } )
857+
858+ # --- Step 3: stale sock.close() must NOT kill the victim ---
859+ # Allocate the socketpair BEFORE sock.close() so the pair
860+ # fds don't collide with stale_fd.
861+ spy_a , spy_b = socket .socketpair ()
862+ spy_b .setblocking (False )
863+
864+ sock .close ()
865+
866+ # Check whether sock.close() broke the victim's fd.
867+ victim_broken = False
868+ try :
869+ os .fstat (victim_fd )
870+ except OSError :
871+ victim_broken = True
872+
873+ if victim_broken :
874+ # The victim's fd was killed — place a spy socket on
875+ # the freed fd (in production this would be a new
876+ # incoming connection).
877+ os .dup2 (spy_a .fileno (), stale_fd )
878+ spy_a .close ()
879+
880+ # Victim writes. If victim_broken, writev(stale_fd) goes
881+ # to the spy; otherwise it goes to the real connection.
882+ victim_tr .write (b'LEAKED' )
883+
884+ try :
885+ leaked = spy_b .recv (4096 )
886+ except BlockingIOError :
887+ leaked = b''
888+
889+ if victim_broken :
890+ os .close (stale_fd )
891+ spy_b .close ()
892+ victim_tr .close ()
893+ srv .close ()
894+ await srv .wait_closed ()
895+
896+ self .assertEqual (leaked , b'' ,
897+ f"Data leaked to an unrelated socket: "
898+ f"got { leaked !r} )
899+
900+ self .loop .run_until_complete (test ())
901+
778902
779903class Test_UV_TCP (_TestTCP , tb .UVTestCase ):
780904 def test_create_server_buffered_1 (self ):
0 commit comments