let module = module::make_module(vm);

super::os::extend_module(vm, &module);

module

use crate::{

PyResult, TryFromObject, VirtualMachine,

builtins::{PyBaseExceptionRef, PyDictRef, PyListRef, PyStrRef, PyTupleRef},

common::{crt_fd, os::last_os_error, suppress_iph, windows::ToWideString},

convert::ToPyException,

function::{Either, OptionalArg},

ospath::OsPath,

stdlib::os::{_os, DirFd, FollowSymlinks, SupportFunc, TargetIsDirectory, errno_err},

};

use libc::intptr_t;

use std::os::windows::io::AsRawHandle;

use std::{env, fs, io, mem::MaybeUninit, os::windows::ffi::OsStringExt};

use windows_sys::Win32::{

Foundation::{self, INVALID_HANDLE_VALUE},

Storage::FileSystem,

System::{Console, Threading},

};

#[pyattr]

use libc::{O_BINARY, O_NOINHERIT, O_RANDOM, O_SEQUENTIAL, O_TEMPORARY, O_TEXT};

// Windows spawn mode constants

#[pyattr]

const P_WAIT: i32 = 0;

#[pyattr]

const P_NOWAIT: i32 = 1;

#[pyattr]

const P_OVERLAY: i32 = 2;

#[pyattr]

const P_NOWAITO: i32 = 3;

#[pyattr]

const P_DETACH: i32 = 4;

// _O_SHORT_LIVED is not in libc, define manually

#[pyattr]

const O_SHORT_LIVED: i32 = 0x1000;

// Exit code constant

#[pyattr]

const EX_OK: i32 = 0;

// Maximum number of temporary files

#[pyattr]

const TMP_MAX: i32 = i32::MAX;

#[pyattr]

use windows_sys::Win32::System::LibraryLoader::{

LOAD_LIBRARY_SEARCH_APPLICATION_DIR as _LOAD_LIBRARY_SEARCH_APPLICATION_DIR,

LOAD_LIBRARY_SEARCH_DEFAULT_DIRS as _LOAD_LIBRARY_SEARCH_DEFAULT_DIRS,

LOAD_LIBRARY_SEARCH_DLL_LOAD_DIR as _LOAD_LIBRARY_SEARCH_DLL_LOAD_DIR,

LOAD_LIBRARY_SEARCH_SYSTEM32 as _LOAD_LIBRARY_SEARCH_SYSTEM32,

LOAD_LIBRARY_SEARCH_USER_DIRS as _LOAD_LIBRARY_SEARCH_USER_DIRS,

};

#[pyfunction]

pub(super) fn access(path: OsPath, mode: u8, vm: &VirtualMachine) -> PyResult<bool> {

let attr = unsafe { FileSystem::GetFileAttributesW(path.to_wide_cstring(vm)?.as_ptr()) };

Ok(attr != FileSystem::INVALID_FILE_ATTRIBUTES

&& (mode & 2 == 0

|| attr & FileSystem::FILE_ATTRIBUTE_READONLY == 0

|| attr & FileSystem::FILE_ATTRIBUTE_DIRECTORY != 0))

}

#[pyfunction]

pub(super) fn _supports_virtual_terminal() -> PyResult<bool> {

// TODO: implement this

Ok(true)

}

#[derive(FromArgs)]

pub(super) struct SymlinkArgs<'fd> {

src: OsPath,

dst: OsPath,

#[pyarg(flatten)]

target_is_directory: TargetIsDirectory,

#[pyarg(flatten)]

_dir_fd: DirFd<'fd, { _os::SYMLINK_DIR_FD as usize }>,

}

#[pyfunction]

pub(super) fn symlink(args: SymlinkArgs<'_>, vm: &VirtualMachine) -> PyResult<()> {

use std::os::windows::fs as win_fs;

let dir = args.target_is_directory.target_is_directory

|| args

.dst

.as_path()

.parent()

.and_then(|dst_parent| dst_parent.join(&args.src).symlink_metadata().ok())

.is_some_and(|meta| meta.is_dir());

let res = if dir {

win_fs::symlink_dir(args.src.path, args.dst.path)

} else {

win_fs::symlink_file(args.src.path, args.dst.path)

};

res.map_err(|err| err.to_pyexception(vm))

}

#[pyfunction]

fn set_inheritable(

fd: crt_fd::Borrowed<'_>,

inheritable: bool,

vm: &VirtualMachine,

) -> PyResult<()> {

let handle = crt_fd::as_handle(fd).map_err(|e| e.to_pyexception(vm))?;

set_handle_inheritable(handle.as_raw_handle() as _, inheritable, vm)

}

#[pyattr]

fn environ(vm: &VirtualMachine) -> PyDictRef {

let environ = vm.ctx.new_dict();

for (key, value) in env::vars() {

environ.set_item(&key, vm.new_pyobj(value), vm).unwrap();

}

environ

}

#[pyfunction]

fn chmod(

path: OsPath,

dir_fd: DirFd<'_, 0>,

mode: u32,

follow_symlinks: FollowSymlinks,

vm: &VirtualMachine,

) -> PyResult<()> {

const S_IWRITE: u32 = 128;

let [] = dir_fd.0;

let metadata = if follow_symlinks.0 {

fs::metadata(&path)

} else {

fs::symlink_metadata(&path)

};

let meta = metadata.map_err(|err| err.to_pyexception(vm))?;

let mut permissions = meta.permissions();

permissions.set_readonly(mode & S_IWRITE == 0);

fs::set_permissions(&path, permissions).map_err(|err| err.to_pyexception(vm))

}

// cwait is available on MSVC only (according to CPython)

#[cfg(target_env = "msvc")]

unsafe extern "C" {

fn _cwait(termstat: *mut i32, procHandle: intptr_t, action: i32) -> intptr_t;

let mut status: i32 = 0;

let pid = unsafe { suppress_iph!(_cwait(&mut status, pid, opt)) };

if pid == -1 {

Err(errno_err(vm))

} else {

// Cast to unsigned to handle large exit codes (like 0xC000013A)

// then shift left by 8 to match POSIX waitpid format

let ustatus = (status as u32) as u64;

Ok((pid, ustatus << 8))

}

let sig = sig as u32;

let pid = pid as u32;

if sig == Console::CTRL_C_EVENT || sig == Console::CTRL_BREAK_EVENT {

let ret = unsafe { Console::GenerateConsoleCtrlEvent(sig, pid) };

let res = if ret == 0 { Err(errno_err(vm)) } else { Ok(()) };

return res;

}

let h = unsafe { Threading::OpenProcess(Threading::PROCESS_ALL_ACCESS, 0, pid) };

if h.is_null() {

return Err(errno_err(vm));

}

let ret = unsafe { Threading::TerminateProcess(h, sig) };

let res = if ret == 0 { Err(errno_err(vm)) } else { Ok(()) };

unsafe { Foundation::CloseHandle(h) };

res

fd: OptionalArg<i32>,

vm: &VirtualMachine,

) -> PyResult<_os::TerminalSizeData> {

let (columns, lines) = {

let stdhandle = match fd {

OptionalArg::Present(0) => Console::STD_INPUT_HANDLE,

OptionalArg::Present(1) | OptionalArg::Missing => Console::STD_OUTPUT_HANDLE,

OptionalArg::Present(2) => Console::STD_ERROR_HANDLE,

_ => return Err(vm.new_value_error("bad file descriptor")),

};

let h = unsafe { Console::GetStdHandle(stdhandle) };

if h.is_null() {

return Err(vm.new_os_error("handle cannot be retrieved".to_owned()));

}

if h == INVALID_HANDLE_VALUE {

return Err(errno_err(vm));

}

let mut csbi = MaybeUninit::uninit();

let ret = unsafe { Console::GetConsoleScreenBufferInfo(h, csbi.as_mut_ptr()) };

let csbi = unsafe { csbi.assume_init() };

if ret == 0 {

return Err(errno_err(vm));

}

let w = csbi.srWindow;

(

(w.Right - w.Left + 1) as usize,

(w.Bottom - w.Top + 1) as usize,

)

};

Ok(_os::TerminalSizeData { columns, lines })

cmdname: *const u16,

argv: *const *const u16,

envp: *const *const u16,

mode: i32,

cmdname: *const u16,

argv: *const *const u16,

envp: *const *const u16,

mode: i32,

path: OsPath,

argv: Either<PyListRef, PyTupleRef>,

vm: &VirtualMachine,

) -> PyResult<intptr_t> {

use std::iter::once;

let make_widestring =

|s: &str| widestring::WideCString::from_os_str(s).map_err(|err| err.to_pyexception(vm));

let path = path.to_wide_cstring(vm)?;

let argv = vm.extract_elements_with(argv.as_ref(), |obj| {

let arg = PyStrRef::try_from_object(vm, obj)?;

make_widestring(arg.as_str())

})?;

let first = argv

.first()

.ok_or_else(|| vm.new_value_error("spawnv() arg 3 must not be empty"))?;

if first.is_empty() {

return Err(vm.new_value_error("spawnv() arg 3 first element cannot be empty"));

}

let argv_spawn: Vec<*const u16> = argv

.iter()

.map(|v| v.as_ptr())

.chain(once(std::ptr::null()))

.collect();

let result = unsafe { suppress_iph!(_wspawnv(mode, path.as_ptr(), argv_spawn.as_ptr())) };

if result == -1 {

Err(errno_err(vm))

} else {

Ok(result)

}

mode: i32,

path: OsPath,

argv: Either<PyListRef, PyTupleRef>,

env: PyDictRef,

vm: &VirtualMachine,

) -> PyResult<intptr_t> {

use std::iter::once;

let make_widestring =

|s: &str| widestring::WideCString::from_os_str(s).map_err(|err| err.to_pyexception(vm));

let path = path.to_wide_cstring(vm)?;

let argv = vm.extract_elements_with(argv.as_ref(), |obj| {

let arg = PyStrRef::try_from_object(vm, obj)?;

make_widestring(arg.as_str())

})?;

let first = argv

.first()

.ok_or_else(|| vm.new_value_error("spawnve() arg 2 cannot be empty"))?;

if first.is_empty() {

return Err(vm.new_value_error("spawnve() arg 2 first element cannot be empty"));

}

let argv_spawn: Vec<*const u16> = argv

.iter()

.map(|v| v.as_ptr())

.chain(once(std::ptr::null()))

.collect();

// Build environment strings as "KEY=VALUE\0" wide strings

let mut env_strings: Vec<widestring::WideCString> = Vec::new();

for (key, value) in env.into_iter() {

let key = PyStrRef::try_from_object(vm, key)?;

let value = PyStrRef::try_from_object(vm, value)?;

let key_str = key.as_str();

let value_str = value.as_str();

// Validate: no null characters in key or value

if key_str.contains('\0') || value_str.contains('\0') {

return Err(vm.new_value_error("embedded null character"));

}

// Validate: no '=' in key

if key_str.contains('=') {

return Err(vm.new_value_error("illegal environment variable name"));

}

let env_str = format!("{}={}", key_str, value_str);

env_strings.push(make_widestring(&env_str)?);

}

let envp: Vec<*const u16> = env_strings

.iter()

.map(|s| s.as_ptr())

.chain(once(std::ptr::null()))

.collect();

let result = unsafe {

suppress_iph!(_wspawnve(

mode,

path.as_ptr(),

argv_spawn.as_ptr(),

envp.as_ptr()

))

};

if result == -1 {

Err(errno_err(vm))

} else {

Ok(result)

}

path: OsPath,

argv: Either<PyListRef, PyTupleRef>,

vm: &VirtualMachine,

) -> PyResult<()> {

use std::iter::once;

let make_widestring =

|s: &str| widestring::WideCString::from_os_str(s).map_err(|err| err.to_pyexception(vm));

let path = path.to_wide_cstring(vm)?;

let argv = vm.extract_elements_with(argv.as_ref(), |obj| {

let arg = PyStrRef::try_from_object(vm, obj)?;

make_widestring(arg.as_str())

})?;

let first = argv

.first()

.ok_or_else(|| vm.new_value_error("execv() arg 2 must not be empty"))?;

if first.is_empty() {

return Err(vm.new_value_error("execv() arg 2 first element cannot be empty"));

}

let argv_execv: Vec<*const u16> = argv

.iter()

.map(|v| v.as_ptr())

.chain(once(std::ptr::null()))

.collect();

if (unsafe { suppress_iph!(_wexecv(path.as_ptr(), argv_execv.as_ptr())) } == -1) {

Err(errno_err(vm))

} else {

Ok(())

}

path: OsPath,

argv: Either<PyListRef, PyTupleRef>,

env: PyDictRef,

vm: &VirtualMachine,

) -> PyResult<()> {

use std::iter::once;

let make_widestring =

|s: &str| widestring::WideCString::from_os_str(s).map_err(|err| err.to_pyexception(vm));

let path = path.to_wide_cstring(vm)?;

let argv = vm.extract_elements_with(argv.as_ref(), |obj| {

let arg = PyStrRef::try_from_object(vm, obj)?;

make_widestring(arg.as_str())

})?;

let first = argv

.first()

.ok_or_else(|| vm.new_value_error("execve: argv must not be empty"))?;

if first.is_empty() {

return Err(vm.new_value_error("execve: argv first element cannot be empty"));

}

let argv_execve: Vec<*const u16> = argv

.iter()

.map(|v| v.as_ptr())

.chain(once(std::ptr::null()))

.collect();

// Build environment strings as "KEY=VALUE\0" wide strings

let mut env_strings: Vec<widestring::WideCString> = Vec::new();

for (key, value) in env.into_iter() {

let key = PyStrRef::try_from_object(vm, key)?;

let value = PyStrRef::try_from_object(vm, value)?;

let key_str = key.as_str();

let value_str = value.as_str();

// Validate: no null characters in key or value

if key_str.contains('\0') || value_str.contains('\0') {

return Err(vm.new_value_error("embedded null character"));

}

// Validate: no '=' in key

if key_str.contains('=') {

return Err(vm.new_value_error("illegal environment variable name"));

}

let env_str = format!("{}={}", key_str, value_str);

env_strings.push(make_widestring(&env_str)?);

}

let envp: Vec<*const u16> = env_strings

.iter()

.map(|s| s.as_ptr())

.chain(once(std::ptr::null()))

.collect();

if (unsafe { suppress_iph!(_wexecve(path.as_ptr(), argv_execve.as_ptr(), envp.as_ptr())) }

== -1)

{

Err(errno_err(vm))

} else {

Ok(())

}

let real = path

.as_ref()

.canonicalize()

.map_err(|e| e.to_pyexception(vm))?;

Ok(path.mode.process_path(real, vm))

let wpath = path.to_wide_cstring(vm)?;

let mut buffer = vec![0u16; Foundation::MAX_PATH as usize];

let ret = unsafe {

FileSystem::GetFullPathNameW(

wpath.as_ptr(),

buffer.len() as _,

buffer.as_mut_ptr(),

std::ptr::null_mut(),

)

};

if ret == 0 {

return Err(errno_err(vm));

}

if ret as usize > buffer.len() {

buffer.resize(ret as usize, 0);

let ret = unsafe {

FileSystem::GetFullPathNameW(

wpath.as_ptr(),

buffer.len() as _,

buffer.as_mut_ptr(),

std::ptr::null_mut(),

)

};

if ret == 0 {

return Err(errno_err(vm));

}

}

let buffer = widestring::WideCString::from_vec_truncate(buffer);

Ok(path.mode.process_path(buffer.to_os_string(), vm))

let wide = path.to_wide_cstring(vm)?;

let buflen = std::cmp::max(wide.len(), Foundation::MAX_PATH as usize);

let mut buffer = vec![0u16; buflen];

let ret = unsafe {

FileSystem::GetVolumePathNameW(wide.as_ptr(), buffer.as_mut_ptr(), buflen as _)

};

if ret == 0 {

return Err(errno_err(vm));

}

let buffer = widestring::WideCString::from_vec_truncate(buffer);

Ok(path.mode.process_path(buffer.to_os_string(), vm))

let orig: Vec<_> = path.path.to_wide();

if orig.is_empty() {

return Ok(("".to_owned(), "".to_owned()));

}

let backslashed: Vec<_> = orig

.iter()

.copied()

.map(|c| if c == b'/' as u16 { b'\\' as u16 } else { c })

.chain(std::iter::once(0)) // null-terminated

.collect();

fn from_utf16(wstr: &[u16], vm: &VirtualMachine) -> PyResult<String> {

String::from_utf16(wstr).map_err(|e| vm.new_unicode_decode_error(e.to_string()))

}

let wbuf = windows::core::PCWSTR::from_raw(backslashed.as_ptr());

let (root, path) = match unsafe { windows::Win32::UI::Shell::PathCchSkipRoot(wbuf) } {

Ok(end) => {

assert!(!end.is_null());

let len: usize = unsafe { end.as_ptr().offset_from(wbuf.as_ptr()) }

.try_into()

.expect("len must be non-negative");

assert!(

len < backslashed.len(), // backslashed is null-terminated

"path: {:?} {} < {}",

std::path::PathBuf::from(std::ffi::OsString::from_wide(&backslashed)),

len,

backslashed.len()

);

(from_utf16(&orig[..len], vm)?, from_utf16(&orig[len..], vm)?)

}

Err(_) => ("".to_owned(), from_utf16(&orig, vm)?),

};

Ok((root, path))

use FileSystem::GetDiskFreeSpaceExW;

let wpath = path.to_wide_cstring(vm)?;

let mut _free_to_me: u64 = 0;

let mut total: u64 = 0;

let mut free: u64 = 0;

let ret =

unsafe { GetDiskFreeSpaceExW(wpath.as_ptr(), &mut _free_to_me, &mut total, &mut free) };

if ret != 0 {

return Ok((total, free));

}

let err = io::Error::last_os_error();

if err.raw_os_error() == Some(Foundation::ERROR_DIRECTORY as i32)

&& let Some(parent) = path.as_ref().parent()

{

let parent = widestring::WideCString::from_os_str(parent).unwrap();

let ret = unsafe {

GetDiskFreeSpaceExW(parent.as_ptr(), &mut _free_to_me, &mut total, &mut free)

};

return if ret == 0 {

Err(errno_err(vm))

} else {

Ok((total, free))

};

}

Err(err.to_pyexception(vm))

let mut flags = 0;

if unsafe { Foundation::GetHandleInformation(handle as _, &mut flags) } == 0 {

Err(errno_err(vm))

} else {

Ok(flags & Foundation::HANDLE_FLAG_INHERIT != 0)

}

let borrowed = unsafe { crt_fd::Borrowed::borrow_raw(fd) };

let handle = crt_fd::as_handle(borrowed).map_err(|e| e.to_pyexception(vm))?;

get_handle_inheritable(handle.as_raw_handle() as _, vm)

let mut buffer = [0u16; 257];

let mut size = buffer.len() as u32;

let success = unsafe {

windows_sys::Win32::System::WindowsProgramming::GetUserNameW(

buffer.as_mut_ptr(),

&mut size,

)

};

if success != 0 {

// Convert the buffer (which is UTF-16) to a Rust String

let username = std::ffi::OsString::from_wide(&buffer[..(size - 1) as usize]);

Ok(username.to_str().unwrap().to_string())

} else {

Err(vm.new_os_error(format!("Error code: {success}")))

}

let flags = if inheritable {

Foundation::HANDLE_FLAG_INHERIT

} else {

0

};

let res = unsafe {

Foundation::SetHandleInformation(handle as _, Foundation::HANDLE_FLAG_INHERIT, flags)

};

if res == 0 {

Err(last_os_error())

} else {

Ok(())

}

use windows_sys::Win32::Foundation::ERROR_MORE_DATA;

let mut buffer = [0u16; 256];

let len =

unsafe { FileSystem::GetLogicalDriveStringsW(buffer.len() as _, buffer.as_mut_ptr()) };

if len == 0 {

return Err(errno_err(vm));

}

if len as usize >= buffer.len() {

return Err(std::io::Error::from_raw_os_error(ERROR_MORE_DATA as _).to_pyexception(vm));

}

let drives: Vec<_> = buffer[..(len - 1) as usize]

.split(|&c| c == 0)

.map(|drive| vm.new_pyobj(String::from_utf16_lossy(drive)))

.collect();

Ok(vm.ctx.new_list(drives))

handle: intptr_t,

inheritable: bool,

vm: &VirtualMachine,

) -> PyResult<()> {

raw_set_handle_inheritable(handle, inheritable).map_err(|e| e.to_pyexception(vm))

#[pyarg(any)]

path: OsPath,

#[pyarg(any, default = 0o777)]

mode: i32,

#[pyarg(flatten)]

dir_fd: DirFd<'a, { _os::MKDIR_DIR_FD as usize }>,

use windows_sys::Win32::Foundation::LocalFree;

use windows_sys::Win32::Security::Authorization::{

ConvertStringSecurityDescriptorToSecurityDescriptorW, SDDL_REVISION_1,

};

use windows_sys::Win32::Security::SECURITY_ATTRIBUTES;

let [] = args.dir_fd.0;

let wide = args.path.to_wide_cstring(vm)?;

// CPython special case: mode 0o700 sets a protected ACL

let res = if args.mode == 0o700 {

let mut sec_attr = SECURITY_ATTRIBUTES {

nLength: std::mem::size_of::<SECURITY_ATTRIBUTES>() as u32,

lpSecurityDescriptor: std::ptr::null_mut(),

bInheritHandle: 0,

};

// Set a discretionary ACL (D) that is protected (P) and includes

// inheritable (OICI) entries that allow (A) full control (FA) to

// SYSTEM (SY), Administrators (BA), and the owner (OW).

let sddl: Vec<u16> = "D:P(A;OICI;FA;;;SY)(A;OICI;FA;;;BA)(A;OICI;FA;;;OW)\0"

.encode_utf16()

.collect();

let convert_result = unsafe {

ConvertStringSecurityDescriptorToSecurityDescriptorW(

sddl.as_ptr(),

SDDL_REVISION_1,

&mut sec_attr.lpSecurityDescriptor,

std::ptr::null_mut(),

)

};

if convert_result == 0 {

return Err(errno_err(vm));

}

let res =

unsafe { FileSystem::CreateDirectoryW(wide.as_ptr(), &sec_attr as *const _ as _) };

unsafe { LocalFree(sec_attr.lpSecurityDescriptor) };

res

} else {

unsafe { FileSystem::CreateDirectoryW(wide.as_ptr(), std::ptr::null_mut()) }

};

if res == 0 {

return Err(errno_err(vm));

}

Ok(())

let result = unsafe { _umask(mask) };

if result < 0 {

Err(errno_err(vm))

} else {

Ok(result)

}

let fd2 = unsafe { suppress_iph!(_dup(fd)) };

if fd2 < 0 {

return Err(errno_err(vm));

}

let borrowed = unsafe { crt_fd::Borrowed::borrow_raw(fd2) };

let handle = crt_fd::as_handle(borrowed).map_err(|e| close_fd_and_raise(fd2, e, vm))?;

raw_set_handle_inheritable(handle.as_raw_handle() as _, false)

.map_err(|e| close_fd_and_raise(fd2, e, vm))?;

Ok(fd2)

#[pyarg(positional)]

fd: i32,

#[pyarg(positional)]

fd2: i32,

#[pyarg(any, default = true)]

inheritable: bool,

let result = unsafe { suppress_iph!(_dup2(args.fd, args.fd2)) };

if result < 0 {

return Err(errno_err(vm));

}

if !args.inheritable {

let borrowed = unsafe { crt_fd::Borrowed::borrow_raw(args.fd2) };

let handle =

crt_fd::as_handle(borrowed).map_err(|e| close_fd_and_raise(args.fd2, e, vm))?;

raw_set_handle_inheritable(handle.as_raw_handle() as _, false)

.map_err(|e| close_fd_and_raise(args.fd2, e, vm))?;

}

Ok(args.fd2)