bpo-32270: Don't close stdin/out/err in pass_fds (GH-6242)

When subprocess.Popen() stdin= stdout= or stderr= handles are specified
and appear in pass_fds=, don't close the original fds after dup'ing them.

This implementation and unittest primarily came from @izbyshev (see the PR)

See also https://github.com/izbyshev/cpython/commit/b89b52f28490b69142d5c061604b3a3989cec66c

This also removes the old manual p2cread, c2pwrite, and errwrite closing logic
as inheritable flags and _close_open_fds takes care of that properly today without special treatment.

This code is within child_exec() where it is the only thread so there is no
race condition between the dup and _Py_set_inheritable_async_safe call.

1 files changed, 30 insertions, 0 deletions

diff --git a/Lib/test/test_subprocess.py b/Lib/test/test_subprocess.py
index 4719773..8419061 100644
--- a/Lib/test/test_subprocess.py
+++ b/Lib/test/test_subprocess.py
@@ -2529,6 +2529,36 @@ class POSIXProcessTestCase(BaseTestCase):
         self.assertEqual(os.get_inheritable(inheritable), True)
         self.assertEqual(os.get_inheritable(non_inheritable), False)
 
+
+    # bpo-32270: Ensure that descriptors specified in pass_fds
+    # are inherited even if they are used in redirections.
+    # Contributed by @izbyshev.
+    def test_pass_fds_redirected(self):
+        """Regression test for https://bugs.python.org/issue32270."""
+        fd_status = support.findfile("fd_status.py", subdir="subprocessdata")
+        pass_fds = []
+        for _ in range(2):
+            fd = os.open(os.devnull, os.O_RDWR)
+            self.addCleanup(os.close, fd)
+            pass_fds.append(fd)
+
+        stdout_r, stdout_w = os.pipe()
+        self.addCleanup(os.close, stdout_r)
+        self.addCleanup(os.close, stdout_w)
+        pass_fds.insert(1, stdout_w)
+
+        with subprocess.Popen([sys.executable, fd_status],
+                              stdin=pass_fds[0],
+                              stdout=pass_fds[1],
+                              stderr=pass_fds[2],
+                              close_fds=True,
+                              pass_fds=pass_fds):
+            output = os.read(stdout_r, 1024)
+        fds = {int(num) for num in output.split(b',')}
+
+        self.assertEqual(fds, {0, 1, 2} | frozenset(pass_fds), f"output={output!a}")
+
+
     def test_stdout_stdin_are_single_inout_fd(self):
         with io.open(os.devnull, "r+") as inout:
             p = subprocess.Popen([sys.executable, "-c", "import sys; sys.exit(0)"],