mirror of
https://github.com/python/cpython.git
synced 2025-08-31 05:58:33 +00:00
bpo-19764: Implemented support for subprocess.Popen(close_fds=True) on Windows (#1218)
Even though Python marks any handles it opens as non-inheritable there is still a race when using `subprocess.Popen` since creating a process with redirected stdio requires temporarily creating inheritable handles. By implementing support for `subprocess.Popen(close_fds=True)` we fix this race. In order to implement this we use PROC_THREAD_ATTRIBUTE_HANDLE_LIST which is available since Windows Vista. Which allows to pass an explicit list of handles to inherit when creating a process. This commit also adds `STARTUPINFO.lpAttributeList["handle_list"]` which can be used to control PROC_THREAD_ATTRIBUTE_HANDLE_LIST directly.
This commit is contained in:
parent
87010e85cb
commit
b2a6083eb0
8 changed files with 425 additions and 54 deletions
|
@ -2743,11 +2743,6 @@ class Win32ProcessTestCase(BaseTestCase):
|
|||
[sys.executable, "-c",
|
||||
"import sys; sys.exit(47)"],
|
||||
preexec_fn=lambda: 1)
|
||||
self.assertRaises(ValueError, subprocess.call,
|
||||
[sys.executable, "-c",
|
||||
"import sys; sys.exit(47)"],
|
||||
stdout=subprocess.PIPE,
|
||||
close_fds=True)
|
||||
|
||||
@support.cpython_only
|
||||
def test_issue31471(self):
|
||||
|
@ -2765,6 +2760,67 @@ class Win32ProcessTestCase(BaseTestCase):
|
|||
close_fds=True)
|
||||
self.assertEqual(rc, 47)
|
||||
|
||||
def test_close_fds_with_stdio(self):
|
||||
import msvcrt
|
||||
|
||||
fds = os.pipe()
|
||||
self.addCleanup(os.close, fds[0])
|
||||
self.addCleanup(os.close, fds[1])
|
||||
|
||||
handles = []
|
||||
for fd in fds:
|
||||
os.set_inheritable(fd, True)
|
||||
handles.append(msvcrt.get_osfhandle(fd))
|
||||
|
||||
p = subprocess.Popen([sys.executable, "-c",
|
||||
"import msvcrt; print(msvcrt.open_osfhandle({}, 0))".format(handles[0])],
|
||||
stdout=subprocess.PIPE, close_fds=False)
|
||||
stdout, stderr = p.communicate()
|
||||
self.assertEqual(p.returncode, 0)
|
||||
int(stdout.strip()) # Check that stdout is an integer
|
||||
|
||||
p = subprocess.Popen([sys.executable, "-c",
|
||||
"import msvcrt; print(msvcrt.open_osfhandle({}, 0))".format(handles[0])],
|
||||
stdout=subprocess.PIPE, stderr=subprocess.PIPE, close_fds=True)
|
||||
stdout, stderr = p.communicate()
|
||||
self.assertEqual(p.returncode, 1)
|
||||
self.assertIn(b"OSError", stderr)
|
||||
|
||||
# The same as the previous call, but with an empty handle_list
|
||||
handle_list = []
|
||||
startupinfo = subprocess.STARTUPINFO()
|
||||
startupinfo.lpAttributeList = {"handle_list": handle_list}
|
||||
p = subprocess.Popen([sys.executable, "-c",
|
||||
"import msvcrt; print(msvcrt.open_osfhandle({}, 0))".format(handles[0])],
|
||||
stdout=subprocess.PIPE, stderr=subprocess.PIPE,
|
||||
startupinfo=startupinfo, close_fds=True)
|
||||
stdout, stderr = p.communicate()
|
||||
self.assertEqual(p.returncode, 1)
|
||||
self.assertIn(b"OSError", stderr)
|
||||
|
||||
# Check for a warning due to using handle_list and close_fds=False
|
||||
with support.check_warnings((".*overriding close_fds", RuntimeWarning)):
|
||||
startupinfo = subprocess.STARTUPINFO()
|
||||
startupinfo.lpAttributeList = {"handle_list": handles[:]}
|
||||
p = subprocess.Popen([sys.executable, "-c",
|
||||
"import msvcrt; print(msvcrt.open_osfhandle({}, 0))".format(handles[0])],
|
||||
stdout=subprocess.PIPE, stderr=subprocess.PIPE,
|
||||
startupinfo=startupinfo, close_fds=False)
|
||||
stdout, stderr = p.communicate()
|
||||
self.assertEqual(p.returncode, 0)
|
||||
|
||||
def test_empty_attribute_list(self):
|
||||
startupinfo = subprocess.STARTUPINFO()
|
||||
startupinfo.lpAttributeList = {}
|
||||
subprocess.call([sys.executable, "-c", "import sys; sys.exit(0)"],
|
||||
startupinfo=startupinfo)
|
||||
|
||||
def test_empty_handle_list(self):
|
||||
startupinfo = subprocess.STARTUPINFO()
|
||||
startupinfo.lpAttributeList = {"handle_list": []}
|
||||
subprocess.call([sys.executable, "-c", "import sys; sys.exit(0)"],
|
||||
startupinfo=startupinfo)
|
||||
|
||||
def test_shell_sequence(self):
|
||||
# Run command through the shell (sequence)
|
||||
newenv = os.environ.copy()
|
||||
|
|
Loading…
Add table
Add a link
Reference in a new issue