Fix pipe transport double-close of the underlying fd

connect_read_pipe and connect_write_pipe passed the Python file
object's fd to libuv, which takes ownership and closes it. Closing
the file object later (or GC) then closed the same fd again, which
can steal a recycled descriptor.

Dup the fd before uv_pipe_open so libuv and the file object each
own a distinct descriptor.

Author: jaideeppyne

tests/test_pipes.py

@@ -271,6 +271,87 @@ def test_write_buffer_full(self):
         self.loop.run_until_complete(asyncio.wait_for(proto.done, 1))
         self.assertEqual('CLOSED', proto.state)
 
+    def _assert_pipe_close_does_not_steal_fd(self, connect, make_pipeobj,
+                                             close_other_end):
+        # Regression test for https://github.com/MagicStack/uvloop/issues/763
+        # libuv closes the fd given to it; the Python file object must not
+        # close that same fd again (it may have been reused).
+        stolen = []
+        probed = []
+
+        class FileIOWithProbe(io.FileIO):
+            def close(self):
+                try:
+                    fd = super().fileno()
+                except (ValueError, OSError):
+                    super().close()
+                    return
+                try:
+                    os.fstat(fd)
+                    already_closed = False
+                except OSError:
+                    already_closed = True
+                probed.append('closed' if already_closed else 'open')
+                new_r, new_w = os.pipe()
+                try:
+                    super().close()
+                finally:
+                    for nfd in (new_r, new_w):
+                        try:
+                            os.fstat(nfd)
+                        except OSError:
+                            stolen.append(nfd)
+                        else:
+                            os.close(nfd)
+                    if already_closed:
+                        stolen.append(fd)
+
+        async def main():
+            pipe_read_fd, pipe_write_fd = os.pipe()
+            lost = self.loop.create_future()
+
+            class Proto(asyncio.Protocol):
+                def connection_lost(self, exc):
+                    if not lost.done():
+                        lost.set_result(None)
+
+            pipeobj = make_pipeobj(FileIOWithProbe, pipe_read_fd,
+                                   pipe_write_fd)
+            transport, _ = await connect(Proto, pipeobj)
+            transport.close()
+            await lost
+            close_other_end(pipe_read_fd, pipe_write_fd)
+
+        self.loop.run_until_complete(main())
+        self.assertIn('open', probed)
+        self.assertEqual(stolen, [])
+
+    def test_read_pipe_close_does_not_steal_fd(self):
+        def make_pipeobj(fileio_cls, read_fd, write_fd):
+            return io.BufferedReader(fileio_cls(read_fd, 'rb'))
+
+        async def connect(proto_factory, pipeobj):
+            return await self.loop.connect_read_pipe(proto_factory, pipeobj)
+
+        def close_other_end(read_fd, write_fd):
+            os.close(write_fd)
+
+        self._assert_pipe_close_does_not_steal_fd(
+            connect, make_pipeobj, close_other_end)
+
+    def test_write_pipe_close_does_not_steal_fd(self):
+        def make_pipeobj(fileio_cls, read_fd, write_fd):
+            return io.BufferedWriter(fileio_cls(write_fd, 'wb'))
+
+        async def connect(proto_factory, pipeobj):
+            return await self.loop.connect_write_pipe(proto_factory, pipeobj)
+
+        def close_other_end(read_fd, write_fd):
+            os.close(read_fd)
+
+        self._assert_pipe_close_does_not_steal_fd(
+            connect, make_pipeobj, close_other_end)
+
 
 class Test_UV_Pipes(_BasePipeTest, tb.UVTestCase):
     pass

uvloop/loop.pyx

@@ -2862,18 +2862,31 @@ cdef class Loop:
         ReadTransport interface."""
         cdef:
             ReadUnixTransport transp
+            int fd
+            bint opened
 
         waiter = self._new_future()
         proto = proto_factory()
         transp = ReadUnixTransport.new(self, proto, None, waiter)
         transp._add_extra_info('pipe', pipe)
+        # Duplicate the fd so libuv and the Python file object each own a
+        # distinct descriptor.  uv_close() closes the fd given to libuv;
+        # without a dup, fileobj.close() (or GC) would close the same number
+        # again and could steal a recycled fd.  See issue #763.
+        fd = os_dup(pipe.fileno())
+        opened = 0
         try:
-            transp._open(pipe.fileno())
+            transp._open(fd)
+            opened = 1
             transp._init_protocol()
             await waiter
         except (KeyboardInterrupt, SystemExit):
+            if not opened:
+                os_close(fd)
             raise
         except BaseException:
+            if not opened:
+                os_close(fd)
             transp._close()
             raise
         transp._attach_fileobj(pipe)
@@ -2889,18 +2902,29 @@ cdef class Loop:
         WriteTransport interface."""
         cdef:
             WriteUnixTransport transp
+            int fd
+            bint opened
 
         waiter = self._new_future()
         proto = proto_factory()
         transp = WriteUnixTransport.new(self, proto, None, waiter)
         transp._add_extra_info('pipe', pipe)
+        # See connect_read_pipe() — dup so libuv and the file object do not
+        # share an fd (issue #763).
+        fd = os_dup(pipe.fileno())
+        opened = 0
         try:
-            transp._open(pipe.fileno())
+            transp._open(fd)
+            opened = 1
             transp._init_protocol()
             await waiter
         except (KeyboardInterrupt, SystemExit):
+            if not opened:
+                os_close(fd)
             raise
         except BaseException:
+            if not opened:
+                os_close(fd)
             transp._close()
             raise
         transp._attach_fileobj(pipe)