Eryk Sun <eryk...@gmail.com> added the comment:
As far as I can tell, reduction.send_handle isn't used internally in the
Windows implementation, and it's also not a documented API function. However,
it is tested on Windows in test_fd_transfer in
Lib/test/_test_multiprocessing.py. As it turns out, the bug that Cameron's
proposed solution fixes slips under the radar. By coincidence,
DUPLICATE_SAME_ACCESS (2) has the same value as the file access right
FILE_WRITE_DATA (2), and test_fd_transfer only checks whether the child can
write to a file handle.
I propose adding test_fd_transfer_windows to _TestConnection in
Lib/test/_test_multiprocessing.py, which will test whether the parent and child
are granted the same access to a kernel file object after the handle is sent to
the child.
@classmethod
def _check_handle_access(cls, conn):
handle = reduction.recv_handle(conn)
conn.send(get_handle_info(handle).GrantedAccess)
@unittest.skipUnless(HAS_REDUCTION, "test needs multiprocessing.reduction")
@unittest.skipIf(sys.platform != "win32", "Windows-only test")
def test_fd_transfer_windows(self):
if self.TYPE != 'processes':
self.skipTest("only makes sense with processes")
conn, child_conn = self.Pipe(duplex=True)
p = self.Process(target=self._check_handle_access, args=(child_conn,))
p.daemon = True
p.start()
try:
with open(test.support.TESTFN, "wb") as f:
self.addCleanup(test.support.unlink, test.support.TESTFN)
handle = msvcrt.get_osfhandle(f.fileno())
parent_access = get_handle_info(handle).GrantedAccess
reduction.send_handle(conn, handle, p.pid)
child_access = conn.recv()
self.assertEqual(parent_access, child_access)
finally:
p.join()
get_handle_info() and the required ctypes support definitions [1] would be
defined at module scope as follows:
if WIN32:
from ctypes import (WinDLL, WinError, Structure, POINTER,
byref, sizeof, c_void_p, c_ulong)
ntdll = WinDLL('ntdll')
ntdll.NtQueryObject.argtypes = (
c_void_p, # Handle
c_ulong, # ObjectInformationClass
c_void_p, # ObjectInformation
c_ulong, # ObjectInformationLength
POINTER(c_ulong)) # ReturnLength
ObjectBasicInformation = 0
class PUBLIC_OBJECT_BASIC_INFORMATION(Structure):
(r"https://docs.microsoft.com/en-us/windows/win32/api";
r"/winternl/nf-winternl-ntqueryobject")
_fields_ = (('Attributes', c_ulong),
('GrantedAccess', c_ulong),
('HandleCount', c_ulong),
('PointerCount', c_ulong),
('Reserved', c_ulong * 10))
def get_handle_info(handle):
info = PUBLIC_OBJECT_BASIC_INFORMATION()
status = ntdll.NtQueryObject(handle, ObjectBasicInformation,
byref(info), sizeof(info), None)
if status < 0:
error = ntdll.RtlNtStatusToDosError(status)
raise WinError(error)
return info
[1]
https://docs.microsoft.com/en-us/windows/win32/api/winternl/nf-winternl-ntqueryobject
----------
_______________________________________
Python tracker <rep...@bugs.python.org>
<https://bugs.python.org/issue38188>
_______________________________________
_______________________________________________
Python-bugs-list mailing list
Unsubscribe:
https://mail.python.org/mailman/options/python-bugs-list/archive%40mail-archive.com
