"""Windows Job Object containment for bash() children: one kill-on-close job
per BashHandle, so the child and every descendant (breakaway is never set)
dies when the job is terminated or its last handle closes -- including on
kernel crash, since the OS closes handles. spawn_in_job() creates the child
CREATE_SUSPENDED and atomically inside its job (PROC_THREAD_ATTRIBUTE_JOB_LIST
at CreateProcessW time), so no assignment window exists. Stdlib ctypes only;
degrades to None/False where kernel32 is missing. Honest note: CI is
Ubuntu-only, so all tests mock the kernel32 boundary (`_kernel32`)."""
from __future__ import annotations
import ctypes
import ctypes.wintypes as wintypes
import os
import subprocess
import threading
from typing import BinaryIO
JOB_OBJECT_LIMIT_KILL_ON_JOB_CLOSE = 0x2000
_JobObjectBasicAccountingInformation, _JobObjectExtendedLimitInformation = 1, 9
_INT64, _SIZE_T, _DWORD = ctypes.c_int64, ctypes.c_size_t, wintypes.DWORD
_INVALID_HANDLE_VALUE = ctypes.c_void_p(-1).value
_RESUME_FAILED = 0xFFFFFFFF # (DWORD)-1 from ResumeThread
_PROC_THREAD_ATTRIBUTE_JOB_LIST = 0x2000D # value 13 | PROC_THREAD_ATTRIBUTE_INPUT
_PROC_THREAD_ATTRIBUTE_HANDLE_LIST = 0x20002 # value 2 | PROC_THREAD_ATTRIBUTE_INPUT
_EXTENDED_STARTUPINFO_PRESENT = 0x00080000
_CREATE_SUSPENDED = 0x4
_CREATE_UNICODE_ENVIRONMENT = 0x400
_STARTF_USESTDHANDLES = 0x100
_HANDLE_FLAG_INHERIT = 0x1
_GENERIC_READ = 0x80000000
_FILE_SHARE_READ_WRITE = 0x3
_OPEN_EXISTING = 3
_WAIT_OBJECT_0 = 0
_WAIT_TIMEOUT = 0x102
_INFINITE = 0xFFFFFFFF
class _JOBOBJECT_BASIC_LIMIT_INFORMATION(ctypes.Structure):
_fields_ = [
("PerProcessUserTimeLimit", _INT64), ("PerJobUserTimeLimit", _INT64),
("LimitFlags", _DWORD), ("MinimumWorkingSetSize", _SIZE_T),
("MaximumWorkingSetSize", _SIZE_T), ("ActiveProcessLimit", _DWORD),
("Affinity", _SIZE_T), ("PriorityClass", _DWORD), ("SchedulingClass", _DWORD)]
class _IO_COUNTERS(ctypes.Structure):
_fields_ = [(name, ctypes.c_uint64) for name in (
"ReadOperationCount", "WriteOperationCount", "OtherOperationCount",
"ReadTransferCount", "WriteTransferCount", "OtherTransferCount")]
class _JOBOBJECT_EXTENDED_LIMIT_INFORMATION(ctypes.Structure):
_fields_ = [
("BasicLimitInformation", _JOBOBJECT_BASIC_LIMIT_INFORMATION),
("IoInfo", _IO_COUNTERS), ("ProcessMemoryLimit", _SIZE_T),
("JobMemoryLimit", _SIZE_T), ("PeakProcessMemoryUsed", _SIZE_T),
("PeakJobMemoryUsed", _SIZE_T)]
class _JOBOBJECT_BASIC_ACCOUNTING_INFORMATION(ctypes.Structure):
_fields_ = [
("TotalUserTime", _INT64), ("TotalKernelTime", _INT64),
("ThisPeriodTotalUserTime", _INT64), ("ThisPeriodTotalKernelTime", _INT64),
("TotalPageFaultCount", _DWORD), ("TotalProcesses", _DWORD),
("ActiveProcesses", _DWORD), ("TotalTerminatedProcesses", _DWORD)]
# Fixed-width Win64 ABI types on every host: wintypes.DWORD is 8-byte c_ulong on LP64 POSIX.
_DWORD32, _WORD16, _PTR, _WSTR = ctypes.c_uint32, ctypes.c_uint16, ctypes.c_void_p, ctypes.c_wchar_p
class _STARTUPINFOW(ctypes.Structure):
_fields_ = [
("cb", _DWORD32), ("lpReserved", _WSTR), ("lpDesktop", _WSTR),
("lpTitle", _WSTR), ("dwX", _DWORD32), ("dwY", _DWORD32),
("dwXSize", _DWORD32), ("dwYSize", _DWORD32), ("dwXCountChars", _DWORD32),
("dwYCountChars", _DWORD32), ("dwFillAttribute", _DWORD32), ("dwFlags", _DWORD32),
("wShowWindow", _WORD16), ("cbReserved2", _WORD16),
("lpReserved2", _PTR), ("hStdInput", _PTR),
("hStdOutput", _PTR), ("hStdError", _PTR)]
class _STARTUPINFOEXW(ctypes.Structure):
_fields_ = [("StartupInfo", _STARTUPINFOW), ("lpAttributeList", _PTR)]
class _PROCESS_INFORMATION(ctypes.Structure):
_fields_ = [
("hProcess", _PTR), ("hThread", _PTR),
("dwProcessId", _DWORD32), ("dwThreadId", _DWORD32)]
_kernel32_cache: ctypes.WinDLL | None = None # type: ignore[name-defined]
def _kernel32():
global _kernel32_cache
if _kernel32_cache is None:
k32 = ctypes.WinDLL("kernel32", use_last_error=True)
# HANDLE argtypes/restype are mandatory: c_int truncates 64-bit handles.
h, b, i, p = wintypes.HANDLE, wintypes.BOOL, ctypes.c_int, wintypes.LPVOID
w = wintypes.LPCWSTR
for name, argtypes, restype in (
("CreateJobObjectW", [p, w], h),
("SetInformationJobObject", [h, i, p, _DWORD], b),
("QueryInformationJobObject", [h, i, p, _DWORD, p], b),
("TerminateJobObject", [h, wintypes.UINT], b),
("CloseHandle", [h], b),
("CreatePipe", [p, p, p, _DWORD], b),
("SetHandleInformation", [h, _DWORD, _DWORD], b),
("CreateFileW", [w, _DWORD, _DWORD, p, _DWORD, _DWORD, h], h),
("InitializeProcThreadAttributeList", [p, _DWORD, _DWORD, p], b),
("UpdateProcThreadAttribute", [p, _DWORD, _SIZE_T, p, _SIZE_T, p, p], b),
("DeleteProcThreadAttributeList", [p], None),
("CreateProcessW", [w, wintypes.LPWSTR, p, p, b, _DWORD, p, w, p, p], b),
("ResumeThread", [h], _DWORD), # (DWORD)-1 on failure
("WaitForSingleObject", [h, _DWORD], _DWORD),
("GetExitCodeProcess", [h, p], b),
("TerminateProcess", [h, wintypes.UINT], b),
):
fn = getattr(k32, name)
fn.argtypes, fn.restype = argtypes, restype
_kernel32_cache = k32
return _kernel32_cache
def _last_error() -> int:
return ctypes.get_last_error() if hasattr(ctypes, "get_last_error") else 0
def _open_reader(handle: int) -> BinaryIO:
"""Wrap the pipe HANDLE in a binary reader, consuming the HANDLE on every path."""
# Windows-only module, so the import cannot live at the top on POSIX.
import msvcrt
try:
fd = msvcrt.open_osfhandle(handle, os.O_RDONLY)
except BaseException:
_kernel32().CloseHandle(handle)
raise
try:
return os.fdopen(fd, "rb")
except BaseException:
os.close(fd) # the CRT fd owns the handle now; closing it closes both
raise
def create_job() -> int | None:
"""A new kill-on-close job handle, or None when jobs are unavailable."""
try:
k32 = _kernel32()
if not (job := k32.CreateJobObjectW(None, None)):
return None
info = _JOBOBJECT_EXTENDED_LIMIT_INFORMATION()
info.BasicLimitInformation.LimitFlags = JOB_OBJECT_LIMIT_KILL_ON_JOB_CLOSE
if not k32.SetInformationJobObject(
job, _JobObjectExtendedLimitInformation, ctypes.byref(info), ctypes.sizeof(info)):
k32.CloseHandle(job)
return None
return job
except (OSError, AttributeError):
return None
class JobProcess:
"""Minimal Popen-like surface for a child spawn_in_job() created suspended."""
def __init__(self, hprocess: int, hthread: int, pid: int, stdout: BinaryIO) -> None:
self.pid = pid
self.stdout: BinaryIO | None = stdout
# Guards every (_exit_code, _hprocess, _hthread, _waiters) transition; never held blocking.
self._lock = threading.Lock()
self._hprocess: int | None = hprocess
self._hthread: int | None = hthread
self._exit_code: int | None = None
self._waiters = 0 # wait() calls with a blocking WFSO in flight
self._close_requested = False
def resume(self) -> bool:
"""Release the suspended primary thread; the handle stays open on failure."""
with self._lock:
if self._hthread is None or _kernel32().ResumeThread(self._hthread) == _RESUME_FAILED:
return False
_kernel32().CloseHandle(self._hthread)
self._hthread = None
return True
def close(self) -> None:
"""Release the process handle (idempotent); owned by BashHandle after reap."""
with self._lock:
self._close_requested = True
self._close_hprocess_locked()
def _cache_exit_code_locked(self) -> int:
# Runs under the lock after a signaled wait, so never the STILL_ACTIVE(259) sentinel.
k32 = _kernel32()
code = _DWORD()
if not k32.GetExitCodeProcess(self._hprocess, ctypes.byref(code)):
# Handles stay open (nothing cached) so a retry is possible.
raise OSError(f"GetExitCodeProcess failed: {_last_error()}")
self._exit_code = int(code.value)
if self._hthread is not None:
k32.CloseHandle(self._hthread)
self._hthread = None
return self._exit_code
def _close_hprocess_locked(self) -> None:
# Closing a handle mid-WaitForSingleObject is UB: the last returning waiter closes it.
if self._close_requested and self._waiters == 0 and self._hprocess is not None:
_kernel32().CloseHandle(self._hprocess)
self._hprocess = None
def poll(self) -> int | None:
# WaitForSingleObject(0) never blocks, so it may run under the lock.
with self._lock:
if self._exit_code is not None:
return self._exit_code
if self._hprocess is None:
raise OSError("process handle closed")
result = _kernel32().WaitForSingleObject(self._hprocess, 0)
if result == _WAIT_TIMEOUT:
return None
if result != _WAIT_OBJECT_0:
raise OSError(f"WaitForSingleObject failed: {_last_error()}")
return self._cache_exit_code_locked()
def wait(self, timeout: float | None = None) -> int:
# Convert before registering: int(nan/inf) raises and would leak _waiters forever.
millis = _INFINITE if timeout is None else max(0, int(timeout * 1000))
with self._lock:
if self._exit_code is not None:
return self._exit_code
if self._hprocess is None:
raise OSError("process handle closed")
hprocess = self._hprocess
self._waiters += 1 # keeps the handle open while the wait runs
result = None
try:
result = _kernel32().WaitForSingleObject(hprocess, millis)
finally:
with self._lock:
try:
# Cache while the handle is still valid: the deferred close
# fires the moment _waiters drops to zero.
if result == _WAIT_OBJECT_0 and self._exit_code is None:
self._cache_exit_code_locked()
finally:
self._waiters -= 1
self._close_hprocess_locked() # last waiter after a close() request closes
with self._lock:
# A code cached concurrently during the wait wins, even over WAIT_FAILED.
if self._exit_code is not None:
return self._exit_code
if result == _WAIT_TIMEOUT:
raise subprocess.TimeoutExpired(f"pid {self.pid}", timeout or 0)
raise OSError(f"WaitForSingleObject failed: {_last_error()}")
def kill(self) -> None:
with self._lock:
if self._exit_code is not None or self._hprocess is None:
return
ok = bool(_kernel32().TerminateProcess(self._hprocess, 1))
if not ok and self.poll() is None:
raise OSError(f"TerminateProcess failed: {_last_error()}")
def spawn_in_job(job: int, argv: list[str], cwd: str, env: dict[str, str]) -> JobProcess:
"""Create argv suspended and atomically inside the caller-owned job (JOB_LIST
attribute, no assignment window); raises OSError on any failure (nothing spawned)."""
try:
k32 = _kernel32()
except (OSError, AttributeError) as exc:
raise OSError("kernel32 unavailable") from exc
read_handle = write_handle = nul_handle = None
attr_list = None
proc = None
try:
read_out, write_out = wintypes.HANDLE(), wintypes.HANDLE()
if not k32.CreatePipe(ctypes.byref(read_out), ctypes.byref(write_out), None, 0):
raise OSError(f"CreatePipe failed: {_last_error()}")
read_handle, write_handle = int(read_out.value or 0), int(write_out.value or 0)
if not k32.SetHandleInformation(write_handle, _HANDLE_FLAG_INHERIT, _HANDLE_FLAG_INHERIT):
raise OSError(f"SetHandleInformation failed: {_last_error()}")
nul = k32.CreateFileW(
"NUL", _GENERIC_READ, _FILE_SHARE_READ_WRITE, None, _OPEN_EXISTING, 0, None)
if not nul or nul == _INVALID_HANDLE_VALUE:
raise OSError(f"CreateFileW(NUL) failed: {_last_error()}")
nul_handle = int(nul)
if not k32.SetHandleInformation(nul_handle, _HANDLE_FLAG_INHERIT, _HANDLE_FLAG_INHERIT):
raise OSError(f"SetHandleInformation failed: {_last_error()}")
# Two-call protocol: the sizing call fails with ERROR_INSUFFICIENT_BUFFER.
size = ctypes.c_size_t(0)
k32.InitializeProcThreadAttributeList(None, 2, 0, ctypes.byref(size))
if not size.value:
raise OSError(f"InitializeProcThreadAttributeList sizing failed: {_last_error()}")
buffer = ctypes.create_string_buffer(size.value)
if not k32.InitializeProcThreadAttributeList(buffer, 2, 0, ctypes.byref(size)):
raise OSError(f"InitializeProcThreadAttributeList failed: {_last_error()}")
attr_list = buffer
jobs = (wintypes.HANDLE * 1)(job)
if not k32.UpdateProcThreadAttribute(
attr_list, 0, _PROC_THREAD_ATTRIBUTE_JOB_LIST, ctypes.byref(jobs),
ctypes.sizeof(jobs), None, None):
raise OSError(f"UpdateProcThreadAttribute failed: {_last_error()}")
# HANDLE_LIST blocks concurrent spawns' handle leaks; both arrays outlive CreateProcessW.
inheritable = (wintypes.HANDLE * 2)(write_handle, nul_handle)
if not k32.UpdateProcThreadAttribute(
attr_list, 0, _PROC_THREAD_ATTRIBUTE_HANDLE_LIST, ctypes.byref(inheritable),
ctypes.sizeof(inheritable), None, None):
raise OSError(f"UpdateProcThreadAttribute failed: {_last_error()}")
startup = _STARTUPINFOEXW()
startup.StartupInfo.cb = ctypes.sizeof(_STARTUPINFOEXW)
startup.StartupInfo.dwFlags = _STARTF_USESTDHANDLES
startup.StartupInfo.hStdInput = nul_handle
startup.StartupInfo.hStdOutput = startup.StartupInfo.hStdError = write_handle
startup.lpAttributeList = ctypes.cast(attr_list, wintypes.LPVOID)
# CreateProcessW may rewrite lpCommandLine in place: a writable buffer is mandatory.
cmdline = ctypes.create_unicode_buffer(subprocess.list2cmdline(argv))
# Sorted case-insensitively per CreateProcessW; the implicit terminator is the second NUL.
env_block = ctypes.create_unicode_buffer(
"\0".join(
f"{key}={value}" for key, value in sorted(env.items(), key=lambda kv: kv[0].upper())
) + "\0")
info = _PROCESS_INFORMATION()
flags = _EXTENDED_STARTUPINFO_PRESENT | _CREATE_SUSPENDED | _CREATE_UNICODE_ENVIRONMENT
if not k32.CreateProcessW(
None, cmdline, None, None, True, flags, env_block, cwd,
ctypes.byref(startup), ctypes.byref(info)):
raise OSError(f"CreateProcessW failed: {_last_error()}")
try:
# _open_reader consumes the read handle on every path: drop it at the call.
reader_handle, read_handle = read_handle, None
proc = JobProcess(
int(info.hProcess or 0), int(info.hThread or 0), int(info.dwProcessId),
_open_reader(reader_handle))
except BaseException:
# No JobProcess owns these yet; the caller's kill-on-close job reaps the child.
for handle in (info.hProcess, info.hThread):
if handle:
k32.CloseHandle(handle)
raise
finally:
if attr_list is not None:
k32.DeleteProcThreadAttributeList(attr_list)
for handle in (write_handle, nul_handle):
if handle is not None:
k32.CloseHandle(handle)
if read_handle is not None: # only when _open_reader was never invoked
k32.CloseHandle(read_handle)
return proc
def is_empty(job: int) -> bool | None:
"""True/False = job has no/some live processes; None = query failed."""
try:
info = _JOBOBJECT_BASIC_ACCOUNTING_INFORMATION()
ok = _kernel32().QueryInformationJobObject(
job, _JobObjectBasicAccountingInformation, ctypes.byref(info), ctypes.sizeof(info), None
)
return info.ActiveProcesses == 0 if ok else None
except (OSError, AttributeError):
return None
def terminate(job: int, exit_code: int = 1) -> bool:
try:
return bool(_kernel32().TerminateJobObject(job, exit_code))
except (OSError, AttributeError):
return False
def close(job: int) -> None:
"""Close the handle; closing the LAST handle fires kill-on-close."""
try:
_kernel32().CloseHandle(job)
except (OSError, AttributeError):
pass