From d4d5a198733f0090bdbc7a4abe6c805bf1a61813 Mon Sep 17 00:00:00 2001 From: Ryan Fowler Date: Sat, 18 Jul 2026 06:53:26 -0400 Subject: [PATCH 1/2] fix: bound external image adapter execution Apply one deadline to process execution and stdout collection, abort immediately when output exceeds the cap, and contain Windows adapters in a kill-on-close job before resuming them. --- Cargo.toml | 2 +- src/image/external.rs | 408 +++++++++++++++++++++++++++++++++++------- 2 files changed, 345 insertions(+), 65 deletions(-) diff --git a/Cargo.toml b/Cargo.toml index 099b5550..e8a8ae2c 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -78,7 +78,7 @@ zstd = "=0.13.3" libc = "=0.2.186" [target.'cfg(windows)'.dependencies] -windows-sys = { version = "=0.61.2", features = ["Win32_Foundation", "Win32_NetworkManagement_Dns", "Win32_Security", "Win32_Storage_FileSystem", "Win32_System_Console", "Win32_System_IO", "Win32_System_Pipes", "Win32_System_Threading"] } +windows-sys = { version = "=0.61.2", features = ["Win32_Foundation", "Win32_NetworkManagement_Dns", "Win32_Security", "Win32_Storage_FileSystem", "Win32_System_Console", "Win32_System_Diagnostics_ToolHelp", "Win32_System_IO", "Win32_System_JobObjects", "Win32_System_Pipes", "Win32_System_Threading"] } [dev-dependencies] assert_cmd = "=2.2.2" diff --git a/src/image/external.rs b/src/image/external.rs index 3f6f1eae..fe9b83ed 100644 --- a/src/image/external.rs +++ b/src/image/external.rs @@ -1,7 +1,8 @@ use std::env; use std::io::{ErrorKind, Read}; use std::path::{Path, PathBuf}; -use std::process::{Child, Command, ExitStatus, Stdio}; +use std::process::{Child, ChildStdout, Command, ExitStatus, Stdio}; +use std::sync::mpsc; use std::thread; use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH}; @@ -93,12 +94,6 @@ fn decode_adaptor(path: &Path, adaptor: Adaptor) -> Result Result, - stdout_truncated: bool, +} + +#[derive(Debug)] +enum StdoutResult { + Complete(std::io::Result>), + CapExceeded, + TimedOut, } fn run_adaptor(cmd: Command, name: &str) -> Result { @@ -129,38 +130,92 @@ fn run_adaptor_with_timeout( .stdout(Stdio::piped()) .stderr(Stdio::null()); let mut child = cmd.spawn()?; + let process_group = AdaptorProcessGroup::attach(&mut child)?; let stdout = child .stdout .take() .ok_or_else(|| ImageError::Message(format!("{name} stdout unavailable")))?; - let stdout_reader = thread::spawn(move || read_capped(stdout, ADAPTOR_STDOUT_CAP)); let deadline = Instant::now() + timeout; - - let status = loop { - if let Some(status) = child.try_wait()? { - break status; + let (stdout_tx, stdout_rx) = mpsc::sync_channel(1); + let stdout_reader = thread::spawn(move || { + let _ = stdout_tx.send(read_capped_until(stdout, ADAPTOR_STDOUT_CAP, deadline)); + }); + + let mut status = None; + let stdout_result = loop { + if status.is_none() { + status = child.try_wait()?; } - if Instant::now() >= deadline { - terminate_adaptor(&mut child); - if stdout_reader.is_finished() { + match stdout_rx.try_recv() { + Ok(result) => break result, + Err(mpsc::TryRecvError::Disconnected) => { let _ = stdout_reader.join(); + terminate_adaptor(&mut child, &process_group); + return Err(ImageError::Message(format!( + "{name} stdout reader panicked" + ))); } - return Err(ImageError::Message(format!( - "{name} timed out after {}", - format_go_duration(timeout) - ))); + Err(mpsc::TryRecvError::Empty) => {} + } + + let now = Instant::now(); + if now >= deadline { + break StdoutResult::TimedOut; } - thread::sleep(Duration::from_millis(10)); + thread::sleep((deadline - now).min(Duration::from_millis(10))); }; - let (stdout, stdout_truncated) = stdout_reader - .join() - .map_err(|_| ImageError::Message(format!("{name} stdout reader panicked")))??; - Ok(AdaptorOutput { - status, - stdout, - stdout_truncated, - }) + match stdout_result { + StdoutResult::Complete(stdout) => { + let stdout = match stdout { + Ok(stdout) => stdout, + Err(err) => { + terminate_adaptor(&mut child, &process_group); + stdout_reader.join().map_err(|_| { + ImageError::Message(format!("{name} stdout reader panicked")) + })?; + return Err(err.into()); + } + }; + while status.is_none() { + let now = Instant::now(); + if now >= deadline { + terminate_adaptor(&mut child, &process_group); + stdout_reader.join().map_err(|_| { + ImageError::Message(format!("{name} stdout reader panicked")) + })?; + return Err(adaptor_timeout_error(name, timeout)); + } + status = child.try_wait()?; + if status.is_none() { + thread::sleep((deadline - now).min(Duration::from_millis(10))); + } + } + stdout_reader + .join() + .map_err(|_| ImageError::Message(format!("{name} stdout reader panicked")))?; + Ok(AdaptorOutput { + status: status.expect("status was checked above"), + stdout, + }) + } + StdoutResult::CapExceeded => { + terminate_adaptor(&mut child, &process_group); + stdout_reader + .join() + .map_err(|_| ImageError::Message(format!("{name} stdout reader panicked")))?; + Err(ImageError::Message(format!( + "{name} produced more than {ADAPTOR_STDOUT_CAP} bytes" + ))) + } + StdoutResult::TimedOut => { + terminate_adaptor(&mut child, &process_group); + stdout_reader + .join() + .map_err(|_| ImageError::Message(format!("{name} stdout reader panicked")))?; + Err(adaptor_timeout_error(name, timeout)) + } + } } #[cfg(unix)] @@ -170,47 +225,249 @@ fn prepare_adaptor_command(cmd: &mut Command) { cmd.process_group(0); } -#[cfg(not(unix))] -fn prepare_adaptor_command(_cmd: &mut Command) {} +#[cfg(windows)] +fn prepare_adaptor_command(cmd: &mut Command) { + use std::os::windows::process::CommandExt; + use windows_sys::Win32::System::Threading::CREATE_SUSPENDED; + + // Assigning the process to its job before its primary thread runs prevents + // short-lived adapters and early helpers from escaping the job. + cmd.creation_flags(CREATE_SUSPENDED); +} -fn terminate_adaptor(child: &mut Child) { - kill_adaptor_process_group(child); +fn adaptor_timeout_error(name: &str, timeout: Duration) -> ImageError { + ImageError::Message(format!( + "{name} timed out after {}", + format_go_duration(timeout) + )) +} + +fn terminate_adaptor(child: &mut Child, process_group: &AdaptorProcessGroup) { + process_group.terminate(child); let _ = child.kill(); let _ = child.wait(); } #[cfg(unix)] -fn kill_adaptor_process_group(child: &Child) { - if let Ok(pid) = i32::try_from(child.id()) { - // The adapter is spawned as a process-group leader, so this also catches helpers it starts. +struct AdaptorProcessGroup; + +#[cfg(unix)] +impl AdaptorProcessGroup { + fn attach(_child: &mut Child) -> std::io::Result { + Ok(Self) + } + + fn terminate(&self, child: &Child) { + if let Ok(pid) = i32::try_from(child.id()) { + // The adapter is spawned as a process-group leader, so this also catches helpers it starts. + unsafe { + libc::kill(-pid, libc::SIGKILL); + } + } + } +} + +#[cfg(windows)] +struct AdaptorProcessGroup { + job: windows_sys::Win32::Foundation::HANDLE, +} + +#[cfg(windows)] +impl AdaptorProcessGroup { + fn attach(child: &mut Child) -> std::io::Result { + use std::os::windows::io::AsRawHandle; + use windows_sys::Win32::System::JobObjects::{ + AssignProcessToJobObject, CreateJobObjectW, JOB_OBJECT_LIMIT_KILL_ON_JOB_CLOSE, + JOBOBJECT_EXTENDED_LIMIT_INFORMATION, JobObjectExtendedLimitInformation, + SetInformationJobObject, + }; + + // SAFETY: all pointers are null or point to initialized values for the duration of each call. unsafe { - libc::kill(-pid, libc::SIGKILL); + let job = CreateJobObjectW(std::ptr::null(), std::ptr::null()); + if job.is_null() { + return Err(std::io::Error::last_os_error()); + } + let mut info = JOBOBJECT_EXTENDED_LIMIT_INFORMATION::default(); + info.BasicLimitInformation.LimitFlags = JOB_OBJECT_LIMIT_KILL_ON_JOB_CLOSE; + if SetInformationJobObject( + job, + JobObjectExtendedLimitInformation, + (&raw const info).cast(), + std::mem::size_of_val(&info) as u32, + ) == 0 + || AssignProcessToJobObject(job, child.as_raw_handle()) == 0 + { + let err = std::io::Error::last_os_error(); + windows_sys::Win32::Foundation::CloseHandle(job); + let _ = child.kill(); + let _ = child.wait(); + return Err(err); + } + if let Err(err) = resume_primary_thread(child.id()) { + windows_sys::Win32::Foundation::CloseHandle(job); + let _ = child.kill(); + let _ = child.wait(); + return Err(err); + } + Ok(Self { job }) + } + } + + fn terminate(&self, _child: &Child) { + use windows_sys::Win32::System::JobObjects::TerminateJobObject; + // SAFETY: self.job remains valid until Drop. + unsafe { + TerminateJobObject(self.job, 1); } } } -#[cfg(not(unix))] -fn kill_adaptor_process_group(_child: &Child) {} +#[cfg(windows)] +fn resume_primary_thread(process_id: u32) -> std::io::Result<()> { + use windows_sys::Win32::Foundation::{CloseHandle, INVALID_HANDLE_VALUE}; + use windows_sys::Win32::System::Diagnostics::ToolHelp::{ + CreateToolhelp32Snapshot, TH32CS_SNAPTHREAD, THREADENTRY32, Thread32First, Thread32Next, + }; + use windows_sys::Win32::System::Threading::{OpenThread, ResumeThread, THREAD_SUSPEND_RESUME}; + + // CreateProcess has completed, but CREATE_SUSPENDED guarantees that the + // process still has only its primary thread and has not executed user code. + unsafe { + let snapshot = CreateToolhelp32Snapshot(TH32CS_SNAPTHREAD, 0); + if snapshot == INVALID_HANDLE_VALUE { + return Err(std::io::Error::last_os_error()); + } + + let mut entry = THREADENTRY32::default(); + entry.dwSize = std::mem::size_of::() as u32; + let mut found = Thread32First(snapshot, &mut entry) != 0; + while found { + if entry.th32OwnerProcessID == process_id { + let thread = OpenThread(THREAD_SUSPEND_RESUME, 0, entry.th32ThreadID); + if thread.is_null() { + let err = std::io::Error::last_os_error(); + CloseHandle(snapshot); + return Err(err); + } + let resumed = ResumeThread(thread); + let err = if resumed == u32::MAX { + Some(std::io::Error::last_os_error()) + } else { + None + }; + CloseHandle(thread); + CloseHandle(snapshot); + return err.map_or(Ok(()), Err); + } + found = Thread32Next(snapshot, &mut entry) != 0; + } -fn read_capped(mut reader: R, cap: usize) -> std::io::Result<(Vec, bool)> { + CloseHandle(snapshot); + Err(std::io::Error::new( + ErrorKind::NotFound, + "unable to find suspended adapter thread", + )) + } +} + +#[cfg(windows)] +impl Drop for AdaptorProcessGroup { + fn drop(&mut self) { + // Closing a kill-on-close job also removes any helpers left after the adapter exits. + unsafe { + windows_sys::Win32::Foundation::CloseHandle(self.job); + } + } +} + +#[cfg(unix)] +fn read_capped_until(mut reader: ChildStdout, cap: usize, deadline: Instant) -> StdoutResult { + use std::os::fd::AsRawFd; + + let fd = reader.as_raw_fd(); + // SAFETY: fd is owned by reader and stays valid for the duration of this function. + let flags = unsafe { libc::fcntl(fd, libc::F_GETFL) }; + if flags < 0 || unsafe { libc::fcntl(fd, libc::F_SETFL, flags | libc::O_NONBLOCK) } < 0 { + return StdoutResult::Complete(Err(std::io::Error::last_os_error())); + } + read_capped_polling( + |buf| reader.read(buf), + cap, + deadline, + |err| err.kind() == ErrorKind::WouldBlock, + ) +} + +#[cfg(windows)] +fn read_capped_until(mut reader: ChildStdout, cap: usize, deadline: Instant) -> StdoutResult { + use std::os::windows::io::AsRawHandle; + use windows_sys::Win32::Foundation::{ERROR_BROKEN_PIPE, ERROR_HANDLE_EOF, GetLastError}; + use windows_sys::Win32::System::Pipes::PeekNamedPipe; + + let handle = reader.as_raw_handle(); + read_capped_polling( + |buf| { + let mut available = 0_u32; + // SAFETY: handle is the live child stdout pipe and output pointers are valid. + let ok = unsafe { + PeekNamedPipe( + handle, + std::ptr::null_mut(), + 0, + std::ptr::null_mut(), + &mut available, + std::ptr::null_mut(), + ) + }; + if ok == 0 { + // SAFETY: reads the thread-local error from PeekNamedPipe. + return match unsafe { GetLastError() } { + ERROR_BROKEN_PIPE | ERROR_HANDLE_EOF => Ok(0), + _ => Err(std::io::Error::last_os_error()), + }; + } + if available == 0 { + return Err(std::io::Error::from(ErrorKind::WouldBlock)); + } + reader.read(&mut buf[..buf.len().min(available as usize)]) + }, + cap, + deadline, + |err| err.kind() == ErrorKind::WouldBlock, + ) +} + +fn read_capped_polling( + mut read: F, + cap: usize, + deadline: Instant, + would_block: W, +) -> StdoutResult +where + F: FnMut(&mut [u8]) -> std::io::Result, + W: Fn(&std::io::Error) -> bool, +{ let mut out = Vec::new(); - let mut truncated = false; let mut buf = [0; 8192]; loop { - let n = reader.read(&mut buf)?; - if n == 0 { - break; - } - let remaining = cap.saturating_sub(out.len()); - if remaining > 0 { - let keep = remaining.min(n); - out.extend_from_slice(&buf[..keep]); + if Instant::now() >= deadline { + return StdoutResult::TimedOut; } - if n > remaining { - truncated = true; + match read(&mut buf) { + Ok(0) => return StdoutResult::Complete(Ok(out)), + Ok(n) => { + let remaining = cap.saturating_sub(out.len()); + if n > remaining { + return StdoutResult::CapExceeded; + } + out.extend_from_slice(&buf[..n]); + } + Err(err) if would_block(&err) => thread::sleep(Duration::from_millis(5)), + Err(err) if err.kind() == ErrorKind::Interrupted => {} + Err(err) => return StdoutResult::Complete(Err(err)), } } - Ok((out, truncated)) } struct TempImageDir { @@ -284,8 +541,6 @@ fn write_temp_image_file(path: &Path, bytes: &[u8]) -> std::io::Result<()> { #[cfg(test)] mod tests { - use std::io::Cursor; - use super::*; #[cfg(unix)] @@ -325,23 +580,48 @@ mod tests { } #[test] - fn capped_reader_preserves_limit_and_drains_input() { - let (out, truncated) = read_capped(Cursor::new(b"abcdef"), 4).unwrap(); - assert_eq!(out, b"abcd"); - assert!(truncated); - - let (out, truncated) = read_capped(Cursor::new(b"abc"), 4).unwrap(); - assert_eq!(out, b"abc"); - assert!(!truncated); + fn capped_reader_stops_immediately() { + let mut first_read = true; + let result = read_capped_polling( + |buf| { + assert!(first_read, "reader was drained after exceeding the cap"); + first_read = false; + buf[..6].copy_from_slice(b"abcdef"); + Ok(6) + }, + 4, + Instant::now() + Duration::from_secs(1), + |_| false, + ); + assert!(matches!(result, StdoutResult::CapExceeded)); } #[cfg(unix)] #[test] fn adaptor_timeout_does_not_wait_for_inherited_stdout() { let mut command = Command::new("/bin/sh"); - command.args(["-c", "sleep 5 >&1 & sleep 5"]); - let started_at = Instant::now(); + // The shell exits immediately while its descendant keeps stdout open. + command.args(["-c", "sleep 5 >&1 &"]); + assert_adapter_times_out_promptly(command); + } + #[cfg(windows)] + #[test] + fn adaptor_job_catches_immediate_stdout_inheriting_child() { + let mut command = Command::new("cmd"); + // `start /B` launches ping asynchronously with the shell's stdout handle. + command.args([ + "/D", + "/S", + "/C", + "start \"\" /B ping -n 6 127.0.0.1 & exit /B 0", + ]); + assert_adapter_times_out_promptly(command); + } + + #[cfg(any(unix, windows))] + fn assert_adapter_times_out_promptly(command: Command) { + let started_at = Instant::now(); let err = run_adaptor_with_timeout(command, "fake-adaptor", Duration::from_millis(100)) .unwrap_err() .to_string(); From 3fbde71d6ab3f9cefd46f3d058c06ddcea930c93 Mon Sep 17 00:00:00 2001 From: Ryan Fowler Date: Sat, 18 Jul 2026 07:30:46 -0400 Subject: [PATCH 2/2] fix: avoid overlapping Windows stdout borrows --- src/image/external.rs | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/src/image/external.rs b/src/image/external.rs index fe9b83ed..cd386634 100644 --- a/src/image/external.rs +++ b/src/image/external.rs @@ -430,7 +430,8 @@ fn read_capped_until(mut reader: ChildStdout, cap: usize, deadline: Instant) -> if available == 0 { return Err(std::io::Error::from(ErrorKind::WouldBlock)); } - reader.read(&mut buf[..buf.len().min(available as usize)]) + let read_len = buf.len().min(available as usize); + reader.read(&mut buf[..read_len]) }, cap, deadline,