#![allow(non_snake_case)]
pub(crate) use _winapi::module_def;
#[pymodule]
mod _winapi {
use crate::{
Py, PyObjectRef, PyPayload, PyResult, TryFromObject, VirtualMachine,
builtins::PyStrRef,
common::lock::{PyMutex, PyMutexGuard},
convert::ToPyException,
exceptions::nul_char_error,
function::{ArgMapping, ArgSequence, OptionalArg},
types::Constructor,
windows::{WinHandle, WindowsSysResult},
};
use core::ptr::null_mut;
use rustpython_common::wtf8::Wtf8Buf;
use rustpython_host_env::overlapped as host_overlapped;
use rustpython_host_env::winapi as host_winapi;
use rustpython_host_env::windows::ToWideString;
#[pyattr]
use host_winapi::{
ABOVE_NORMAL_PRIORITY_CLASS, BELOW_NORMAL_PRIORITY_CLASS,
COPY_FILE_ALLOW_DECRYPTED_DESTINATION, COPY_FILE_COPY_SYMLINK, COPY_FILE_DIRECTORY,
COPY_FILE_FAIL_IF_EXISTS, COPY_FILE_NO_BUFFERING, COPY_FILE_NO_OFFLOAD,
COPY_FILE_OPEN_SOURCE_FOR_WRITE, COPY_FILE_REQUEST_COMPRESSED_TRAFFIC,
COPY_FILE_REQUEST_SECURITY_PRIVILEGES, COPY_FILE_RESTARTABLE, COPY_FILE_RESUME_FROM_PAUSE,
COPYFILE2_CALLBACK_CHUNK_FINISHED, COPYFILE2_CALLBACK_CHUNK_STARTED,
COPYFILE2_CALLBACK_ERROR, COPYFILE2_CALLBACK_POLL_CONTINUE,
COPYFILE2_CALLBACK_STREAM_FINISHED, COPYFILE2_CALLBACK_STREAM_STARTED,
COPYFILE2_PROGRESS_CANCEL, COPYFILE2_PROGRESS_CONTINUE, COPYFILE2_PROGRESS_PAUSE,
COPYFILE2_PROGRESS_QUIET, COPYFILE2_PROGRESS_STOP, CREATE_BREAKAWAY_FROM_JOB,
CREATE_DEFAULT_ERROR_MODE, CREATE_NEW_CONSOLE, CREATE_NEW_PROCESS_GROUP, CREATE_NO_WINDOW,
DETACHED_PROCESS, DUPLICATE_CLOSE_SOURCE, DUPLICATE_SAME_ACCESS, ERROR_ACCESS_DENIED,
ERROR_ALREADY_EXISTS, ERROR_BROKEN_PIPE, ERROR_IO_PENDING, ERROR_MORE_DATA,
ERROR_NETNAME_DELETED, ERROR_NO_DATA, ERROR_NO_SYSTEM_RESOURCES, ERROR_OPERATION_ABORTED,
ERROR_PIPE_BUSY, ERROR_PIPE_CONNECTED, ERROR_PRIVILEGE_NOT_HELD, ERROR_SEM_TIMEOUT,
FILE_FLAG_FIRST_PIPE_INSTANCE, FILE_FLAG_OVERLAPPED, FILE_GENERIC_READ, FILE_GENERIC_WRITE,
FILE_MAP_ALL_ACCESS, FILE_MAP_COPY, FILE_MAP_EXECUTE, FILE_MAP_READ, FILE_MAP_WRITE,
FILE_TYPE_CHAR, FILE_TYPE_DISK, FILE_TYPE_PIPE, FILE_TYPE_REMOTE, FILE_TYPE_UNKNOWN,
GENERIC_READ, GENERIC_WRITE, HIGH_PRIORITY_CLASS, IDLE_PRIORITY_CLASS, LCMAP_FULLWIDTH,
LCMAP_HALFWIDTH, LCMAP_HIRAGANA, LCMAP_KATAKANA, LCMAP_LINGUISTIC_CASING, LCMAP_LOWERCASE,
LCMAP_SIMPLIFIED_CHINESE, LCMAP_TITLECASE, LCMAP_TRADITIONAL_CHINESE, LCMAP_UPPERCASE,
LOCALE_NAME_MAX_LENGTH, MEM_COMMIT, MEM_FREE, MEM_IMAGE, MEM_MAPPED, MEM_PRIVATE,
MEM_RESERVE, NMPWAIT_WAIT_FOREVER, NORMAL_PRIORITY_CLASS, OPEN_EXISTING, PAGE_EXECUTE,
PAGE_EXECUTE_READ, PAGE_EXECUTE_READWRITE, PAGE_EXECUTE_WRITECOPY, PAGE_GUARD,
PAGE_NOACCESS, PAGE_NOCACHE, PAGE_READONLY, PAGE_READWRITE, PAGE_WRITECOMBINE,
PAGE_WRITECOPY, PIPE_ACCESS_DUPLEX, PIPE_ACCESS_INBOUND, PIPE_READMODE_MESSAGE,
PIPE_TYPE_MESSAGE, PIPE_UNLIMITED_INSTANCES, PIPE_WAIT, PROCESS_ALL_ACCESS,
PROCESS_DUP_HANDLE, REALTIME_PRIORITY_CLASS, SEC_COMMIT, SEC_IMAGE, SEC_LARGE_PAGES,
SEC_NOCACHE, SEC_RESERVE, SEC_WRITECOMBINE, STARTF_FORCEOFFFEEDBACK,
STARTF_FORCEONFEEDBACK, STARTF_PREVENTPINNING, STARTF_RUNFULLSCREEN, STARTF_TITLEISAPPID,
STARTF_TITLEISLINKNAME, STARTF_UNTRUSTEDSOURCE, STARTF_USECOUNTCHARS,
STARTF_USEFILLATTRIBUTE, STARTF_USEHOTKEY, STARTF_USEPOSITION, STARTF_USESHOWWINDOW,
STARTF_USESIZE, STARTF_USESTDHANDLES, STD_ERROR_HANDLE, STD_INPUT_HANDLE,
STD_OUTPUT_HANDLE, STILL_ACTIVE, SW_HIDE, SYNCHRONIZE, WAIT_ABANDONED_0, WAIT_OBJECT_0,
WAIT_TIMEOUT,
};
#[pyattr]
const NULL: isize = 0;
#[pyattr]
const INVALID_HANDLE_VALUE: isize = host_overlapped::INVALID_HANDLE_VALUE_ISIZE;
#[pyattr]
const INFINITE: u32 = host_winapi::INFINITE_TIMEOUT;
#[pyfunction]
fn CloseHandle(handle: WinHandle) -> WindowsSysResult<i32> {
WindowsSysResult(host_winapi::close_handle(handle.0))
}
#[expect(
clippy::too_many_arguments,
reason = "matches Win32 CreateFile parameter structure"
)]
#[pyfunction]
fn CreateFile(
file_name: PyStrRef,
desired_access: u32,
share_mode: u32,
_security_attributes: PyObjectRef, creation_disposition: u32,
flags_and_attributes: u32,
_template_file: PyObjectRef, vm: &VirtualMachine,
) -> PyResult<WinHandle> {
let file_name_wide = file_name
.as_wtf8()
.to_wide_cstring()
.map_err(|_| nul_char_error(vm))?;
host_winapi::create_file_w(
&file_name_wide,
desired_access,
share_mode,
creation_disposition,
flags_and_attributes,
)
.map(WinHandle)
.map_err(|e| e.to_pyexception(vm))
}
#[pyfunction]
fn GetStdHandle(
std_handle: host_winapi::StdHandle,
vm: &VirtualMachine,
) -> PyResult<Option<WinHandle>> {
host_winapi::get_std_handle(std_handle)
.map(|handle| handle.map(WinHandle))
.map_err(|e| e.to_pyexception(vm))
}
#[pyfunction]
fn CreatePipe(
_pipe_attrs: PyObjectRef,
size: u32,
vm: &VirtualMachine,
) -> PyResult<(WinHandle, WinHandle)> {
host_winapi::create_pipe(size)
.map(|(read, write)| (WinHandle(read), WinHandle(write)))
.map_err(|e| e.to_pyexception(vm))
}
#[pyfunction]
fn DuplicateHandle(
src_process: WinHandle,
src: WinHandle,
target_process: WinHandle,
access: u32,
inherit: i32,
options: OptionalArg<u32>,
vm: &VirtualMachine,
) -> PyResult<WinHandle> {
host_winapi::duplicate_handle(
src_process.0,
src.0,
target_process.0,
access,
inherit,
options.unwrap_or(0),
)
.map(WinHandle)
.map_err(|e| e.to_pyexception(vm))
}
#[pyfunction]
fn GetACP() -> u32 {
host_winapi::get_acp()
}
#[pyfunction]
fn GetCurrentProcess() -> WinHandle {
WinHandle(host_winapi::get_current_process())
}
#[pyfunction]
fn GetFileType(h: WinHandle, vm: &VirtualMachine) -> PyResult<host_winapi::FileType> {
host_winapi::get_file_type(h.0).map_err(|e| e.to_pyexception(vm))
}
#[pyfunction]
fn GetLastError() -> u32 {
host_winapi::get_last_error()
}
#[pyfunction]
fn GetVersion() -> u32 {
host_winapi::get_version()
}
#[derive(FromArgs)]
struct CreateProcessArgs {
#[pyarg(positional)]
name: Option<PyStrRef>,
#[pyarg(positional)]
command_line: Option<PyStrRef>,
#[pyarg(positional)]
_proc_attrs: PyObjectRef,
#[pyarg(positional)]
_thread_attrs: PyObjectRef,
#[pyarg(positional)]
inherit_handles: i32,
#[pyarg(positional)]
creation_flags: u32,
#[pyarg(positional)]
env_mapping: Option<ArgMapping>,
#[pyarg(positional)]
current_dir: Option<PyStrRef>,
#[pyarg(positional)]
startup_info: PyObjectRef,
}
#[pyfunction]
fn CreateProcess(
args: CreateProcessArgs,
vm: &VirtualMachine,
) -> PyResult<(WinHandle, WinHandle, u32, u32)> {
macro_rules! si_attr {
($attr:ident, $t:ty) => {{
<Option<$t>>::try_from_object(
vm,
args.startup_info.get_attr(stringify!($attr), vm)?,
)?
.unwrap_or(0) as _
}};
($attr:ident) => {{
<Option<_>>::try_from_object(
vm,
args.startup_info.get_attr(stringify!($attr), vm)?,
)?
.unwrap_or(0)
}};
}
let startup_info = host_winapi::StartupInfoData {
flags: si_attr!(dwFlags),
show_window: si_attr!(wShowWindow),
std_input: si_attr!(hStdInput, isize),
std_output: si_attr!(hStdOutput, isize),
std_error: si_attr!(hStdError, isize),
};
let env = args
.env_mapping
.map(|m| getenvironment(m, vm))
.transpose()?;
let handle_list = get_handle_list(args.startup_info.get_attr("lpAttributeList", vm)?, vm)?;
let wcstring = |s: PyStrRef| {
s.as_wtf8()
.to_wide_cstring()
.map_err(|_| nul_char_error(vm))
};
let app_name = args.name.map(wcstring).transpose()?;
let current_dir = args.current_dir.map(wcstring).transpose()?;
let mut command_line = args
.command_line
.map(|s| wcstring(s).map(widestring::WideCString::into_vec_with_nul))
.transpose()
.map_err(|_| nul_char_error(vm))?;
let procinfo = host_winapi::create_process(
app_name.as_deref(),
command_line.as_deref_mut(),
args.inherit_handles,
args.creation_flags,
env.as_deref(),
current_dir.as_deref(),
startup_info,
handle_list,
)
.map_err(|e| e.to_pyexception(vm))?;
Ok((
WinHandle(procinfo.process),
WinHandle(procinfo.thread),
procinfo.process_id,
procinfo.thread_id,
))
}
#[pyfunction]
fn OpenProcess(
desired_access: u32,
inherit_handle: bool,
process_id: u32,
vm: &VirtualMachine,
) -> PyResult<WinHandle> {
host_winapi::open_process(desired_access, inherit_handle, process_id)
.map(WinHandle)
.map_err(|e| e.to_pyexception(vm))
}
#[pyfunction]
fn ExitProcess(exit_code: u32) {
host_winapi::exit_process(exit_code)
}
#[pyfunction]
fn NeedCurrentDirectoryForExePath(exe_name: PyStrRef, vm: &VirtualMachine) -> PyResult<bool> {
exe_name
.as_wtf8()
.to_wide_cstring()
.map(|exe_name| host_winapi::need_current_directory_for_exe_path_w(&exe_name))
.map_err(|_| nul_char_error(vm))
}
#[pyfunction]
fn CreateJunction(
src_path: PyStrRef,
dest_path: PyStrRef,
vm: &VirtualMachine,
) -> PyResult<()> {
let src_path = std::path::Path::new(src_path.expect_str());
let dest_path = std::path::Path::new(dest_path.expect_str());
host_winapi::create_junction(src_path, dest_path).map_err(|e| e.to_pyexception(vm))
}
fn getenvironment(env: ArgMapping, vm: &VirtualMachine) -> PyResult<Vec<u16>> {
let keys = env.mapping().keys(vm)?;
let values = env.mapping().values(vm)?;
let keys = ArgSequence::try_from_object(vm, keys)?.into_vec();
let values = ArgSequence::try_from_object(vm, values)?.into_vec();
if keys.len() != values.len() {
return Err(vm.new_runtime_error("environment changed size during iteration"));
}
let mut entries = Vec::with_capacity(keys.len());
for (k, v) in keys.into_iter().zip(values) {
let k = PyStrRef::try_from_object(vm, k)?;
let k = k.expect_str().to_owned();
let v = PyStrRef::try_from_object(vm, v)?;
let v = v.expect_str().to_owned();
entries.push((k, v));
}
host_winapi::build_environment_block(entries).map_err(|err| err.to_pyexception(vm))
}
fn get_handle_list(obj: PyObjectRef, vm: &VirtualMachine) -> PyResult<Option<Vec<usize>>> {
let Some(mapping) = <Option<ArgMapping>>::try_from_object(vm, obj)? else {
return Ok(None);
};
mapping
.as_ref()
.get_item("handle_list", vm)
.ok()
.and_then(|obj| {
<Option<ArgSequence<usize>>>::try_from_object(vm, obj)
.map(|s| match s {
Some(s) if !s.is_empty() => Some(s.into_vec()),
_ => None,
})
.transpose()
})
.transpose()
}
#[pyfunction]
fn WaitForSingleObject(h: WinHandle, ms: i64, vm: &VirtualMachine) -> PyResult<u32> {
let ms = if ms < 0 {
host_winapi::INFINITE_TIMEOUT
} else if ms > u32::MAX as i64 {
return Err(vm.new_overflow_error("timeout value is too large"));
} else {
ms as u32
};
vm.allow_threads(|| host_winapi::wait_for_single_object(h.0, ms))
.map_err(|e| e.to_pyexception(vm))
}
#[pyfunction]
fn WaitForMultipleObjects(
handle_seq: ArgSequence<isize>,
wait_all: bool,
milliseconds: u32,
vm: &VirtualMachine,
) -> PyResult<u32> {
let handles: Vec<host_winapi::Handle> = handle_seq
.into_vec()
.into_iter()
.map(|h| h as host_winapi::Handle)
.collect();
if handles.is_empty() {
return Err(vm.new_value_error("handle_seq must not be empty"));
}
if handles.len() > 64 {
return Err(vm.new_value_error("WaitForMultipleObjects supports at most 64 handles"));
}
vm.allow_threads(|| {
host_winapi::wait_for_multiple_objects(&handles, wait_all, milliseconds)
})
.map_err(|e| e.to_pyexception(vm))
}
#[pyfunction]
fn GetExitCodeProcess(h: WinHandle, vm: &VirtualMachine) -> PyResult<u32> {
host_winapi::get_exit_code_process(h.0).map_err(|e| e.to_pyexception(vm))
}
#[pyfunction]
fn TerminateProcess(h: WinHandle, exit_code: u32) -> WindowsSysResult<i32> {
WindowsSysResult(host_winapi::terminate_process(h.0, exit_code))
}
#[derive(FromArgs)]
struct OptionalJobName {
#[pyarg(positional, optional)]
name: Option<PyStrRef>,
}
#[pyfunction]
fn CreateJobObject(
_security_attributes: PyObjectRef,
name: OptionalJobName,
vm: &VirtualMachine,
) -> PyResult<WinHandle> {
let name = name
.name
.map(|name| name.as_wtf8().to_wide_cstring())
.transpose()
.map_err(|_| nul_char_error(vm))?;
host_winapi::create_job_object_w(name.as_deref())
.map(WinHandle)
.map_err(|e| e.to_pyexception(vm))
}
#[pyfunction]
fn AssignProcessToJobObject(
job: WinHandle,
process: WinHandle,
vm: &VirtualMachine,
) -> PyResult<()> {
host_winapi::assign_process_to_job_object(job.0, process.0)
.map_err(|e| e.to_pyexception(vm))
}
#[pyfunction]
fn TerminateJobObject(job: WinHandle, exit_code: u32, vm: &VirtualMachine) -> PyResult<()> {
host_winapi::terminate_job_object(job.0, exit_code).map_err(|e| e.to_pyexception(vm))
}
#[pyfunction]
fn SetJobObjectKillOnClose(job: WinHandle, vm: &VirtualMachine) -> PyResult<()> {
host_winapi::set_job_object_kill_on_close(job.0).map_err(|e| e.to_pyexception(vm))
}
#[pyfunction]
fn GetModuleFileName(handle: isize, vm: &VirtualMachine) -> PyResult<String> {
let mut path: Vec<u16> = vec![0; host_winapi::MAX_PATH_USIZE];
let length = host_winapi::get_module_file_name(handle as _, &mut path);
if length == 0 {
return Err(vm.new_runtime_error("GetModuleFileName failed"));
}
let (path, _) = path.split_at(length as usize);
Ok(String::from_utf16(path).unwrap())
}
#[pyfunction]
fn OpenMutexW(
desired_access: u32,
inherit_handle: bool,
name: PyStrRef,
vm: &VirtualMachine,
) -> PyResult<WinHandle> {
let name = name
.as_wtf8()
.to_wide_cstring()
.map_err(|_| nul_char_error(vm))?;
host_winapi::open_mutex_w(desired_access, inherit_handle, &name)
.map(WinHandle)
.map_err(|e| e.to_pyexception(vm))
}
#[pyfunction]
fn ReleaseMutex(handle: WinHandle) -> WindowsSysResult<i32> {
WindowsSysResult(host_winapi::release_mutex(handle.0))
}
#[pyattr]
use host_winapi::{LOCALE_NAME_INVARIANT, LOCALE_NAME_SYSTEM_DEFAULT};
#[pyattr(name = "LOCALE_NAME_USER_DEFAULT")]
fn locale_name_user_default(vm: &VirtualMachine) -> PyObjectRef {
vm.ctx.none()
}
#[pyfunction]
fn LCMapStringEx(
locale: PyStrRef,
flags: u32,
src: PyStrRef,
vm: &VirtualMachine,
) -> PyResult<PyStrRef> {
if flags
& (host_winapi::LCMAP_SORTHANDLE_FLAG
| host_winapi::LCMAP_HASH_FLAG
| host_winapi::LCMAP_BYTEREV_FLAG
| host_winapi::LCMAP_SORTKEY_FLAG)
!= 0
{
return Err(vm.new_value_error("unsupported flags"));
}
let locale_wide = locale
.as_wtf8()
.to_wide_cstring()
.map_err(|_| nul_char_error(vm))?;
let src_wide: Vec<_> = src.as_wtf8().encode_wide().collect();
if src_wide.len() > i32::MAX as usize {
return Err(vm.new_overflow_error("input string is too long"));
}
let dest = host_winapi::lc_map_string_ex(&locale_wide, flags, &src_wide)
.map_err(|e| e.to_pyexception(vm))?;
let result = Wtf8Buf::from_wide(&dest);
Ok(vm.ctx.new_str(result))
}
#[derive(FromArgs)]
struct CreateNamedPipeArgs {
#[pyarg(positional)]
name: PyStrRef,
#[pyarg(positional)]
open_mode: u32,
#[pyarg(positional)]
pipe_mode: u32,
#[pyarg(positional)]
max_instances: u32,
#[pyarg(positional)]
out_buffer_size: u32,
#[pyarg(positional)]
in_buffer_size: u32,
#[pyarg(positional)]
default_timeout: u32,
#[pyarg(positional)]
_security_attributes: PyObjectRef, }
#[pyfunction]
fn CreateNamedPipe(args: CreateNamedPipeArgs, vm: &VirtualMachine) -> PyResult<WinHandle> {
let name = args
.name
.as_wtf8()
.to_wide_cstring()
.map_err(|_| nul_char_error(vm))?;
host_winapi::create_named_pipe_w(
&name,
args.open_mode,
args.pipe_mode,
args.max_instances,
args.out_buffer_size,
args.in_buffer_size,
args.default_timeout,
)
.map(WinHandle)
.map_err(|e| e.to_pyexception(vm))
}
#[pyattr]
#[pyclass(name = "Overlapped", module = "_winapi")]
#[derive(Debug, PyPayload)]
struct Overlapped {
inner: PyMutex<host_overlapped::Operation>,
}
impl Overlapped {
fn new_with_handle(handle: host_winapi::Handle, vm: &VirtualMachine) -> PyResult<Self> {
host_overlapped::Operation::new(handle)
.map(|inner| Self {
inner: PyMutex::new(inner),
})
.map_err(|e| e.to_pyexception(vm))
}
fn lock_inner(&self, vm: &VirtualMachine) -> PyMutexGuard<'_, host_overlapped::Operation> {
vm.allow_threads(|| self.inner.lock())
}
}
#[pyclass(with(Constructor))]
impl Py<Overlapped> {
#[pymethod]
fn GetOverlappedResult(&self, wait: bool, vm: &VirtualMachine) -> PyResult<(u32, u32)> {
let mut inner = self.lock_inner(vm);
vm.allow_threads(|| inner.get_result(wait))
.map(|result| (result.transferred, result.error))
.map_err(|e| e.to_pyexception(vm))
}
#[pymethod]
fn getbuffer(&self, vm: &VirtualMachine) -> PyResult<Option<PyObjectRef>> {
let inner = self.lock_inner(vm);
if !inner.is_completed() {
return Err(vm.new_value_error(
"can't get read buffer before GetOverlappedResult() signals the operation completed",
));
}
Ok(inner
.read_buffer()
.map(|buf| vm.ctx.new_bytes(buf.to_vec()).into()))
}
#[pymethod]
fn cancel(&self, vm: &VirtualMachine) -> PyResult<()> {
let mut inner = self.lock_inner(vm);
inner.cancel().map_err(|e| e.to_pyexception(vm))
}
#[pygetset]
fn event(&self, vm: &VirtualMachine) -> isize {
let inner = self.lock_inner(vm);
inner.event() as isize
}
}
impl Constructor for Overlapped {
type Args = ();
fn py_new(
_cls: &Py<crate::builtins::PyType>,
_args: Self::Args,
vm: &VirtualMachine,
) -> PyResult<Self> {
Self::new_with_handle(null_mut(), vm)
}
}
#[derive(FromArgs)]
struct ConnectNamedPipeArgs {
#[pyarg(any)]
handle: WinHandle,
#[pyarg(any, optional)]
overlapped: OptionalArg<bool>,
}
#[pyfunction]
fn ConnectNamedPipe(args: ConnectNamedPipeArgs, vm: &VirtualMachine) -> PyResult<PyObjectRef> {
let handle = args.handle;
let use_overlapped = args.overlapped.unwrap_or(false);
if use_overlapped {
let ov = Overlapped::new_with_handle(handle.0, vm)?;
{
let mut inner = ov.inner.lock();
inner
.connect_named_pipe()
.map_err(|e| e.to_pyexception(vm))?;
}
Ok(ov.into_pyobject(vm))
} else {
vm.allow_threads(|| host_winapi::connect_named_pipe(handle.0))
.map_err(|e| e.to_pyexception(vm))?;
Ok(vm.ctx.none())
}
}
fn path_name_result_to_pystr(wide: Vec<u16>, vm: &VirtualMachine) -> PyStrRef {
let result_str = Wtf8Buf::from_wide(&wide);
vm.ctx.new_str(result_str)
}
#[pyfunction]
fn GetShortPathName(path: PyStrRef, vm: &VirtualMachine) -> PyResult<PyStrRef> {
let path = path
.as_wtf8()
.to_wide_cstring()
.map_err(|_| nul_char_error(vm))?;
let wide = host_winapi::get_short_path_name_w(&path).map_err(|e| e.to_pyexception(vm))?;
Ok(path_name_result_to_pystr(wide, vm))
}
#[pyfunction]
fn GetLongPathName(path: PyStrRef, vm: &VirtualMachine) -> PyResult<PyStrRef> {
let path = path
.as_wtf8()
.to_wide_cstring()
.map_err(|_| nul_char_error(vm))?;
let wide = host_winapi::get_long_path_name_w(&path).map_err(|e| e.to_pyexception(vm))?;
Ok(path_name_result_to_pystr(wide, vm))
}
#[pyfunction]
fn WaitNamedPipe(name: PyStrRef, timeout: u32, vm: &VirtualMachine) -> PyResult<()> {
let name = name
.as_wtf8()
.to_wide_cstring()
.map_err(|_| nul_char_error(vm))?;
host_winapi::wait_named_pipe_w(&name, timeout).map_err(|e| e.to_pyexception(vm))
}
#[pyfunction]
fn PeekNamedPipe(
handle: WinHandle,
size: OptionalArg<i32>,
vm: &VirtualMachine,
) -> PyResult<PyObjectRef> {
let size = size.unwrap_or(0);
if size < 0 {
return Err(vm.new_value_error("negative size"));
}
if size > 0 {
let result = host_winapi::peek_named_pipe(handle.0, Some(size as u32))
.map_err(|e| e.to_pyexception(vm))?;
let buf = result.data.unwrap_or_default();
let bytes: PyObjectRef = vm.ctx.new_bytes(buf).into();
Ok(vm
.ctx
.new_tuple(vec![
bytes,
vm.ctx.new_int(result.available).into(),
vm.ctx.new_int(result.left_this_message).into(),
])
.into())
} else {
let result =
host_winapi::peek_named_pipe(handle.0, None).map_err(|e| e.to_pyexception(vm))?;
Ok(vm
.ctx
.new_tuple(vec![
vm.ctx.new_int(result.available).into(),
vm.ctx.new_int(result.left_this_message).into(),
])
.into())
}
}
#[pyfunction]
fn CreateEventW(
security_attributes: isize, manual_reset: bool,
initial_state: bool,
name: Option<PyStrRef>,
vm: &VirtualMachine,
) -> PyResult<WinHandle> {
let _ = security_attributes;
let name = name
.map(|n| n.as_wtf8().to_wide_cstring())
.transpose()
.map_err(|_| nul_char_error(vm))?;
host_winapi::create_event_w(manual_reset, initial_state, name.as_deref())
.map(WinHandle)
.map_err(|e| e.to_pyexception(vm))
}
#[pyfunction]
fn SetEvent(event: WinHandle, vm: &VirtualMachine) -> PyResult<()> {
host_winapi::set_event(event.0).map_err(|e| e.to_pyexception(vm))
}
#[derive(FromArgs)]
struct WriteFileArgs {
#[pyarg(any)]
handle: WinHandle,
#[pyarg(any)]
buffer: crate::function::ArgBytesLike,
#[pyarg(any, default)]
overlapped: bool,
}
#[pyfunction]
fn WriteFile(args: WriteFileArgs, vm: &VirtualMachine) -> PyResult<PyObjectRef> {
let handle = args.handle;
let use_overlapped = args.overlapped;
let buf = args.buffer.borrow_buf();
if use_overlapped {
let ov = Overlapped::new_with_handle(handle.0, vm)?;
let err = {
let mut inner = ov.inner.lock();
inner.write(&buf).map_err(|e| e.to_pyexception(vm))?
};
let result = vm
.ctx
.new_tuple(vec![ov.into_pyobject(vm), vm.ctx.new_int(err).into()]);
return Ok(result.into());
}
let result = host_winapi::write_file(handle.0, &buf).map_err(|e| e.to_pyexception(vm))?;
Ok(vm
.ctx
.new_tuple(vec![
vm.ctx.new_int(result.written).into(),
vm.ctx.new_int(result.error).into(),
])
.into())
}
#[derive(FromArgs)]
struct ReadFileArgs {
#[pyarg(any)]
handle: WinHandle,
#[pyarg(any)]
size: u32,
#[pyarg(any, default)]
overlapped: bool,
}
#[pyfunction]
fn ReadFile(args: ReadFileArgs, vm: &VirtualMachine) -> PyResult<PyObjectRef> {
let handle = args.handle;
let size = args.size;
let use_overlapped = args.overlapped;
if use_overlapped {
let ov = Overlapped::new_with_handle(handle.0, vm)?;
let err = {
let mut inner = ov.inner.lock();
inner.read(size).map_err(|e| e.to_pyexception(vm))?
};
let result = vm
.ctx
.new_tuple(vec![ov.into_pyobject(vm), vm.ctx.new_int(err).into()]);
return Ok(result.into());
}
let result = vm
.allow_threads(|| host_winapi::read_file(handle.0, size))
.map_err(|e| e.to_pyexception(vm))?;
Ok(vm
.ctx
.new_tuple(vec![
vm.ctx.new_bytes(result.data).into(),
vm.ctx.new_int(result.error).into(),
])
.into())
}
#[pyfunction]
fn SetNamedPipeHandleState(
named_pipe: WinHandle,
mode: PyObjectRef,
max_collection_count: PyObjectRef,
collect_data_timeout: PyObjectRef,
vm: &VirtualMachine,
) -> PyResult<()> {
let objs = [&mode, &max_collection_count, &collect_data_timeout];
let mut values = [None; 3];
for (index, obj) in objs.iter().enumerate() {
if !vm.is_none(obj) {
values[index] = Some(u32::try_from_object(vm, (*obj).clone())?);
}
}
host_winapi::set_named_pipe_handle_state(named_pipe.0, values[0], values[1], values[2])
.map_err(|e| e.to_pyexception(vm))
}
#[pyfunction]
fn ResetEvent(event: WinHandle, vm: &VirtualMachine) -> PyResult<()> {
host_winapi::reset_event(event.0).map_err(|e| e.to_pyexception(vm))
}
#[pyfunction]
fn CreateMutexW(
security_attributes: isize,
initial_owner: bool,
name: Option<PyStrRef>,
vm: &VirtualMachine,
) -> PyResult<WinHandle> {
let _ = security_attributes;
let name = name
.map(|n| n.as_wtf8().to_wide_cstring())
.transpose()
.map_err(|_| nul_char_error(vm))?;
host_winapi::create_mutex_w(initial_owner, name.as_deref())
.map(WinHandle)
.map_err(|e| e.to_pyexception(vm))
}
#[pyfunction]
fn OpenEventW(
desired_access: u32,
inherit_handle: bool,
name: PyStrRef,
vm: &VirtualMachine,
) -> PyResult<WinHandle> {
let name = name
.as_wtf8()
.to_wide_cstring()
.map_err(|_| nul_char_error(vm))?;
host_winapi::open_event_w(desired_access, inherit_handle, &name)
.map(WinHandle)
.map_err(|e| e.to_pyexception(vm))
}
const MAXIMUM_WAIT_OBJECTS: usize = host_winapi::MAXIMUM_WAIT_OBJECTS as usize;
#[pyfunction]
fn BatchedWaitForMultipleObjects(
handle_seq: PyObjectRef,
wait_all: bool,
milliseconds: OptionalArg<u32>,
vm: &VirtualMachine,
) -> PyResult<PyObjectRef> {
let milliseconds = milliseconds.unwrap_or(host_winapi::INFINITE_TIMEOUT);
let seq = ArgSequence::<isize>::try_from_object(vm, handle_seq)?;
let handles: Vec<host_winapi::Handle> = seq
.into_vec()
.into_iter()
.map(|handle| handle as _)
.collect();
let nhandles = handles.len();
if nhandles == 0 {
return if wait_all {
Ok(vm.ctx.none())
} else {
Ok(vm.ctx.new_list(vec![]).into())
};
}
let max_total_objects = (MAXIMUM_WAIT_OBJECTS - 1) * (MAXIMUM_WAIT_OBJECTS - 1);
if nhandles > max_total_objects {
return Err(vm.new_value_error(format!(
"need at most {max_total_objects} handles, got a sequence of length {nhandles}"
)));
}
#[cfg(feature = "threading")]
let sigint_event = {
let is_main = crate::stdlib::_thread::get_ident() == vm.state.main_thread_ident.load();
if is_main {
let handle = crate::signal::get_sigint_event().map_or_else(
|| {
let handle = host_winapi::create_event_w(true, false, None)
.unwrap_or(core::ptr::null_mut());
if !handle.is_null() {
crate::signal::set_sigint_event(handle as isize);
}
handle
},
|handle| handle as host_winapi::Handle,
);
if handle.is_null() { None } else { Some(handle) }
} else {
None
}
};
#[cfg(not(feature = "threading"))]
let sigint_event: Option<host_winapi::Handle> = None;
let batched_result = vm.allow_threads(|| {
host_winapi::batched_wait_for_multiple_objects(
&handles,
wait_all,
milliseconds,
sigint_event,
)
});
match batched_result {
Ok(host_winapi::BatchedWaitResult::All) => Ok(vm.ctx.none()),
Ok(host_winapi::BatchedWaitResult::Indices(indices)) => Ok(vm
.ctx
.new_list(
indices
.into_iter()
.map(|index| vm.ctx.new_int(index).into())
.collect(),
)
.into()),
Err(err) => Err(err.to_pyexception(vm)),
}
}
#[pyfunction]
fn CreateFileMapping(
file_handle: WinHandle,
_security_attributes: PyObjectRef,
protect: u32,
max_size_high: u32,
max_size_low: u32,
name: Option<PyStrRef>,
vm: &VirtualMachine,
) -> PyResult<WinHandle> {
let name = name
.map(|n| n.as_wtf8().to_wide_cstring())
.transpose()
.map_err(|_| nul_char_error(vm))?;
host_winapi::create_file_mapping_w(
file_handle.0,
protect,
max_size_high,
max_size_low,
name.as_deref(),
)
.map(WinHandle)
.map_err(|e| e.to_pyexception(vm))
}
#[pyfunction]
fn OpenFileMapping(
desired_access: u32,
inherit_handle: bool,
name: PyStrRef,
vm: &VirtualMachine,
) -> PyResult<WinHandle> {
let name = name
.as_wtf8()
.to_wide_cstring()
.map_err(|_| nul_char_error(vm))?;
host_winapi::open_file_mapping_w(desired_access, inherit_handle, &name)
.map(WinHandle)
.map_err(|e| e.to_pyexception(vm))
}
#[pyfunction]
fn MapViewOfFile(
file_map: WinHandle,
desired_access: u32,
file_offset_high: u32,
file_offset_low: u32,
number_bytes: usize,
vm: &VirtualMachine,
) -> PyResult<isize> {
host_winapi::map_view_of_file(
file_map.0,
desired_access,
file_offset_high,
file_offset_low,
number_bytes,
)
.map_err(|e| e.to_pyexception(vm))
}
#[pyfunction]
fn UnmapViewOfFile(address: isize, vm: &VirtualMachine) -> PyResult<()> {
host_winapi::unmap_view_of_file(address).map_err(|e| e.to_pyexception(vm))
}
#[pyfunction]
fn VirtualQuerySize(address: isize, vm: &VirtualMachine) -> PyResult<usize> {
host_winapi::virtual_query_size(address).map_err(|e| e.to_pyexception(vm))
}
#[pyfunction]
fn CopyFile2(
existing_file_name: PyStrRef,
new_file_name: PyStrRef,
flags: u32,
_progress_routine: OptionalArg<PyObjectRef>,
vm: &VirtualMachine,
) -> PyResult<()> {
let src_wide = existing_file_name
.as_wtf8()
.to_wide_cstring()
.map_err(|_| nul_char_error(vm))?;
let dst_wide = new_file_name
.as_wtf8()
.to_wide_cstring()
.map_err(|_| nul_char_error(vm))?;
host_winapi::copy_file2(&src_wide, &dst_wide, flags).map_err(|e| e.to_pyexception(vm))
}
#[pyfunction]
fn _mimetypes_read_windows_registry(
on_type_read: PyObjectRef,
vm: &VirtualMachine,
) -> PyResult<()> {
host_winapi::read_windows_mimetype_registry_in_batches(|entries| {
for (mime_type, ext) in entries.drain(..) {
on_type_read.call((vm.ctx.new_str(mime_type), vm.ctx.new_str(ext)), vm)?;
}
Ok(())
})
.map_err(|err| match err {
host_winapi::MimeRegistryReadError::Os(err) => vm.new_os_error(err as i32),
host_winapi::MimeRegistryReadError::Callback(err) => err,
})
}
}