Mirror
/
cpython
mirror of https://github.com/python/cpython
4
1
Fork 0
Code Issues Releases Wiki Activity

gh-119690: Adds Unicode support for named pipes in _winapi (GH-119717)

This commit is contained in:
Steve Dower 2024-05-29 16:51:09 +01:00 committed by GitHub
parent 34f9b3e724
commit 78d697b7d5
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
6 changed files with 92 additions and 29 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

11
Lib/test/audit-tests.py
View File

@ -556,6 +556,17 @@ def test_sys_monitoring_register_callback():
sys.monitoring.register_callback(1, 1, None)
def test_winapi_createnamedpipe(pipe_name):
import _winapi
def hook(event, args):
if event == "_winapi.CreateNamedPipe":
print(event, args)
sys.addaudithook(hook)
_winapi.CreateNamedPipe(pipe_name, _winapi.PIPE_ACCESS_DUPLEX, 8, 2, 0, 0, 0, 0)
if __name__ == "__main__":
from test.support import suppress_msvcrt_asserts

14
Lib/test/test_audit.py
View File

@ -291,6 +291,20 @@ class AuditTest(unittest.TestCase):
self.assertEqual(actual, expected)
def test_winapi_createnamedpipe(self):
winapi = import_helper.import_module("_winapi")
pipe_name = r"\\.\pipe\LOCAL\test_winapi_createnamed_pipe"
returncode, events, stderr = self.run_python("test_winapi_createnamedpipe", pipe_name)
if returncode:
self.fail(stderr)
if support.verbose:
print(*events, sep='\n')
actual = [(ev[0], ev[2]) for ev in events]
expected = [("_winapi.CreateNamedPipe", f"({pipe_name!r}, 3, 8)")]
self.assertEqual(actual, expected)
if __name__ == "__main__":
unittest.main()

34
Lib/test/test_winapi.py
View File

@ -7,7 +7,7 @@ import re
import threading
import time
import unittest
from test.support import import_helper
from test.support import import_helper, os_helper
_winapi = import_helper.import_module('_winapi', required_on=['win'])
@ -127,3 +127,35 @@ class WinAPITests(unittest.TestCase):
# Should contain "PROGRA~" but we can't predict the number
self.assertIsNotNone(re.match(r".\:\\PROGRA~\d", actual.upper()), actual)
def test_namedpipe(self):
pipe_name = rf"\\.\pipe\LOCAL\{os_helper.TESTFN}"
# Pipe does not exist, so this raises
with self.assertRaises(FileNotFoundError):
_winapi.WaitNamedPipe(pipe_name, 0)
pipe = _winapi.CreateNamedPipe(
pipe_name,
_winapi.PIPE_ACCESS_DUPLEX,
8, # 8=PIPE_REJECT_REMOTE_CLIENTS
2, # two instances available
32, 32, 0, 0)
self.addCleanup(_winapi.CloseHandle, pipe)
# Pipe instance is available, so this passes
_winapi.WaitNamedPipe(pipe_name, 0)
with open(pipe_name, 'w+b') as pipe2:
# No instances available, so this times out
# (WinError 121 does not get mapped to TimeoutError)
with self.assertRaises(OSError):
_winapi.WaitNamedPipe(pipe_name, 0)
_winapi.WriteFile(pipe, b'testdata')
self.assertEqual(b'testdata', pipe2.read(8))
self.assertEqual((b'', 0), _winapi.PeekNamedPipe(pipe, 8)[:2])
pipe2.write(b'testdata')
pipe2.flush()
self.assertEqual((b'testdata', 8), _winapi.PeekNamedPipe(pipe, 8)[:2])

1
Misc/NEWS.d/next/Windows/2024-05-29-11-06-12.gh-issue-119690.8q6e1p.rst Normal file
View File

@ -0,0 +1 @@
Adds Unicode support and fixes audit events for ``_winapi.CreateNamedPipe``.

31
Modules/_winapi.c
View File

@ -224,7 +224,6 @@ create_converter('LPCVOID', '" F_POINTER "')
create_converter('BOOL', 'i') # F_BOOL used previously (always 'i')
create_converter('DWORD', 'k') # F_DWORD is always "k" (which is much shorter)
create_converter('LPCTSTR', 's')
create_converter('UINT', 'I') # F_UINT used previously (always 'I')
class LPCWSTR_converter(Py_UNICODE_converter):
@ -259,7 +258,7 @@ class LPVOID_return_converter(CReturnConverter):
data.return_conversion.append(
'return_value = HANDLE_TO_PYNUM(_return_value);\n')
[python start generated code]*/
/*[python end generated code: output=da39a3ee5e6b4b0d input=ef52a757a1830d92]*/
/*[python end generated code: output=da39a3ee5e6b4b0d input=da0a4db751936ee7]*/
#include "clinic/_winapi.c.h"
@ -530,7 +529,7 @@ _winapi_CreateFile_impl(PyObject *module, LPCWSTR file_name,
{
HANDLE handle;
if (PySys_Audit("_winapi.CreateFile", "uIIII",
if (PySys_Audit("_winapi.CreateFile", "ukkkk",
file_name, desired_access, share_mode,
creation_disposition, flags_and_attributes) < 0) {
return INVALID_HANDLE_VALUE;
@ -777,7 +776,7 @@ _winapi_CreateMutexW_impl(PyObject *module,
/*[clinic input]
_winapi.CreateNamedPipe -> HANDLE
name: LPCTSTR
name: LPCWSTR
open_mode: DWORD
pipe_mode: DWORD
max_instances: DWORD
@ -789,22 +788,22 @@ _winapi.CreateNamedPipe -> HANDLE
[clinic start generated code]*/
static HANDLE
_winapi_CreateNamedPipe_impl(PyObject *module, LPCTSTR name, DWORD open_mode,
_winapi_CreateNamedPipe_impl(PyObject *module, LPCWSTR name, DWORD open_mode,
DWORD pipe_mode, DWORD max_instances,
DWORD out_buffer_size, DWORD in_buffer_size,
DWORD default_timeout,
LPSECURITY_ATTRIBUTES security_attributes)
/*[clinic end generated code: output=80f8c07346a94fbc input=5a73530b84d8bc37]*/
/*[clinic end generated code: output=7d6fde93227680ba input=5bd4e4a55639ee02]*/
{
HANDLE handle;
if (PySys_Audit("_winapi.CreateNamedPipe", "uII",
if (PySys_Audit("_winapi.CreateNamedPipe", "ukk",
name, open_mode, pipe_mode) < 0) {
return INVALID_HANDLE_VALUE;
}
Py_BEGIN_ALLOW_THREADS
handle = CreateNamedPipe(name, open_mode, pipe_mode,
handle = CreateNamedPipeW(name, open_mode, pipe_mode,
max_instances, out_buffer_size,
in_buffer_size, default_timeout,
security_attributes);
@ -1790,7 +1789,7 @@ _winapi_OpenEventW_impl(PyObject *module, DWORD desired_access,
{
HANDLE handle;
if (PySys_Audit("_winapi.OpenEventW", "Iu", desired_access, name) < 0) {
if (PySys_Audit("_winapi.OpenEventW", "ku", desired_access, name) < 0) {
return INVALID_HANDLE_VALUE;
}
@ -1821,7 +1820,7 @@ _winapi_OpenMutexW_impl(PyObject *module, DWORD desired_access,
{
HANDLE handle;
if (PySys_Audit("_winapi.OpenMutexW", "Iu", desired_access, name) < 0) {
if (PySys_Audit("_winapi.OpenMutexW", "ku", desired_access, name) < 0) {
return INVALID_HANDLE_VALUE;
}
@ -1882,7 +1881,7 @@ _winapi_OpenProcess_impl(PyObject *module, DWORD desired_access,
{
HANDLE handle;
if (PySys_Audit("_winapi.OpenProcess", "II",
if (PySys_Audit("_winapi.OpenProcess", "kk",
process_id, desired_access) < 0) {
return INVALID_HANDLE_VALUE;
}
@ -2236,19 +2235,19 @@ _winapi_VirtualQuerySize_impl(PyObject *module, LPCVOID address)
/*[clinic input]
_winapi.WaitNamedPipe
name: LPCTSTR
name: LPCWSTR
timeout: DWORD
/
[clinic start generated code]*/
static PyObject *
_winapi_WaitNamedPipe_impl(PyObject *module, LPCTSTR name, DWORD timeout)
/*[clinic end generated code: output=c2866f4439b1fe38 input=36fc781291b1862c]*/
_winapi_WaitNamedPipe_impl(PyObject *module, LPCWSTR name, DWORD timeout)
/*[clinic end generated code: output=e161e2e630b3e9c2 input=099a4746544488fa]*/
{
BOOL success;
Py_BEGIN_ALLOW_THREADS
success = WaitNamedPipe(name, timeout);
success = WaitNamedPipeW(name, timeout);
Py_END_ALLOW_THREADS
if (!success)
@ -2917,7 +2916,7 @@ _winapi_CopyFile2_impl(PyObject *module, LPCWSTR existing_file_name,
HRESULT hr;
COPYFILE2_EXTENDED_PARAMETERS params = { sizeof(COPYFILE2_EXTENDED_PARAMETERS) };
if (PySys_Audit("_winapi.CopyFile2", "uuI",
if (PySys_Audit("_winapi.CopyFile2", "uuk",
existing_file_name, new_file_name, flags) < 0) {
return NULL;
}

24
Modules/clinic/_winapi.c.h generated
View File

@ -445,7 +445,7 @@ PyDoc_STRVAR(_winapi_CreateNamedPipe__doc__,
{"CreateNamedPipe", _PyCFunction_CAST(_winapi_CreateNamedPipe), METH_FASTCALL, _winapi_CreateNamedPipe__doc__},
static HANDLE
_winapi_CreateNamedPipe_impl(PyObject *module, LPCTSTR name, DWORD open_mode,
_winapi_CreateNamedPipe_impl(PyObject *module, LPCWSTR name, DWORD open_mode,
DWORD pipe_mode, DWORD max_instances,
DWORD out_buffer_size, DWORD in_buffer_size,
DWORD default_timeout,
@ -455,7 +455,7 @@ static PyObject *
_winapi_CreateNamedPipe(PyObject *module, PyObject *const *args, Py_ssize_t nargs)
{
PyObject *return_value = NULL;
LPCTSTR name;
LPCWSTR name = NULL;
DWORD open_mode;
DWORD pipe_mode;
DWORD max_instances;
@ -465,8 +465,8 @@ _winapi_CreateNamedPipe(PyObject *module, PyObject *const *args, Py_ssize_t narg
LPSECURITY_ATTRIBUTES security_attributes;
HANDLE _return_value;
if (!_PyArg_ParseStack(args, nargs, "skkkkkk" F_POINTER ":CreateNamedPipe",
&name, &open_mode, &pipe_mode, &max_instances, &out_buffer_size, &in_buffer_size, &default_timeout, &security_attributes)) {
if (!_PyArg_ParseStack(args, nargs, "O&kkkkkk" F_POINTER ":CreateNamedPipe",
_PyUnicode_WideCharString_Converter, &name, &open_mode, &pipe_mode, &max_instances, &out_buffer_size, &in_buffer_size, &default_timeout, &security_attributes)) {
goto exit;
}
_return_value = _winapi_CreateNamedPipe_impl(module, name, open_mode, pipe_mode, max_instances, out_buffer_size, in_buffer_size, default_timeout, security_attributes);
@ -479,6 +479,9 @@ _winapi_CreateNamedPipe(PyObject *module, PyObject *const *args, Py_ssize_t narg
return_value = HANDLE_TO_PYNUM(_return_value);
exit:
/* Cleanup for name */
PyMem_Free((void *)name);
return return_value;
}
@ -1660,22 +1663,25 @@ PyDoc_STRVAR(_winapi_WaitNamedPipe__doc__,
{"WaitNamedPipe", _PyCFunction_CAST(_winapi_WaitNamedPipe), METH_FASTCALL, _winapi_WaitNamedPipe__doc__},
static PyObject *
_winapi_WaitNamedPipe_impl(PyObject *module, LPCTSTR name, DWORD timeout);
_winapi_WaitNamedPipe_impl(PyObject *module, LPCWSTR name, DWORD timeout);
static PyObject *
_winapi_WaitNamedPipe(PyObject *module, PyObject *const *args, Py_ssize_t nargs)
{
PyObject *return_value = NULL;
LPCTSTR name;
LPCWSTR name = NULL;
DWORD timeout;
if (!_PyArg_ParseStack(args, nargs, "sk:WaitNamedPipe",
&name, &timeout)) {
if (!_PyArg_ParseStack(args, nargs, "O&k:WaitNamedPipe",
_PyUnicode_WideCharString_Converter, &name, &timeout)) {
goto exit;
}
return_value = _winapi_WaitNamedPipe_impl(module, name, timeout);
exit:
/* Cleanup for name */
PyMem_Free((void *)name);
return return_value;
}
@ -2118,4 +2124,4 @@ exit:
return return_value;
}
/*[clinic end generated code: output=ed94a2482ede3744 input=a9049054013a1b77]*/
/*[clinic end generated code: output=2304c62187a90140 input=a9049054013a1b77]*/