bpo-32270: Don't close stdin/out/err in pass_fds (GH-6242) (GH-9148)
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.
(cherry picked from commit ce34410b8b67f49d8275c05d51b3ead50cf97f48)
Co-authored-by: Gregory P. Smith <greg@krypto.org> [Google]
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 @@
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)"],