diff --git a/.github/workflows/test.yml b/.github/workflows/test.yml index 2418e7a..0f24cf3 100644 --- a/.github/workflows/test.yml +++ b/.github/workflows/test.yml @@ -24,6 +24,7 @@ jobs: features: - tokio1 - std + - pty name: "Test on ${{ matrix.platform }} with Rust ${{ matrix.toolchain }} (feat: ${{ matrix.features }})" runs-on: "${{ matrix.platform }}-latest" @@ -93,6 +94,54 @@ jobs: - run: tests/multiproc_helper.rs 1 1 - run: cargo test --locked --all-features --all-targets + portable-unix-pty: + strategy: + fail-fast: false + matrix: + include: + - target: aarch64-apple-ios + toolchain: stable + - target: aarch64-linux-android + toolchain: stable + - target: x86_64-unknown-freebsd + toolchain: stable + - target: x86_64-unknown-netbsd + toolchain: stable + - target: x86_64-unknown-illumos + toolchain: stable + - target: x86_64-pc-solaris + toolchain: stable + - target: x86_64-unknown-dragonfly + toolchain: nightly + - target: x86_64-unknown-openbsd + toolchain: nightly + + name: "Check PTY on ${{ matrix.target }}" + runs-on: ubuntu-latest + + steps: + - uses: actions/checkout@11bd71901bbe5b1630ceea73d27597364c9af683 # v4.2.2 + with: + persist-credentials: false + + - name: Configure Rust + run: | + rustup toolchain install --profile minimal --no-self-update ${{ matrix.toolchain }} + rustup default ${{ matrix.toolchain }} + if [ "${{ matrix.toolchain }}" = nightly ]; then + rustup component add rust-src + else + rustup target add "${{ matrix.target }}" + fi + + - name: Check portable Unix PTY backend + run: | + if [ "${{ matrix.toolchain }}" = nightly ]; then + cargo check --locked -Zbuild-std=std --target "${{ matrix.target }}" --no-default-features --features pty --lib + else + cargo check --locked --target "${{ matrix.target }}" --no-default-features --features pty --lib + fi + rustdoc: name: Rustdoc runs-on: ubuntu-latest @@ -112,6 +161,28 @@ jobs: RUSTDOCFLAGS: -D warnings run: cargo doc --locked --all-features --no-deps + - name: Test documentation examples + run: cargo test --locked --all-features --doc + + package: + name: Package contents + runs-on: ubuntu-latest + + steps: + - uses: actions/checkout@11bd71901bbe5b1630ceea73d27597364c9af683 # v4.2.2 + with: + persist-credentials: false + + - name: Configure Rust + run: | + rustup toolchain install --profile minimal --no-self-update stable + rustup default stable + + - name: Verify and list the release package + run: | + cargo package --locked + cargo package --locked --list + semver: name: Semver against PR base if: github.event_name == 'pull_request' diff --git a/Cargo.lock b/Cargo.lock index a721e93..1b04b8e 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -724,6 +724,16 @@ version = "1.15.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "67b1b7a3b5fe4f1376887184045fcf45c69e92af734b7aaddc05fb777b6fbd03" +[[package]] +name = "socket2" +version = "0.6.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c3d1e2c7f27f8d4cb10542a02c49005dbd6e93095799d6f3be745fae9f8fedd4" +dependencies = [ + "libc", + "windows-sys", +] + [[package]] name = "stable_deref_trait" version = "1.2.1" @@ -764,6 +774,7 @@ dependencies = [ "mio", "pin-project-lite", "signal-hook-registry", + "socket2", "tokio-macros", "windows-sys", ] diff --git a/Cargo.toml b/Cargo.toml index f16192d..294e69c 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -52,6 +52,9 @@ std = ["dep:nix"] ## Frontend: tokio::Command tokio1 = ["dep:nix", "dep:futures", "dep:tokio"] +## Tokio pseudo-terminal transport +pty = ["tokio1", "nix/term", "tokio/net"] + ## Wrapper: Creation Flags creation-flags = ["dep:windows", "windows/Win32_System_Threading"] @@ -73,8 +76,13 @@ reset-sigmask = [] [package.metadata.docs.rs] all-features = true targets = [ + "aarch64-linux-android", "x86_64-unknown-linux-gnu", "x86_64-apple-darwin", + "x86_64-unknown-freebsd", + "x86_64-unknown-netbsd", + "x86_64-unknown-illumos", + "x86_64-pc-solaris", "x86_64-pc-windows-msvc", ] diff --git a/README.md b/README.md index 7feded5..d41cb09 100644 --- a/README.md +++ b/README.md @@ -86,6 +86,60 @@ let status = child.wait().await?; dbg!(status); ``` +### or in a pseudo-terminal + +The non-default `pty` feature enables Tokio PTY transport on Linux, Android, macOS, FreeBSD, +NetBSD 10 and newer, OpenBSD, DragonFly BSD, illumos, and Solaris. It implies `tokio1`, selecting +the Tokio frontend and its terminal dependencies explicitly. + +```toml +[dependencies] +process-wrap = { version = "10.0.0", features = ["pty"] } +``` + +```rust +use process_wrap::tokio::*; +use tokio::io::AsyncReadExt; + +let mut command = Command::new("ls"); +command.wrap(ProcessSession).wrap(Pty::default()); +let mut child = command.spawn()?; +let controller = child + .take_pty_controller() + .expect("a successful PTY spawn installs one controller"); +let (input, mut output, _resize) = controller.into_parts(); +drop(input); + +let drain = tokio::spawn(async move { + let mut bytes = Vec::new(); + output.read_to_end(&mut bytes).await?; + Ok::<_, std::io::Error>(bytes) +}); +let status = child.wait().await?; +let terminal_bytes = drain.await??; +dbg!(status, terminal_bytes); +``` + +A PTY has one ordered terminal stream, so standard output and standard error are merged. +`PtyInput` and `PtyOutput` are strong owners of one bidirectional master descriptor, so dropping +either one alone does not half-close the terminal. The terminal hangs up after both are gone; +`PtyResize` is weak and cannot keep it alive. Send the terminal's VEOF character when that is the +desired terminal policy instead of expecting a separate input half-close or clonable force-close +handle. + +Child waiting and PTY draining are independent. On most supported Unix systems, descendants can +retain the slave after the direct child exits. On macOS, drain output concurrently with waiting: the +kernel drains queued output as the session leader exits, then revokes the controlling terminal from +its descendants. The transport passes terminal bytes through without owning parent-terminal raw +mode, relays, key handling, VT parsing, scrollback, or pager policy. + +A bare PTY creates the required session. `ProcessGroup::leader()` and `ProcessSession` each preserve +group-wide signalling while the direct child is live; waiting still follows that direct child. +`ProcessGroup::attach_to(...)` and explicitly registering both wrappers return `InvalidInput`. +`ResetSigmask` composes normally. `KillOnDrop` remains Tokio's direct-child behavior—it does not +promise to kill an entire group or session. Spawning returns the ordinary boxed Tokio child, and +`take_pty_controller()` traverses any outer child wrappers and yields the controller once. + ### or with std ```toml @@ -342,6 +396,8 @@ Both can exist at the same time, but generally you should use one or the other. - `kill-on-drop`: **default**, enables the [kill on drop](#kill-on-drop) wrapper. - `process-group`: **default**, enables the [process group](#process-group) wrapper. - `process-session`: **default**, enables the [process session](#process-session) wrapper. +- `pty`: enables the Tokio [pseudo-terminal transport](#or-in-a-pseudo-terminal) and implies + `tokio1`. - `reset-sigmask`: enables the [reset signal mask](#reset-signal-mask) wrapper. ### Diagnostics diff --git a/src/command.rs b/src/command.rs index f5fe685..b42fe25 100644 --- a/src/command.rs +++ b/src/command.rs @@ -1460,6 +1460,40 @@ impl SpawnAttempt { self.native_command_mut() } + /// Take the fresh native command for an alternate provider without applying native platform setup + /// or making the portable attempt opaque. + /// + /// Removing the provider's command leaves the tracked intent intact. If a later lifecycle hook + /// requests native access, the attempt materializes a separate clean command which contains none of + /// the provider's private stdio or child-setup state. + /// + /// The provider must apply every accepted portable policy itself before spawning. + #[cfg(all( + feature = "pty", + any( + target_os = "android", + target_os = "dragonfly", + target_os = "freebsd", + target_os = "illumos", + target_os = "linux", + target_os = "macos", + target_os = "netbsd", + target_os = "openbsd", + target_os = "solaris" + ) + ))] + pub(crate) fn take_native_for_provider_spawn(&mut self) -> B::NativeCommand { + self.materialize_native(); + match &mut self.state { + AttemptState::Tracked { native, .. } => native + .take() + .expect("the tracked provider attempt was materialized above"), + AttemptState::NativeOnly(_) => { + unreachable!("portable providers reject native-only attempts before spawning") + } + } + } + pub(crate) fn native_for_explicit_spawn(&mut self) -> &mut B::NativeCommand { self.make_native_only(); self.prepare_platform(); diff --git a/src/lib.rs b/src/lib.rs index f2bbadd..e104f30 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -33,10 +33,9 @@ //! //! This crate provides a composable process-wrap-owned [`Command`] configuration shared by the std //! and Tokio frontends. It is a more flexible and composable successor to the `command-group` crate, -//! and is meant to be adaptable to additional use cases: for example spawning processes in PTYs -//! currently requires a different crate (such as `pty-process`) which won't function with -//! `command-group`. Implementing a PTY wrapper for `process-wrap` would instead keep the same API and -//! be composable with the existing process group/session implementations. +//! and is meant to be adaptable to additional use cases. The optional Tokio PTY provider demonstrates +//! that adaptability by keeping terminal process creation in the same wrapper lifecycle as process +//! groups, sessions, signal policy, and custom wrappers. //! //! # Usage //! @@ -109,6 +108,47 @@ //! Methods on `Child` mimic those on `process::Child`, but may be customised by the wrappers. For //! example, `kill` will send a signal to the process group if the `ProcessGroup` wrapper is used. //! +//! # Pseudo-terminals +//! +//! The non-default `pty` feature selects the Tokio frontend and terminal dependencies. It provides +//! native transport on Linux, Android, macOS, FreeBSD, NetBSD 10 and newer, OpenBSD, DragonFly BSD, +//! illumos, and Solaris; unavailable platforms report `std::io::ErrorKind::Unsupported`. +//! +//! ```rust,no_run +//! # #[cfg(feature = "pty")] +//! # mod example { +//! # fn run() -> std::io::Result<()> { +//! use process_wrap::tokio::{Command, Pty}; +//! +//! let mut command = Command::with_new("sh", |command| { +//! command.args(["-c", "printf terminal"]); +//! }); +//! command.wrap(Pty::default()); +//! let mut child = command.spawn()?; +//! let controller = child +//! .take_pty_controller() +//! .expect("a successful PTY spawn installs one controller"); +//! # drop(controller); +//! # Ok(()) } +//! # } +//! # fn main() {} +//! ``` +//! +//! A terminal has one ordered output stream, so PTY standard output and standard error are merged. +//! Input and output each strongly own the bidirectional master; resize handles are weak. Dropping one +//! I/O side is not a half-close, and the terminal hangs up only after both are gone. Send VEOF when +//! terminal input policy calls for end-of-file. Waiting for the direct child and draining terminal +//! output are separate lifecycles because descendants may retain the slave. On macOS they should run +//! concurrently while the kernel drains and revokes the terminal during session-leader teardown. +//! +//! The transport owns terminal bytes and resize, not parent-terminal raw mode, relaying, key handling, +//! VT parsing, scrollback, or pager policy. A bare PTY creates its required session. +//! `ProcessGroup::leader()` or `ProcessSession` may independently add group-wide signalling while the +//! direct child is live; waiting continues to follow that child. Attaching to an existing group or +//! explicitly registering both is invalid. `ResetSigmask` composes, while Tokio `KillOnDrop` continues +//! to target only the direct child. The returned boxed child keeps arbitrary outer wrappers, and +//! `take_pty_controller()` traverses them and yields the controller once. +//! //! # KillOnDrop and CreationFlags //! //! Calling native `.kill_on_drop()` or `.creation_flags()` makes a command native-only: those @@ -527,6 +567,7 @@ //! - `kill-on-drop`: **default**, enables the kill on drop wrapper (Tokio-only). //! - `process-group`: **default**, enables the process group wrapper (Unix-only). //! - `process-session`: **default**, enables the process session wrapper (Unix-only). +//! - `pty`: enables Tokio pseudo-terminal transport and implies `tokio1`. //! - `reset-sigmask`: enables the sigmask reset wrapper (Unix-only). //! //! ## Diagnostics diff --git a/src/std/process_group.rs b/src/std/process_group.rs index 2722df1..eff9047 100644 --- a/src/std/process_group.rs +++ b/src/std/process_group.rs @@ -1,17 +1,10 @@ use std::{ io::{Error, Result}, - ops::ControlFlow, - os::unix::process::ExitStatusExt, process::ExitStatus, }; use nix::{ - errno::Errno, - libc, - sys::{ - signal::{Signal, killpg}, - wait::WaitPidFlag, - }, + sys::signal::{Signal, killpg}, unistd::Pid, }; #[cfg(feature = "tracing")] @@ -53,25 +46,29 @@ impl ProcessGroup { } } -/// Wrapper for `Child` which ensures that all processes in the group are reaped. +/// Wrapper for `Child` which signals the process group while its direct child is live. +/// +/// Waiting follows the direct child. Process-wrap deliberately stops using a numeric process-group ID +/// after that child has been reaped, because the operating system may immediately reuse the ID for an +/// unrelated group. #[derive(Debug)] pub struct ProcessGroupChild { inner: Box, exit_status: ChildExitStatus, - direct_pid: Pid, pgid: Pid, - group_drained: bool, } impl ProcessGroupChild { #[cfg_attr(feature = "tracing", instrument(level = "debug"))] - pub(crate) fn new(inner: Box, direct_pid: Pid, pgid: Pid) -> Self { + pub(crate) fn new( + inner: Box, + pgid: Pid, + exit_status: Option, + ) -> Self { Self { inner, - exit_status: ChildExitStatus::Running, - direct_pid, + exit_status: exit_status.map_or(ChildExitStatus::Running, ChildExitStatus::Exited), pgid, - group_drained: false, } } @@ -92,73 +89,29 @@ impl CommandWrapper for ProcessGroup { #[cfg_attr(feature = "tracing", instrument(level = "debug", skip(self)))] fn wrap_child( &mut self, - inner: Box, + mut inner: Box, _core: &CommandWrap, ) -> Result> { - let direct_pid = Pid::from_raw(i32::try_from(inner.id()).expect("Command PID > i32::MAX")); + let direct_pid = Pid::from_raw(i32::try_from(inner.id()).map_err(Error::other)?); let pgid = match self.target { ProcessGroupTarget::Leader => direct_pid, ProcessGroupTarget::AttachTo(pgid) => Pid::from_raw( i32::try_from(pgid).expect("process group IDs are validated before spawning"), ), }; + let exit_status = inner.try_wait()?; - Ok(Box::new(ProcessGroupChild::new(inner, direct_pid, pgid))) + Ok(Box::new(ProcessGroupChild::new(inner, pgid, exit_status))) } } impl ProcessGroupChild { #[cfg_attr(feature = "tracing", instrument(level = "debug", skip(self)))] fn signal_imp(&self, sig: Signal) -> Result<()> { - killpg(self.pgid, sig).map_err(Error::from) - } - - #[cfg_attr(feature = "tracing", instrument(level = "debug"))] - fn wait_imp( - direct_pid: Pid, - pgid: Pid, - flag: WaitPidFlag, - ) -> Result, Option>> { - // wait for processes in a loop until every process in this group has - // exited (this ensures that we reap any zombies that may have been - // created if the parent exited after spawning children, but didn't wait - // for those children to exit) - let mut parent_exit_status: Option = None; - loop { - // we can't use the safe wrapper directly because it doesn't return - // the raw status, and we need it to convert to the std's ExitStatus - let mut status: i32 = 0; - match unsafe { - libc::waitpid(-pgid.as_raw(), &mut status as *mut libc::c_int, flag.bits()) - } { - 0 => { - // zero should only happen if WNOHANG was passed in, - // and means that no processes have yet to exit - return Ok(ControlFlow::Continue(parent_exit_status)); - } - -1 => { - match Errno::last() { - Errno::ECHILD => { - // no more children to reap; this is a graceful exit - return Ok(ControlFlow::Break(parent_exit_status)); - } - errno => { - return Err(Error::from(errno)); - } - } - } - pid => { - // a process exited. was it the parent process that we - // started? if so, collect the exit signal, otherwise we - // reaped a zombie process and should continue looping - if direct_pid == Pid::from_raw(pid) { - parent_exit_status = Some(ExitStatus::from_raw(status)); - } else { - // reaped a zombie child; keep looping - } - } - }; + if matches!(self.exit_status, ChildExitStatus::Exited(_)) { + return Ok(()); } + killpg(self.pgid, sig).map_err(Error::from) } } @@ -175,73 +128,37 @@ impl ChildWrapper for ProcessGroupChild { #[cfg_attr(feature = "tracing", instrument(level = "debug", skip(self)))] fn start_kill(&mut self) -> Result<()> { + if matches!(self.exit_status, ChildExitStatus::Running) { + if let Some(status) = self.inner.try_wait()? { + self.exit_status = ChildExitStatus::Exited(status); + } + } self.signal_imp(Signal::SIGKILL) } #[cfg_attr(feature = "tracing", instrument(level = "debug", skip(self)))] fn wait(&mut self) -> Result { - let status = match self.exit_status { + match self.exit_status { ChildExitStatus::Running => { let status = self.inner.wait()?; self.exit_status = ChildExitStatus::Exited(status); - status + Ok(status) } - ChildExitStatus::Exited(status) => status, - }; - - if !self.group_drained { - if let ControlFlow::Break(reaped) = - Self::wait_imp(self.direct_pid, self.pgid, WaitPidFlag::empty())? - { - if let Some(reaped) = reaped { - self.exit_status = ChildExitStatus::Exited(reaped); - } - self.group_drained = true; - } - } - - match self.exit_status { ChildExitStatus::Exited(status) => Ok(status), - ChildExitStatus::Running => Ok(status), } } #[cfg_attr(feature = "tracing", instrument(level = "debug", skip(self)))] fn try_wait(&mut self) -> Result> { - if self.group_drained { - return match self.exit_status { - ChildExitStatus::Exited(status) => Ok(Some(status)), - ChildExitStatus::Running => { - let status = self.inner.try_wait()?; - if let Some(status) = status { - self.exit_status = ChildExitStatus::Exited(status); - } - Ok(status) + match self.exit_status { + ChildExitStatus::Running => { + let status = self.inner.try_wait()?; + if let Some(status) = status { + self.exit_status = ChildExitStatus::Exited(status); } - }; - } - - let (drained, reaped) = - match Self::wait_imp(self.direct_pid, self.pgid, WaitPidFlag::WNOHANG)? { - ControlFlow::Break(status) => (true, status), - ControlFlow::Continue(status) => (false, status), - }; - if let Some(status) = reaped { - self.exit_status = ChildExitStatus::Exited(status); - } - if matches!(self.exit_status, ChildExitStatus::Running) { - if let Some(status) = self.inner.try_wait()? { - self.exit_status = ChildExitStatus::Exited(status); + Ok(status) } - } - self.group_drained = drained; - - if !self.group_drained { - return Ok(None); - } - match self.exit_status { ChildExitStatus::Exited(status) => Ok(Some(status)), - ChildExitStatus::Running => Ok(None), } } diff --git a/src/std/process_session.rs b/src/std/process_session.rs index a005e48..2e121de 100644 --- a/src/std/process_session.rs +++ b/src/std/process_session.rs @@ -1,4 +1,4 @@ -use std::io::Result; +use std::io::{Error, Result}; use nix::unistd::Pid; #[cfg(feature = "tracing")] @@ -30,13 +30,16 @@ impl CommandWrapper for ProcessSession { #[cfg_attr(feature = "tracing", instrument(level = "debug", skip(self)))] fn wrap_child( &mut self, - inner: Box, + mut inner: Box, _core: &CommandWrap, ) -> Result> { - let direct_pid = Pid::from_raw(i32::try_from(inner.id()).expect("Command PID > i32::MAX")); + let direct_pid = Pid::from_raw(i32::try_from(inner.id()).map_err(Error::other)?); + let exit_status = inner.try_wait()?; Ok(Box::new(super::ProcessGroupChild::new( - inner, direct_pid, direct_pid, + inner, + direct_pid, + exit_status, ))) } } diff --git a/src/tokio.rs b/src/tokio.rs index ade45c2..374af26 100644 --- a/src/tokio.rs +++ b/src/tokio.rs @@ -45,6 +45,10 @@ pub use process_group::{ProcessGroup, ProcessGroupChild}; #[cfg_attr(docsrs, doc(cfg(all(unix, feature = "process-session"))))] #[doc(inline)] pub use process_session::ProcessSession; +#[cfg(feature = "pty")] +#[cfg_attr(docsrs, doc(cfg(feature = "pty")))] +#[doc(inline)] +pub use pty::{Pty, PtyController, PtyInput, PtyOutput, PtyResize, PtySize}; #[cfg(all(unix, feature = "reset-sigmask"))] #[cfg_attr(docsrs, doc(cfg(all(unix, feature = "reset-sigmask"))))] #[doc(inline)] @@ -61,5 +65,7 @@ mod kill_on_drop; mod process_group; #[cfg(all(unix, feature = "process-session"))] mod process_session; +#[cfg(feature = "pty")] +mod pty; #[cfg(all(unix, feature = "reset-sigmask"))] mod reset_sigmask; diff --git a/src/tokio/core.rs b/src/tokio/core.rs index 4f9f165..3d40452 100644 --- a/src/tokio/core.rs +++ b/src/tokio/core.rs @@ -115,6 +115,28 @@ pub trait ChildWrapper: Any + std::fmt::Debug + Send + Sync { None } + /// Return the original PID retained by this exact provider-child layer. + /// + /// This internal capability lets a wrapper finish installation after an earlier post-spawn hook + /// observed and reaped a fast provider child. Ordinary child operations must continue to use + /// [`ChildWrapper::id`] so they never act on a recycled PID. + #[doc(hidden)] + #[cfg(all( + unix, + feature = "pty", + any(feature = "process-group", feature = "process-session") + ))] + fn spawned_id_layer(&self) -> Option { + None + } + + /// Take the PTY controller owned by this exact child layer. + #[doc(hidden)] + #[cfg(feature = "pty")] + fn take_pty_controller_layer(&mut self) -> Option { + None + } + /// Finalize Windows spawn state owned by this child layer. /// /// Process-wrap invokes this internal lifecycle hook after all child wrappers have been installed. @@ -340,6 +362,51 @@ impl dyn ChildWrapper + '_ { self.downcast_ref::().is_some() } + #[cfg(all( + unix, + feature = "pty", + any(feature = "process-group", feature = "process-session") + ))] + pub(crate) fn try_spawned_id(&self) -> Option { + let mut inner = self; + loop { + if let Some(pid) = inner.spawned_id_layer() { + return Some(pid); + } + + let next = inner.inner(); + if same_child(inner, next) { + return None; + } + inner = next; + } + } + + /// Take the controller installed by a PTY spawn. + /// + /// This traverses arbitrary child-wrapper layers without removing them. The controller can be taken + /// only once; subsequent calls and non-PTY children return `None`. + #[cfg(feature = "pty")] + #[cfg_attr(docsrs, doc(cfg(feature = "pty")))] + pub fn take_pty_controller(&mut self) -> Option { + let mut inner = self; + loop { + if let Some(controller) = inner.take_pty_controller_layer() { + return Some(controller); + } + + let inner_type = (&*inner as &dyn Any).type_id(); + let inner_ptr = std::ptr::from_mut(inner); + let next = inner.inner_mut(); + if std::ptr::addr_eq(inner_ptr, std::ptr::from_mut(next)) + && inner_type == (&*next as &dyn Any).type_id() + { + return None; + } + inner = next; + } + } + /// Find the first Windows process-handle capability in this wrapper chain. /// /// Unlike [`ChildWrapper::process_handle`], this traverses legacy transparent layers which do not diff --git a/src/tokio/kill_on_drop.rs b/src/tokio/kill_on_drop.rs index 7d0ae3d..15b71b8 100644 --- a/src/tokio/kill_on_drop.rs +++ b/src/tokio/kill_on_drop.rs @@ -7,6 +7,10 @@ use super::{CommandWrap, CommandWrapper, SpawnAttempt}; /// Calling the native-shaped `Command::kill_on_drop` method makes the command native-only because the /// setting cannot be queried afterward. This wrapper instead records the policy on each `SpawnAttempt`, /// allowing `JobObject` and alternate spawn providers to preserve it. +/// +/// On Unix, dropping a process-group or session child still applies Tokio's kill-on-drop behavior to +/// the direct native child only. Use the group-aware child's explicit kill or signal methods when the +/// entire process group must be targeted. #[derive(Clone, Copy, Debug)] pub struct KillOnDrop; diff --git a/src/tokio/process_group.rs b/src/tokio/process_group.rs index f6bd5cf..6017041 100644 --- a/src/tokio/process_group.rs +++ b/src/tokio/process_group.rs @@ -1,22 +1,14 @@ use std::{ future::Future, io::{Error, Result}, - ops::ControlFlow, - os::unix::process::ExitStatusExt, pin::Pin, process::ExitStatus, }; use nix::{ - errno::Errno, - libc, - sys::{ - signal::{Signal, killpg}, - wait::WaitPidFlag, - }, + sys::signal::{Signal, killpg}, unistd::Pid, }; -use tokio::task::spawn_blocking; #[cfg(feature = "tracing")] use tracing::instrument; @@ -34,6 +26,11 @@ use super::{ChildWrapper, CommandWrap, CommandWrapper, SpawnAttempt}; /// Process groups direct signals to all members of the group, and also serve to control job /// placement in foreground or background, among other actions. /// +/// With the `Pty` wrapper, [`leader`](Self::leader) leaves session and process-group creation to the +/// terminal provider while retaining group-wide signalling until the direct child exits. Waiting +/// still follows that child. A PTY must create a new session, so [`attach_to`](Self::attach_to) is +/// invalid for that transport. +/// /// This wrapper provides a child wrapper: [`ProcessGroupChild`]. #[derive(Clone, Copy, Debug)] pub struct ProcessGroup { @@ -56,25 +53,29 @@ impl ProcessGroup { } } -/// Wrapper for `Child` which ensures that all processes in the group are reaped. +/// Wrapper for `Child` which signals the process group while its direct child is live. +/// +/// Waiting follows the direct child. Process-wrap deliberately stops using a numeric process-group ID +/// after that child has been reaped, because the operating system may immediately reuse the ID for an +/// unrelated group. #[derive(Debug)] pub struct ProcessGroupChild { inner: Box, exit_status: ChildExitStatus, - direct_pid: Pid, pgid: Pid, - group_drained: bool, } impl ProcessGroupChild { #[cfg_attr(feature = "tracing", instrument(level = "debug"))] - pub(crate) fn new(inner: Box, direct_pid: Pid, pgid: Pid) -> Self { + pub(crate) fn new( + inner: Box, + pgid: Pid, + exit_status: Option, + ) -> Self { Self { inner, - exit_status: ChildExitStatus::Running, - direct_pid, + exit_status: exit_status.map_or(ChildExitStatus::Running, ChildExitStatus::Exited), pgid, - group_drained: false, } } @@ -95,80 +96,40 @@ impl CommandWrapper for ProcessGroup { #[cfg_attr(feature = "tracing", instrument(level = "debug", skip(self)))] fn wrap_child( &mut self, - inner: Box, + mut inner: Box, _core: &CommandWrap, ) -> Result> { - let direct_pid = Pid::from_raw( - i32::try_from( - inner - .id() - .expect("Command was reaped before we could read its PID"), + let mut direct_id = inner.id(); + #[cfg(feature = "pty")] + if direct_id.is_none() { + direct_id = inner.try_spawned_id(); + } + let direct_id = direct_id.ok_or_else(|| { + Error::new( + std::io::ErrorKind::InvalidInput, + "the child exited before process-group supervision could retain its PID", ) - .expect("Command PID > i32::MAX"), - ); + })?; + let direct_pid = Pid::from_raw(i32::try_from(direct_id).map_err(Error::other)?); let pgid = match self.target { ProcessGroupTarget::Leader => direct_pid, ProcessGroupTarget::AttachTo(pgid) => Pid::from_raw( i32::try_from(pgid).expect("process group IDs are validated before spawning"), ), }; + let exit_status = inner.try_wait()?; - Ok(Box::new(ProcessGroupChild::new(inner, direct_pid, pgid))) + Ok(Box::new(ProcessGroupChild::new(inner, pgid, exit_status))) } } impl ProcessGroupChild { #[cfg_attr(feature = "tracing", instrument(level = "debug", skip(self)))] fn signal_imp(&self, sig: Signal) -> Result<()> { - killpg(self.pgid, sig).map_err(Error::from) - } - - #[cfg_attr(feature = "tracing", instrument(level = "debug"))] - fn wait_imp( - direct_pid: Pid, - pgid: Pid, - flag: WaitPidFlag, - ) -> Result, Option>> { - // wait for processes in a loop until every process in this group has - // exited (this ensures that we reap any zombies that may have been - // created if the parent exited after spawning children, but didn't wait - // for those children to exit) - let mut parent_exit_status: Option = None; - loop { - // we can't use the safe wrapper directly because it doesn't return - // the raw status, and we need it to convert to the std's ExitStatus - let mut status: i32 = 0; - match unsafe { - libc::waitpid(-pgid.as_raw(), &mut status as *mut libc::c_int, flag.bits()) - } { - 0 => { - // zero should only happen if WNOHANG was passed in, - // and means that no processes have yet to exit - return Ok(ControlFlow::Continue(parent_exit_status)); - } - -1 => { - match Errno::last() { - Errno::ECHILD => { - // no more children to reap; this is a graceful exit - return Ok(ControlFlow::Break(parent_exit_status)); - } - errno => { - return Err(Error::from(errno)); - } - } - } - pid => { - // a process exited. was it the parent process that we - // started? if so, collect the exit signal, otherwise we - // reaped a zombie process and should continue looping - if direct_pid == Pid::from_raw(pid) { - parent_exit_status = Some(ExitStatus::from_raw(status)); - } else { - // reaped a zombie child; keep looping - } - } - }; + if matches!(self.exit_status, ChildExitStatus::Exited(_)) || self.inner.id().is_none() { + return Ok(()); } + killpg(self.pgid, sig).map_err(Error::from) } } @@ -185,98 +146,39 @@ impl ChildWrapper for ProcessGroupChild { #[cfg_attr(feature = "tracing", instrument(level = "debug", skip(self)))] fn start_kill(&mut self) -> Result<()> { + if matches!(self.exit_status, ChildExitStatus::Running) { + if let Some(status) = self.inner.try_wait()? { + self.exit_status = ChildExitStatus::Exited(status); + } + } self.signal_imp(Signal::SIGKILL) } #[cfg_attr(feature = "tracing", instrument(level = "debug", skip(self)))] fn wait(&mut self) -> Pin> + Send + '_>> { Box::pin(async { - let status = match self.exit_status { + match self.exit_status { ChildExitStatus::Running => { let status = self.inner.wait().await?; self.exit_status = ChildExitStatus::Exited(status); - status - } - ChildExitStatus::Exited(status) => status, - }; - - if !self.group_drained { - const MAX_RETRY_ATTEMPT: usize = 10; - for _ in 1..MAX_RETRY_ATTEMPT { - match Self::wait_imp(self.direct_pid, self.pgid, WaitPidFlag::WNOHANG)? { - ControlFlow::Break(reaped) => { - if let Some(reaped) = reaped { - self.exit_status = ChildExitStatus::Exited(reaped); - } - self.group_drained = true; - break; - } - ControlFlow::Continue(reaped) => { - if let Some(reaped) = reaped { - self.exit_status = ChildExitStatus::Exited(reaped); - } - } - } - } - } - - if !self.group_drained { - let direct_pid = self.direct_pid; - let pgid = self.pgid; - let result = - spawn_blocking(move || Self::wait_imp(direct_pid, pgid, WaitPidFlag::empty())) - .await??; - if let ControlFlow::Break(reaped) = result { - if let Some(reaped) = reaped { - self.exit_status = ChildExitStatus::Exited(reaped); - } - self.group_drained = true; + Ok(status) } - } - - match self.exit_status { ChildExitStatus::Exited(status) => Ok(status), - ChildExitStatus::Running => Ok(status), } }) } #[cfg_attr(feature = "tracing", instrument(level = "debug", skip(self)))] fn try_wait(&mut self) -> Result> { - if self.group_drained { - return match self.exit_status { - ChildExitStatus::Exited(status) => Ok(Some(status)), - ChildExitStatus::Running => { - let status = self.inner.try_wait()?; - if let Some(status) = status { - self.exit_status = ChildExitStatus::Exited(status); - } - Ok(status) + match self.exit_status { + ChildExitStatus::Running => { + let status = self.inner.try_wait()?; + if let Some(status) = status { + self.exit_status = ChildExitStatus::Exited(status); } - }; - } - - let (drained, reaped) = - match Self::wait_imp(self.direct_pid, self.pgid, WaitPidFlag::WNOHANG)? { - ControlFlow::Break(status) => (true, status), - ControlFlow::Continue(status) => (false, status), - }; - if let Some(status) = reaped { - self.exit_status = ChildExitStatus::Exited(status); - } - if matches!(self.exit_status, ChildExitStatus::Running) { - if let Some(status) = self.inner.try_wait()? { - self.exit_status = ChildExitStatus::Exited(status); + Ok(status) } - } - self.group_drained = drained; - - if !self.group_drained { - return Ok(None); - } - match self.exit_status { ChildExitStatus::Exited(status) => Ok(Some(status)), - ChildExitStatus::Running => Ok(None), } } diff --git a/src/tokio/process_session.rs b/src/tokio/process_session.rs index 61b5077..80d658f 100644 --- a/src/tokio/process_session.rs +++ b/src/tokio/process_session.rs @@ -1,4 +1,4 @@ -use std::io::Result; +use std::io::{Error, ErrorKind, Result}; use nix::unistd::Pid; #[cfg(feature = "tracing")] @@ -18,6 +18,9 @@ use super::{CommandWrap, CommandWrapper, SpawnAttempt}; /// /// This wrapper uses [the same child wrapper as `ProcessGroup`](super::ProcessGroupChild) and does /// the same setup (plus the session setup); using both together is unnecessary and may misbehave. +/// With the `Pty` wrapper, the terminal provider performs the required session setup while this +/// wrapper retains group-wide signalling until the direct child exits. Waiting still follows that +/// child. Explicitly combining both supervision wrappers is invalid for that transport. #[derive(Clone, Copy, Debug)] pub struct ProcessSession; @@ -30,20 +33,27 @@ impl CommandWrapper for ProcessSession { #[cfg_attr(feature = "tracing", instrument(level = "debug", skip(self)))] fn wrap_child( &mut self, - inner: Box, + mut inner: Box, _core: &CommandWrap, ) -> Result> { - let direct_pid = Pid::from_raw( - i32::try_from( - inner - .id() - .expect("Command was reaped before we could read its PID"), + let mut direct_id = inner.id(); + #[cfg(feature = "pty")] + if direct_id.is_none() { + direct_id = inner.try_spawned_id(); + } + let direct_id = direct_id.ok_or_else(|| { + Error::new( + ErrorKind::InvalidInput, + "the child exited before session supervision could retain its PID", ) - .expect("Command PID > i32::MAX"), - ); + })?; + let direct_pid = Pid::from_raw(i32::try_from(direct_id).map_err(Error::other)?); + let exit_status = inner.try_wait()?; Ok(Box::new(super::ProcessGroupChild::new( - inner, direct_pid, direct_pid, + inner, + direct_pid, + exit_status, ))) } } diff --git a/src/tokio/pty.rs b/src/tokio/pty.rs new file mode 100644 index 0000000..f6b4400 --- /dev/null +++ b/src/tokio/pty.rs @@ -0,0 +1,554 @@ +use std::{ + io, + pin::Pin, + task::{Context, Poll}, +}; + +#[cfg(any( + target_os = "android", + target_os = "dragonfly", + target_os = "freebsd", + target_os = "illumos", + target_os = "linux", + target_os = "macos", + target_os = "netbsd", + target_os = "openbsd", + target_os = "solaris" +))] +use std::{ + future::Future, + process::ExitStatus, + sync::{ + Arc, Mutex, + atomic::{AtomicBool, Ordering}, + }, +}; + +use tokio::io::{AsyncRead, AsyncWrite, ReadBuf}; +#[cfg(any( + target_os = "android", + target_os = "dragonfly", + target_os = "freebsd", + target_os = "illumos", + target_os = "linux", + target_os = "macos", + target_os = "netbsd", + target_os = "openbsd", + target_os = "solaris" +))] +use tokio::process::{Child, ChildStderr, ChildStdin, ChildStdout}; + +#[cfg(any( + target_os = "android", + target_os = "dragonfly", + target_os = "freebsd", + target_os = "illumos", + target_os = "linux", + target_os = "macos", + target_os = "netbsd", + target_os = "openbsd", + target_os = "solaris" +))] +use super::ChildWrapper; +use super::{Command, CommandWrapper, ProviderProduct, SpawnAttempt, SpawnProvider}; + +#[cfg(any( + target_os = "android", + target_os = "dragonfly", + target_os = "freebsd", + target_os = "illumos", + target_os = "linux", + target_os = "macos", + target_os = "netbsd", + target_os = "openbsd", + target_os = "solaris" +))] +mod unix; +#[cfg(not(any( + target_os = "android", + target_os = "dragonfly", + target_os = "freebsd", + target_os = "illumos", + target_os = "linux", + target_os = "macos", + target_os = "netbsd", + target_os = "openbsd", + target_os = "solaris" +)))] +mod unsupported; +#[cfg(any( + target_os = "android", + target_os = "dragonfly", + target_os = "freebsd", + target_os = "illumos", + target_os = "linux", + target_os = "macos", + target_os = "netbsd", + target_os = "openbsd", + target_os = "solaris" +))] +use unix as imp; +#[cfg(not(any( + target_os = "android", + target_os = "dragonfly", + target_os = "freebsd", + target_os = "illumos", + target_os = "linux", + target_os = "macos", + target_os = "netbsd", + target_os = "openbsd", + target_os = "solaris" +)))] +use unsupported as imp; + +/// A pseudo-terminal spawn provider for Tokio commands. +/// +/// Register this like any other process wrapper. Spawning still returns the ordinary boxed Tokio +/// child contract; call `take_pty_controller()` on that child to take its terminal I/O and resize +/// controller once. +/// +/// ```rust,no_run +/// # use std::io; +/// use process_wrap::tokio::{Command, Pty}; +/// # fn run() -> io::Result<()> { +/// let mut command = Command::with_new("sh", |command| { +/// command.args(["-c", "printf terminal"]); +/// }); +/// command.wrap(Pty::default()); +/// let mut child = command.spawn()?; +/// let controller = child +/// .take_pty_controller() +/// .expect("a successful PTY spawn installs one controller"); +/// # drop(controller); +/// # Ok(()) +/// # } +/// # fn main() {} +/// ``` +#[derive(Clone, Copy, Debug, Default, Eq, PartialEq)] +pub struct Pty { + size: PtySize, +} + +impl Pty { + /// Construct a PTY provider with the requested initial terminal size. + pub fn new(size: PtySize) -> Self { + Self { size } + } + + /// Return the configured initial terminal size. + pub fn size(&self) -> PtySize { + self.size + } +} + +impl CommandWrapper for Pty { + fn extend(&mut self, other: Self) { + *self = other; + } + + fn spawn_provider(&self) -> Option<&dyn SpawnProvider> { + Some(self) + } +} + +impl SpawnProvider for Pty { + fn check_available(&self) -> io::Result<()> { + imp::check_available() + } + + fn validate_command(&self, _command: &Command) -> io::Result<()> { + self.size.validate() + } + + fn validate_attempt(&self, attempt: &SpawnAttempt, _command: &Command) -> io::Result<()> { + #[cfg(unix)] + { + match attempt.process_group_target() { + Some(crate::ProcessGroupTarget::AttachTo(_)) => { + return Err(io::Error::new( + io::ErrorKind::InvalidInput, + "ProcessGroup::attach_to cannot be used with a PTY", + )); + } + Some(crate::ProcessGroupTarget::Leader) if attempt.creates_process_session() => { + return Err(io::Error::new( + io::ErrorKind::InvalidInput, + "ProcessGroup and ProcessSession cannot both be used with a PTY", + )); + } + None | Some(crate::ProcessGroupTarget::Leader) => {} + } + } + #[cfg(not(unix))] + let _ = attempt; + Ok(()) + } + + fn spawn(&self, attempt: &mut SpawnAttempt, _command: &Command) -> io::Result { + imp::spawn(attempt, self.size) + } +} + +/// The character and pixel dimensions of a pseudo-terminal. +/// +/// Character dimensions must both be nonzero. Pixel dimensions may be zero when they are unknown or +/// not meaningful to the caller. Sizes are validated before spawning and by [`PtyResize::resize`]. +#[derive(Clone, Copy, Debug, Eq, PartialEq)] +pub struct PtySize { + /// Character rows. + pub rows: u16, + /// Character columns. + pub columns: u16, + /// Pixel width, or zero when unspecified. + pub pixel_width: u16, + /// Pixel height, or zero when unspecified. + pub pixel_height: u16, +} + +impl PtySize { + /// Construct a character-cell size with unspecified pixel dimensions. + pub fn new(rows: u16, columns: u16) -> io::Result { + let size = Self { + rows, + columns, + pixel_width: 0, + pixel_height: 0, + }; + size.validate()?; + Ok(size) + } + + /// Set the optional pixel dimensions. + pub fn with_pixels(mut self, pixel_width: u16, pixel_height: u16) -> Self { + self.pixel_width = pixel_width; + self.pixel_height = pixel_height; + self + } + + /// Validate that both character dimensions are nonzero. + pub fn validate(self) -> io::Result<()> { + if self.rows == 0 || self.columns == 0 { + return Err(io::Error::new( + io::ErrorKind::InvalidInput, + "PTY rows and columns must both be nonzero", + )); + } + Ok(()) + } +} + +impl Default for PtySize { + fn default() -> Self { + Self { + rows: 24, + columns: 80, + pixel_width: 0, + pixel_height: 0, + } + } +} + +/// The asynchronous input side of a pseudo-terminal. +/// +/// Input and output are strong owners of the same bidirectional PTY master. Shutting down or +/// dropping only this input object releases its owner but cannot close the master or produce child +/// EOF while [`PtyOutput`] still exists. There is no independent transport half-close; send the +/// terminal's VEOF control character when that is the desired terminal policy. +#[derive(Debug)] +pub struct PtyInput { + inner: imp::Input, +} + +impl AsyncWrite for PtyInput { + fn poll_write( + mut self: Pin<&mut Self>, + cx: &mut Context<'_>, + buffer: &[u8], + ) -> Poll> { + Pin::new(&mut self.inner).poll_write(cx, buffer) + } + + fn poll_flush(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll> { + Pin::new(&mut self.inner).poll_flush(cx) + } + + fn poll_shutdown(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll> { + Pin::new(&mut self.inner).poll_shutdown(cx) + } +} + +/// The asynchronous merged output side of a pseudo-terminal. +/// +/// This is a strong owner of the shared bidirectional PTY master. On most supported Unix systems, +/// output EOF means that every slave descriptor has closed and a descendant may retain the slave +/// after the direct child exits. On macOS, drain output concurrently with waiting: session-leader +/// exit drains queued output and then revokes the controlling terminal from its descendants. +#[derive(Debug)] +pub struct PtyOutput { + inner: imp::Output, +} + +impl AsyncRead for PtyOutput { + fn poll_read( + mut self: Pin<&mut Self>, + cx: &mut Context<'_>, + buffer: &mut ReadBuf<'_>, + ) -> Poll> { + Pin::new(&mut self.inner).poll_read(cx, buffer) + } +} + +/// A cloneable weak handle for resizing a pseudo-terminal. +/// +/// Resize handles do not keep the PTY master alive. +#[derive(Clone, Debug)] +pub struct PtyResize { + inner: imp::Resize, +} + +impl PtyResize { + /// Change the terminal size. + /// + /// Returns [`io::ErrorKind::BrokenPipe`] after both strong I/O owners have been dropped. + pub fn resize(&self, size: PtySize) -> io::Result<()> { + size.validate()?; + self.inner.resize(size) + } +} + +/// The I/O and resize controls for a spawned pseudo-terminal. +/// +/// [`PtyInput`] and [`PtyOutput`] each strongly own one shared bidirectional master. The terminal +/// hangs up only after both owners are gone; dropping either one alone is not a half-close. A +/// [`PtyResize`] is weak and never keeps the terminal alive. There is deliberately no clonable +/// force-close handle, parent-terminal raw mode, byte relay, key handling, VT parsing, scrollback, +/// or pager policy in this transport. +/// +/// Process supervision and PTY draining are separate lifecycles. Waiting for the direct child does +/// not imply output EOF on most supported Unix systems, because descendants can retain slave +/// descriptors. On macOS, drive waiting and draining concurrently because session-leader exit waits +/// for queued output before revoking the controlling terminal. +#[derive(Debug)] +pub struct PtyController { + input: PtyInput, + output: PtyOutput, + resize: PtyResize, +} + +impl PtyController { + #[allow(dead_code)] + fn new(input: imp::Input, output: imp::Output, resize: imp::Resize) -> Self { + Self { + input: PtyInput { inner: input }, + output: PtyOutput { inner: output }, + resize: PtyResize { inner: resize }, + } + } + + /// Borrow the input side. + pub fn input(&mut self) -> &mut PtyInput { + &mut self.input + } + + /// Borrow the merged output side. + pub fn output(&mut self) -> &mut PtyOutput { + &mut self.output + } + + /// Clone the weak resize handle. + pub fn resizer(&self) -> PtyResize { + self.resize.clone() + } + + /// Split the controller into independently owned input, output, and resize handles. + pub fn into_parts(self) -> (PtyInput, PtyOutput, PtyResize) { + (self.input, self.output, self.resize) + } +} + +#[cfg(any( + target_os = "android", + target_os = "dragonfly", + target_os = "freebsd", + target_os = "illumos", + target_os = "linux", + target_os = "macos", + target_os = "netbsd", + target_os = "openbsd", + target_os = "solaris" +))] +#[derive(Debug)] +pub(super) struct ControllerSlot { + controller: Mutex>, + committed: AtomicBool, +} + +#[cfg(any( + target_os = "android", + target_os = "dragonfly", + target_os = "freebsd", + target_os = "illumos", + target_os = "linux", + target_os = "macos", + target_os = "netbsd", + target_os = "openbsd", + target_os = "solaris" +))] +impl ControllerSlot { + pub(super) fn new(controller: PtyController) -> Self { + Self { + controller: Mutex::new(Some(controller)), + committed: AtomicBool::new(false), + } + } + + pub(super) fn commit(&self) { + self.committed.store(true, Ordering::Release); + } + + pub(super) fn rollback(&self) { + self.controller + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner) + .take(); + } + + fn take(&self) -> Option { + if !self.committed.load(Ordering::Acquire) { + return None; + } + self.controller + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner) + .take() + } +} + +#[cfg(any( + target_os = "android", + target_os = "dragonfly", + target_os = "freebsd", + target_os = "illumos", + target_os = "linux", + target_os = "macos", + target_os = "netbsd", + target_os = "openbsd", + target_os = "solaris" +))] +#[derive(Debug)] +pub(super) struct PtyChild { + child: Arc>, + #[cfg_attr( + not(any(feature = "process-group", feature = "process-session")), + allow(dead_code) + )] + pid: u32, + controller: Arc, + stdin: Option, + stdout: Option, + stderr: Option, +} + +#[cfg(any( + target_os = "android", + target_os = "dragonfly", + target_os = "freebsd", + target_os = "illumos", + target_os = "linux", + target_os = "macos", + target_os = "netbsd", + target_os = "openbsd", + target_os = "solaris" +))] +impl PtyChild { + pub(super) fn new(child: Arc>, pid: u32, controller: Arc) -> Self { + Self { + child, + pid, + controller, + stdin: None, + stdout: None, + stderr: None, + } + } + + fn lock_child(&self) -> std::sync::MutexGuard<'_, Child> { + self.child + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner) + } +} + +#[cfg(any( + target_os = "android", + target_os = "dragonfly", + target_os = "freebsd", + target_os = "illumos", + target_os = "linux", + target_os = "macos", + target_os = "netbsd", + target_os = "openbsd", + target_os = "solaris" +))] +impl ChildWrapper for PtyChild { + fn inner(&self) -> &dyn ChildWrapper { + self + } + + fn inner_mut(&mut self) -> &mut dyn ChildWrapper { + self + } + + fn into_inner(self: Box) -> Box { + self + } + + fn stdin(&mut self) -> &mut Option { + &mut self.stdin + } + + fn stdout(&mut self) -> &mut Option { + &mut self.stdout + } + + fn stderr(&mut self) -> &mut Option { + &mut self.stderr + } + + fn id(&self) -> Option { + self.lock_child().id() + } + + fn start_kill(&mut self) -> io::Result<()> { + self.lock_child().start_kill() + } + + fn try_wait(&mut self) -> io::Result> { + self.lock_child().try_wait() + } + + fn wait(&mut self) -> Pin> + Send + '_>> { + let child = Arc::clone(&self.child); + Box::pin(std::future::poll_fn(move |cx| { + let mut child = child + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner); + let mut wait = Box::pin(child.wait()); + wait.as_mut().poll(cx) + })) + } + + #[cfg(unix)] + fn signal(&self, sig: i32) -> io::Result<()> { + ChildWrapper::signal(&*self.lock_child(), sig) + } + + #[cfg(any(feature = "process-group", feature = "process-session"))] + fn spawned_id_layer(&self) -> Option { + Some(self.pid) + } + + fn take_pty_controller_layer(&mut self) -> Option { + self.controller.take() + } +} diff --git a/src/tokio/pty/unix.rs b/src/tokio/pty/unix.rs new file mode 100644 index 0000000..0272aa8 --- /dev/null +++ b/src/tokio/pty/unix.rs @@ -0,0 +1,728 @@ +#[cfg(any( + target_os = "dragonfly", + target_os = "freebsd", + target_os = "macos", + target_os = "netbsd" +))] +use std::ffi::CStr; +#[cfg(any(target_os = "dragonfly", target_os = "illumos", target_os = "solaris"))] +use std::ffi::CString; +#[cfg(any( + target_os = "android", + target_os = "illumos", + target_os = "linux", + target_os = "solaris" +))] +use std::path::Path; +use std::{ + io, + os::fd::{AsFd, AsRawFd, FromRawFd, OwnedFd}, + panic::{AssertUnwindSafe, catch_unwind, resume_unwind}, + pin::Pin, + process::Stdio, + sync::{Arc, Mutex, Weak}, + task::{Context, Poll, ready}, +}; + +#[cfg(any(target_os = "android", target_os = "linux"))] +use nix::pty::ptsname_r; +#[cfg(any( + target_os = "android", + target_os = "dragonfly", + target_os = "freebsd", + target_os = "linux", + target_os = "macos", + target_os = "netbsd" +))] +use nix::pty::{PtyMaster, grantpt, posix_openpt, unlockpt}; +#[cfg(any(target_os = "illumos", target_os = "solaris"))] +use nix::sys::stat::fstat; +#[cfg(any( + target_os = "android", + target_os = "dragonfly", + target_os = "freebsd", + target_os = "illumos", + target_os = "linux", + target_os = "macos", + target_os = "netbsd", + target_os = "solaris" +))] +use nix::{fcntl::open, sys::stat::Mode}; +use nix::{ + fcntl::{FcntlArg, FdFlag, OFlag, fcntl}, + libc, + pty::Winsize, +}; +#[cfg(any( + target_os = "android", + target_os = "dragonfly", + target_os = "freebsd", + target_os = "linux", + target_os = "macos", + target_os = "netbsd" +))] +use std::os::fd::IntoRawFd; +use tokio::io::{AsyncRead, AsyncWrite, ReadBuf, unix::AsyncFd}; + +use super::{ControllerSlot, ProviderProduct, PtyChild, PtyController, PtySize, SpawnAttempt}; +use crate::SpawnTransaction; + +#[cfg(any(target_os = "openbsd", all(test, target_os = "linux")))] +mod descriptor_pair; +#[cfg(target_os = "openbsd")] +mod openbsd; + +type Master = Arc>; + +#[derive(Debug)] +pub(super) struct Input { + master: Option, +} + +impl AsyncWrite for Input { + fn poll_write( + self: Pin<&mut Self>, + cx: &mut Context<'_>, + buffer: &[u8], + ) -> Poll> { + if buffer.is_empty() { + return Poll::Ready(Ok(0)); + } + + let Some(master) = self.get_mut().master.as_ref() else { + return Poll::Ready(Err(closed())); + }; + + loop { + let mut ready = ready!(master.poll_write_ready(cx))?; + match ready.try_io(|master| write(master.get_ref(), buffer)) { + Ok(result) => return Poll::Ready(result), + Err(_would_block) => continue, + } + } + } + + fn poll_flush(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll> { + Poll::Ready(Ok(())) + } + + fn poll_shutdown(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll> { + self.get_mut().master.take(); + Poll::Ready(Ok(())) + } +} + +#[derive(Debug)] +pub(super) struct Output { + master: Master, +} + +impl AsyncRead for Output { + fn poll_read( + self: Pin<&mut Self>, + cx: &mut Context<'_>, + buffer: &mut ReadBuf<'_>, + ) -> Poll> { + if buffer.remaining() == 0 { + return Poll::Ready(Ok(())); + } + + loop { + let mut ready = ready!(self.master.poll_read_ready(cx))?; + let result = + ready.try_io(|master| read(master.get_ref(), buffer.initialize_unfilled())); + match result { + Ok(Ok(read)) => { + buffer.advance(read); + return Poll::Ready(Ok(())); + } + Ok(Err(error)) if is_eof(&error) => return Poll::Ready(Ok(())), + Ok(Err(error)) => return Poll::Ready(Err(error)), + Err(_would_block) => continue, + } + } + } +} + +#[derive(Clone, Debug)] +pub(super) struct Resize { + master: Weak>, +} + +impl Resize { + pub(super) fn resize(&self, size: PtySize) -> io::Result<()> { + let master = self.master.upgrade().ok_or_else(closed)?; + set_size(master.get_ref(), size) + } +} + +pub(super) fn check_available() -> io::Result<()> { + #[cfg(target_os = "netbsd")] + check_netbsd_version()?; + Ok(()) +} + +#[cfg(target_os = "netbsd")] +fn check_netbsd_version() -> io::Result<()> { + let mut name = std::mem::MaybeUninit::::uninit(); + // SAFETY: uname initializes the complete utsname value when it succeeds. + if unsafe { libc::uname(name.as_mut_ptr()) } == -1 { + return Err(io::Error::last_os_error()); + } + // SAFETY: uname succeeded, and utsname release is a NUL-terminated character array. + let name = unsafe { name.assume_init() }; + let release = unsafe { CStr::from_ptr(name.release.as_ptr()) }.to_bytes(); + let digits = release.iter().copied().take_while(u8::is_ascii_digit); + let mut major = None; + for digit in digits { + major = Some( + major + .unwrap_or(0_u32) + .checked_mul(10) + .and_then(|value| value.checked_add(u32::from(digit - b'0'))) + .ok_or_else(|| { + io::Error::new(io::ErrorKind::InvalidData, "invalid NetBSD release") + })?, + ); + } + if major.is_some_and(|major| major >= 10) { + return Ok(()); + } + Err(io::Error::new( + io::ErrorKind::Unsupported, + "atomic PTY descriptors require NetBSD 10 or newer", + )) +} + +pub(super) fn spawn(attempt: &mut SpawnAttempt, size: PtySize) -> io::Result { + let kill_on_drop = attempt.kills_on_drop(); + let reset_sigmask = attempt.resets_sigmask(); + let (master, slave) = open_pty(size)?; + let slave_stdin = duplicate(&slave)?; + let slave_stdout = duplicate(&slave)?; + + let master = Arc::new(AsyncFd::new(master)?); + let input = Input { + master: Some(Arc::clone(&master)), + }; + let resize = Resize { + master: Arc::downgrade(&master), + }; + let output = Output { master }; + let controller = Arc::new(ControllerSlot::new(PtyController::new( + input, output, resize, + ))); + + let mut command = attempt.take_native_for_provider_spawn(); + command.kill_on_drop(kill_on_drop); + let spawned = catch_unwind(AssertUnwindSafe(|| { + with_slave_stdio(&mut command, slave_stdin, slave_stdout, slave, |command| { + // SAFETY: the callback only invokes async-signal-safe libc functions and reports the + // operating system's error without accessing shared process state. + unsafe { + command.pre_exec(move || setup_child(reset_sigmask)); + } + command.spawn() + }) + })); + let child = match spawned { + Ok(child) => child?, + Err(payload) => resume_unwind(payload), + }; + let pid = child + .id() + .expect("Tokio reports a process ID for a newly spawned child"); + let child = Arc::new(Mutex::new(child)); + let transaction = PtyTransaction::new(Arc::clone(&child), Arc::clone(&controller)); + let child = PtyChild::new(child, pid, controller); + + Ok(ProviderProduct::new(Box::new(child), Box::new(transaction))) +} + +fn with_slave_stdio( + command: &mut tokio::process::Command, + stdin: OwnedFd, + stdout: OwnedFd, + stderr: OwnedFd, + operation: impl FnOnce(&mut tokio::process::Command) -> T, +) -> T { + command + .stdin(Stdio::from(stdin)) + .stdout(Stdio::from(stdout)) + .stderr(Stdio::from(stderr)); + let result = catch_unwind(AssertUnwindSafe(|| operation(command))); + + // Drop every parent-side slave descriptor before the provider returns and process-wrap runs + // post-spawn or child-wrapping hooks. Stdio::null stores no open descriptor on this attempt. + command + .stdin(Stdio::null()) + .stdout(Stdio::null()) + .stderr(Stdio::null()); + + match result { + Ok(result) => result, + Err(payload) => resume_unwind(payload), + } +} + +#[derive(Debug)] +struct PtyTransaction { + child: Arc>, + controller: Arc, + armed: bool, +} + +impl PtyTransaction { + fn new(child: Arc>, controller: Arc) -> Self { + Self { + child, + controller, + armed: true, + } + } + + fn rollback_inner(&mut self) -> io::Result<()> { + if !std::mem::replace(&mut self.armed, false) { + return Ok(()); + } + self.controller.rollback(); + terminate_and_reap(&self.child) + } +} + +impl SpawnTransaction for PtyTransaction { + fn commit(&mut self) -> io::Result<()> { + if self.armed { + self.controller.commit(); + self.armed = false; + } + Ok(()) + } + + fn rollback(&mut self) -> io::Result<()> { + self.rollback_inner() + } +} + +impl Drop for PtyTransaction { + fn drop(&mut self) { + let _ = self.rollback_inner(); + } +} + +fn terminate_and_reap(child: &Mutex) -> io::Result<()> { + let mut child = child + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner); + let Some(pid) = child.id() else { + // A hook already observed and reaped the direct child through this shared object. Its cached + // numeric identity is no longer safe for signalling. + return Ok(()); + }; + let pid = libc::pid_t::try_from(pid).map_err(io::Error::other)?; + + // The PTY setup makes the direct child the leader of a fresh session and process group. Until the + // direct child is reaped, its PID anchors that group identity even if it has already become a zombie, + // so signalling the negative PID cannot target a recycled group. Signal before try_wait can reap an + // exited leader; this also cleans up descendants which are still present at rollback time. + // SAFETY: `pid` is positive and belongs to the unreaped child locked above. + let group_error = if unsafe { libc::kill(-pid, libc::SIGKILL) } == -1 { + let error = io::Error::last_os_error(); + (error.raw_os_error() != Some(libc::ESRCH)).then_some(error) + } else { + None + }; + + // Kill the direct child independently in case process-group signalling was unavailable. Holding the + // only operational child lock keeps wait/kill state synchronized with every PTY child capability. + if let Err(error) = child.start_kill() { + if error.kind() != io::ErrorKind::InvalidInput { + return Err(error); + } + } + + loop { + if child.try_wait()?.is_some() { + return match group_error { + Some(error) => Err(error), + None => Ok(()), + }; + } + std::thread::yield_now(); + } +} + +#[cfg(any(target_os = "android", target_os = "linux", target_os = "macos"))] +fn open_pty(size: PtySize) -> io::Result<(OwnedFd, OwnedFd)> { + let master = posix_openpt(OFlag::O_RDWR | OFlag::O_NOCTTY | OFlag::O_CLOEXEC)?; + verify_close_on_exec(&master)?; + set_nonblocking(&master)?; + grantpt(&master)?; + unlockpt(&master)?; + let slave = open_slave(&master)?; + verify_close_on_exec(&slave)?; + // SAFETY: ownership moves from PtyMaster into exactly one OwnedFd. + let master = unsafe { OwnedFd::from_raw_fd(master.into_raw_fd()) }; + set_size(&master, size)?; + Ok((master, slave)) +} + +#[cfg(any(target_os = "illumos", target_os = "solaris"))] +fn open_pty(size: PtySize) -> io::Result<(OwnedFd, OwnedFd)> { + let master = open( + Path::new("/dev/ptmx"), + OFlag::O_RDWR | OFlag::O_NOCTTY | OFlag::O_CLOEXEC, + Mode::empty(), + )?; + verify_close_on_exec(&master)?; + set_nonblocking(&master)?; + // SAFETY: the descriptor is an open PTY master and remains owned for both calls. + if unsafe { libc::grantpt(master.as_raw_fd()) } == -1 + || unsafe { libc::unlockpt(master.as_raw_fd()) } == -1 + { + return Err(io::Error::last_os_error()); + } + let slave = open_solarish_slave(&master)?; + verify_close_on_exec(&slave)?; + set_size(&master, size)?; + Ok((master, slave)) +} + +#[cfg(any(target_os = "dragonfly", target_os = "freebsd", target_os = "netbsd"))] +fn open_pty(size: PtySize) -> io::Result<(OwnedFd, OwnedFd)> { + let flags = OFlag::O_RDWR | OFlag::O_NOCTTY | OFlag::O_CLOEXEC; + let master = match posix_openpt(flags) { + Ok(master) => master, + #[cfg(target_os = "netbsd")] + Err(nix::errno::Errno::EINVAL) => { + return Err(io::Error::new( + io::ErrorKind::Unsupported, + "atomic PTY descriptors require NetBSD 10 or newer", + )); + } + Err(error) => return Err(io::Error::from(error)), + }; + verify_close_on_exec(&master)?; + set_nonblocking(&master)?; + grantpt(&master)?; + unlockpt(&master)?; + let slave = open_bsd_slave(&master)?; + verify_close_on_exec(&slave)?; + // SAFETY: ownership moves from PtyMaster into exactly one OwnedFd. + let master = unsafe { OwnedFd::from_raw_fd(master.into_raw_fd()) }; + set_size(&master, size)?; + Ok((master, slave)) +} + +#[cfg(target_os = "openbsd")] +fn open_pty(size: PtySize) -> io::Result<(OwnedFd, OwnedFd)> { + openbsd::open_pty(size) +} + +fn verify_close_on_exec(fd: &impl AsFd) -> io::Result<()> { + let flags = FdFlag::from_bits_truncate(fcntl(fd, FcntlArg::F_GETFD)?); + if !flags.contains(FdFlag::FD_CLOEXEC) { + return Err(io::Error::other( + "the PTY descriptor was not created close-on-exec", + )); + } + Ok(()) +} + +fn set_nonblocking(fd: &impl AsFd) -> io::Result<()> { + let flags = OFlag::from_bits_truncate(fcntl(fd, FcntlArg::F_GETFL)?); + fcntl(fd, FcntlArg::F_SETFL(flags | OFlag::O_NONBLOCK))?; + Ok(()) +} + +fn duplicate(fd: &OwnedFd) -> io::Result { + let duplicated = fcntl(fd, FcntlArg::F_DUPFD_CLOEXEC(0))?; + // SAFETY: F_DUPFD_CLOEXEC returned a new owned descriptor on success. + Ok(unsafe { OwnedFd::from_raw_fd(duplicated) }) +} + +#[cfg(any( + target_os = "android", + target_os = "dragonfly", + target_os = "freebsd", + target_os = "illumos", + target_os = "linux", + target_os = "macos", + target_os = "netbsd", + target_os = "solaris" +))] +fn slave_flags() -> OFlag { + OFlag::O_RDWR | OFlag::O_NOCTTY | OFlag::O_CLOEXEC +} + +#[cfg(any(target_os = "android", target_os = "linux"))] +fn open_slave(master: &PtyMaster) -> io::Result { + let name = ptsname_r(master)?; + open(Path::new(&name), slave_flags(), Mode::empty()).map_err(io::Error::from) +} + +#[cfg(target_os = "macos")] +fn open_slave(master: &PtyMaster) -> io::Result { + let mut name = [0_u8; 128]; + // SAFETY: name is the 128-byte output buffer encoded by Darwin's TIOCPTYGNAME request. + if unsafe { + libc::ioctl( + master.as_raw_fd(), + libc::TIOCPTYGNAME.into(), + name.as_mut_ptr(), + ) + } == -1 + { + return Err(io::Error::last_os_error()); + } + let name = CStr::from_bytes_until_nul(&name) + .map_err(|error| io::Error::new(io::ErrorKind::InvalidData, error))?; + open(name, slave_flags(), Mode::empty()).map_err(io::Error::from) +} + +#[cfg(any(target_os = "freebsd", target_os = "netbsd"))] +fn open_bsd_slave(master: &PtyMaster) -> io::Result { + let mut name = [0 as libc::c_char; 1024]; + // SAFETY: name is writable for its full reported length and master is an open PTY descriptor. + let result = unsafe { libc::ptsname_r(master.as_raw_fd(), name.as_mut_ptr(), name.len()) }; + if result != 0 { + return Err(if result > 0 { + io::Error::from_raw_os_error(result) + } else { + io::Error::last_os_error() + }); + } + // SAFETY: ptsname_r succeeded and therefore wrote a NUL-terminated path into name. + let name = unsafe { CStr::from_ptr(name.as_ptr()) }; + open(name, slave_flags(), Mode::empty()).map_err(io::Error::from) +} + +#[cfg(target_os = "dragonfly")] +fn open_bsd_slave(master: &PtyMaster) -> io::Result { + // DragonFly's ptsname storage is thread-local. Copy it before making another libc call. + // SAFETY: master is an open, granted and unlocked PTY descriptor. + let name = unsafe { libc::ptsname(master.as_raw_fd()) }; + if name.is_null() { + return Err(io::Error::last_os_error()); + } + // SAFETY: a non-null result from ptsname points to a NUL-terminated path. + let name = CString::from(unsafe { CStr::from_ptr(name) }); + open(name.as_c_str(), slave_flags(), Mode::empty()).map_err(io::Error::from) +} + +#[cfg(any(target_os = "illumos", target_os = "solaris"))] +fn open_solarish_slave(master: &OwnedFd) -> io::Result { + let stat = fstat(master)?; + // SAFETY: st_rdev came from fstat on the open PTY master. + let device = unsafe { libc::minor(stat.st_rdev) }; + let name = CString::new(format!("/dev/pts/{device}")) + .map_err(|error| io::Error::new(io::ErrorKind::InvalidData, error))?; + let slave = open(name.as_c_str(), slave_flags(), Mode::empty())?; + setup_solarish_streams(&slave)?; + Ok(slave) +} + +#[cfg(any(target_os = "illumos", target_os = "solaris"))] +fn setup_solarish_streams(slave: &OwnedFd) -> io::Result<()> { + let ldterm = c"ldterm"; + // SAFETY: the descriptor is an open PTY slave and the module names are static C strings. + let present = unsafe { libc::ioctl(slave.as_raw_fd(), libc::I_FIND, ldterm.as_ptr()) }; + if present == -1 { + return Err(io::Error::last_os_error()); + } + if present != 0 { + return Ok(()); + } + + // __I_PUSH_NOCTTY is the Solarish variant of I_PUSH which deliberately skips controlling-terminal + // acquisition after ptem marks the stream as a terminal. This matters when the parent is a session + // leader without an existing controlling terminal. + // SAFETY: the descriptor is an open PTY slave and each argument is a static C string. + if unsafe { libc::ioctl(slave.as_raw_fd(), libc::__I_PUSH_NOCTTY, c"ptem".as_ptr()) } == -1 + || unsafe { libc::ioctl(slave.as_raw_fd(), libc::__I_PUSH_NOCTTY, ldterm.as_ptr()) } == -1 + { + return Err(io::Error::last_os_error()); + } + #[cfg(target_os = "solaris")] + // SAFETY: the descriptor is an open PTY slave and the argument is a static C string. + if unsafe { + libc::ioctl( + slave.as_raw_fd(), + libc::__I_PUSH_NOCTTY, + c"ttcompat".as_ptr(), + ) + } == -1 + { + return Err(io::Error::last_os_error()); + } + Ok(()) +} + +fn winsize(size: PtySize) -> Winsize { + Winsize { + ws_row: size.rows, + ws_col: size.columns, + ws_xpixel: size.pixel_width, + ws_ypixel: size.pixel_height, + } +} + +fn set_size(master: &OwnedFd, size: PtySize) -> io::Result<()> { + let size = winsize(size); + // SAFETY: master is a live PTY descriptor and size points to a valid winsize for the duration of + // the ioctl call. + if unsafe { libc::ioctl(master.as_raw_fd(), libc::TIOCSWINSZ, &size) } == -1 { + return Err(io::Error::last_os_error()); + } + Ok(()) +} + +fn setup_child(reset_sigmask: bool) -> io::Result<()> { + if reset_sigmask { + crate::unix::reset_sigmask()?; + } + // SAFETY: this function runs after fork and before exec. Each call is async-signal-safe and uses + // only the already-installed standard input descriptor. + unsafe { + if libc::setsid() == -1 { + return Err(io::Error::last_os_error()); + } + if libc::ioctl(libc::STDIN_FILENO, libc::TIOCSCTTY as _, 0) == -1 { + return Err(io::Error::last_os_error()); + } + if libc::tcsetpgrp(libc::STDIN_FILENO, libc::getpgrp()) == -1 { + return Err(io::Error::last_os_error()); + } + } + Ok(()) +} + +fn read(fd: &OwnedFd, buffer: &mut [u8]) -> io::Result { + // SAFETY: the buffer is writable for its full length and remains live for the call. + let read = unsafe { libc::read(fd.as_raw_fd(), buffer.as_mut_ptr().cast(), buffer.len()) }; + if read == -1 { + return Err(io::Error::last_os_error()); + } + Ok(read as usize) +} + +fn write(fd: &OwnedFd, buffer: &[u8]) -> io::Result { + // SAFETY: the buffer is readable for its full length and remains live for the call. + let written = unsafe { libc::write(fd.as_raw_fd(), buffer.as_ptr().cast(), buffer.len()) }; + if written == -1 { + return Err(io::Error::last_os_error()); + } + Ok(written as usize) +} + +fn is_eof(error: &io::Error) -> bool { + error.raw_os_error() == Some(libc::EIO) +} + +fn closed() -> io::Error { + io::Error::new(io::ErrorKind::BrokenPipe, "the PTY master is closed") +} + +#[cfg(test)] +mod tests { + use std::os::fd::RawFd; + + use nix::unistd::pipe; + + use super::*; + + fn pipe_pair() -> (OwnedFd, OwnedFd) { + let (reader, writer) = pipe().unwrap(); + set_nonblocking(&reader).unwrap(); + (reader, writer) + } + + fn assert_open(fd: RawFd) { + // SAFETY: the caller retains ownership of a descriptor which must still be open here. + assert_ne!(unsafe { libc::fcntl(fd, libc::F_GETFD) }, -1); + } + + fn assert_closed(reader: &OwnedFd) { + let mut byte = 0_u8; + // SAFETY: reader is a live nonblocking pipe descriptor and byte is writable for one byte. + let read = + unsafe { libc::read(reader.as_raw_fd(), std::ptr::from_mut(&mut byte).cast(), 1) }; + if read != 0 { + panic!( + "pipe reader did not observe writer closure: {}", + io::Error::last_os_error() + ); + } + } + + fn descriptors() -> (OwnedFd, OwnedFd, OwnedFd, [OwnedFd; 3], [RawFd; 3]) { + let (stdin_reader, stdin) = pipe_pair(); + let (stdout_reader, stdout) = pipe_pair(); + let (stderr_reader, stderr) = pipe_pair(); + let raw = [stdin.as_raw_fd(), stdout.as_raw_fd(), stderr.as_raw_fd()]; + ( + stdin, + stdout, + stderr, + [stdin_reader, stdout_reader, stderr_reader], + raw, + ) + } + + #[test] + fn winsize_preserves_character_and_pixel_dimensions() { + let native = winsize(PtySize { + rows: 31, + columns: 97, + pixel_width: 640, + pixel_height: 480, + }); + assert_eq!(native.ws_row, 31); + assert_eq!(native.ws_col, 97); + assert_eq!(native.ws_xpixel, 640); + assert_eq!(native.ws_ypixel, 480); + } + + #[test] + fn allocated_master_has_close_on_exec_and_nonblocking() { + let (master, _slave) = open_pty(PtySize::default()).unwrap(); + let descriptor_flags = FdFlag::from_bits_truncate( + fcntl(&master, FcntlArg::F_GETFD).expect("read descriptor flags"), + ); + let status_flags = OFlag::from_bits_truncate( + fcntl(&master, FcntlArg::F_GETFL).expect("read status flags"), + ); + assert!(descriptor_flags.contains(FdFlag::FD_CLOEXEC)); + assert!(status_flags.contains(OFlag::O_NONBLOCK)); + } + + #[test] + fn slave_stdio_is_dropped_when_the_spawn_operation_returns() { + let mut command = tokio::process::Command::new("ignored"); + let (stdin, stdout, stderr, readers, raw) = descriptors(); + + let result = with_slave_stdio(&mut command, stdin, stdout, stderr, |_| { + raw.into_iter().for_each(assert_open); + 42 + }); + + assert_eq!(result, 42); + readers.iter().for_each(assert_closed); + } + + #[test] + fn slave_stdio_is_dropped_when_the_spawn_operation_panics() { + let mut command = tokio::process::Command::new("ignored"); + let (stdin, stdout, stderr, readers, raw) = descriptors(); + + let panic = catch_unwind(AssertUnwindSafe(|| { + with_slave_stdio(&mut command, stdin, stdout, stderr, |_| { + raw.into_iter().for_each(assert_open); + panic!("spawn operation panic"); + }); + })); + + assert!(panic.is_err()); + readers.iter().for_each(assert_closed); + } +} diff --git a/src/tokio/pty/unix/descriptor_pair.rs b/src/tokio/pty/unix/descriptor_pair.rs new file mode 100644 index 0000000..4b557db --- /dev/null +++ b/src/tokio/pty/unix/descriptor_pair.rs @@ -0,0 +1,345 @@ +use std::{ + io, + mem::{MaybeUninit, size_of}, + os::fd::{AsRawFd, FromRawFd, OwnedFd, RawFd}, +}; + +use nix::libc; + +const DESCRIPTOR_COUNT: usize = 2; +const DESCRIPTOR_BYTES: usize = size_of::<[RawFd; DESCRIPTOR_COUNT]>(); +// SAFETY: the requested payload length is the size of exactly two RawFd values. +const CONTROL_BYTES: usize = unsafe { libc::CMSG_SPACE(DESCRIPTOR_BYTES as libc::c_uint) as usize }; + +#[repr(C, align(16))] +struct ControlBuffer([u8; CONTROL_BYTES]); + +impl ControlBuffer { + fn zeroed() -> Self { + Self([0; CONTROL_BYTES]) + } +} + +#[repr(C)] +#[derive(Clone, Copy)] +struct Response { + error: libc::c_int, +} + +pub(super) fn socket_pair() -> io::Result<(OwnedFd, OwnedFd)> { + let mut sockets = [-1; 2]; + // SAFETY: sockets is writable for two descriptors. SOCK_CLOEXEC makes descriptor installation + // atomic with respect to a concurrent fork and exec. + if unsafe { + libc::socketpair( + libc::AF_UNIX, + libc::SOCK_SEQPACKET | libc::SOCK_CLOEXEC, + 0, + sockets.as_mut_ptr(), + ) + } == -1 + { + return Err(io::Error::last_os_error()); + } + + // SAFETY: socketpair initialized two independently owned descriptors on success. + Ok(unsafe { + ( + OwnedFd::from_raw_fd(sockets[0]), + OwnedFd::from_raw_fd(sockets[1]), + ) + }) +} + +/// Sends either an operating-system error or exactly two descriptors. +/// +/// # Safety +/// +/// `socket` and every descriptor in `descriptors` must be live for the duration of this call. This +/// function is suitable for the restricted child side of a post-fork helper: it uses only stack data, +/// pointer operations, `sendmsg`, and the thread-local errno slot. +pub(super) unsafe fn send_response( + socket: RawFd, + error: libc::c_int, + descriptors: Option<[RawFd; DESCRIPTOR_COUNT]>, +) -> bool { + let mut response = Response { error }; + let mut io_vector = libc::iovec { + iov_base: std::ptr::from_mut(&mut response).cast(), + iov_len: size_of::(), + }; + // SAFETY: an all-zero msghdr represents no address, vectors, or ancillary data. The fields used by + // sendmsg are initialized below. + let mut message = unsafe { MaybeUninit::::zeroed().assume_init() }; + message.msg_iov = std::ptr::from_mut(&mut io_vector); + message.msg_iovlen = 1 as _; + + let mut control = ControlBuffer::zeroed(); + if let Some(descriptors) = descriptors { + message.msg_control = control.0.as_mut_ptr().cast(); + message.msg_controllen = control.0.len() as _; + + // SAFETY: the control buffer is aligned for cmsghdr, has CMSG_SPACE for two descriptors, and the + // message points to the complete buffer. + let header = unsafe { libc::CMSG_FIRSTHDR(&message) }; + if header.is_null() { + return false; + } + // SAFETY: header points into the live, sufficiently sized control buffer. + unsafe { + (*header).cmsg_len = libc::CMSG_LEN(DESCRIPTOR_BYTES as libc::c_uint) as _; + (*header).cmsg_level = libc::SOL_SOCKET; + (*header).cmsg_type = libc::SCM_RIGHTS; + std::ptr::copy_nonoverlapping( + descriptors.as_ptr(), + libc::CMSG_DATA(header).cast::(), + DESCRIPTOR_COUNT, + ); + } + } + + loop { + // SAFETY: message points to live response, vector, and optional control-buffer storage. + let sent = unsafe { libc::sendmsg(socket, &message, 0) }; + if sent == size_of::() as libc::ssize_t { + return true; + } + if sent != -1 { + return false; + } + // SAFETY: errno returns the calling thread's live errno value. + if unsafe { errno() } != libc::EINTR { + return false; + } + } +} + +pub(super) fn receive_response(socket: &OwnedFd) -> io::Result<[OwnedFd; DESCRIPTOR_COUNT]> { + let mut response = MaybeUninit::::uninit(); + let mut io_vector = libc::iovec { + iov_base: response.as_mut_ptr().cast(), + iov_len: size_of::(), + }; + let mut control = ControlBuffer::zeroed(); + // SAFETY: an all-zero msghdr represents no address, vectors, or ancillary data. All receive storage + // fields are initialized below. + let mut message = unsafe { MaybeUninit::::zeroed().assume_init() }; + message.msg_iov = std::ptr::from_mut(&mut io_vector); + message.msg_iovlen = 1 as _; + message.msg_control = control.0.as_mut_ptr().cast(); + message.msg_controllen = control.0.len() as _; + + let received = loop { + // SAFETY: message points to writable response, vector, and control-buffer storage. The flag makes + // every received descriptor close-on-exec as part of descriptor installation. + let received = + unsafe { libc::recvmsg(socket.as_raw_fd(), &mut message, libc::MSG_CMSG_CLOEXEC) }; + if received != -1 { + break received; + } + let error = io::Error::last_os_error(); + if error.kind() != io::ErrorKind::Interrupted { + return Err(error); + } + }; + + let (mut rights, control_is_valid) = ReceivedRights::decode(&message); + if received != size_of::() as libc::ssize_t + || message.msg_flags & (libc::MSG_CTRUNC | libc::MSG_TRUNC) != 0 + || !control_is_valid + { + return Err(protocol_error( + "the PTY allocation helper returned a malformed response", + )); + } + + // SAFETY: recvmsg initialized the complete response because its exact size was returned above. + let response = unsafe { response.assume_init() }; + if response.error != 0 { + if !rights.is_empty() || response.error < 0 { + return Err(protocol_error( + "the PTY allocation helper returned an invalid error", + )); + } + return Err(io::Error::from_raw_os_error(response.error)); + } + + rights.take_pair() +} + +#[derive(Debug)] +struct ReceivedRights { + descriptors: [RawFd; DESCRIPTOR_COUNT], + count: usize, +} + +impl ReceivedRights { + fn empty() -> Self { + Self { + descriptors: [-1; DESCRIPTOR_COUNT], + count: 0, + } + } + + fn decode(message: &libc::msghdr) -> (Self, bool) { + let mut rights = Self::empty(); + if message.msg_controllen == 0 { + return (rights, true); + } + + // SAFETY: message's control pointer and length still refer to the live receive buffer. + let header = unsafe { libc::CMSG_FIRSTHDR(message) }; + if header.is_null() { + return (rights, false); + } + + // SAFETY: zero is a valid ancillary payload length. + let header_bytes = unsafe { libc::CMSG_LEN(0) as usize }; + // OpenBSD defines msg_controllen as socklen_t while Linux uses usize. + #[allow(clippy::unnecessary_cast)] + let available = message.msg_controllen as usize; + // SAFETY: CMSG_FIRSTHDR returned a header within the receive buffer. + let length = unsafe { (*header).cmsg_len as usize }; + if length < header_bytes || length > available { + return (rights, false); + } + + // SAFETY: the complete cmsghdr lies within the receive buffer after the bounds check above. + let is_rights = unsafe { + (*header).cmsg_level == libc::SOL_SOCKET && (*header).cmsg_type == libc::SCM_RIGHTS + }; + if !is_rights { + return (rights, false); + } + + let descriptor_bytes = length - header_bytes; + let count = descriptor_bytes / size_of::(); + if descriptor_bytes % size_of::() != 0 || count > DESCRIPTOR_COUNT { + return (rights, false); + } + + // SAFETY: cmsg_len covers count complete descriptors and the destination has room for both. + unsafe { + std::ptr::copy_nonoverlapping( + libc::CMSG_DATA(header).cast::(), + rights.descriptors.as_mut_ptr(), + count, + ); + } + rights.count = count; + // SAFETY: count is bounded to the two-descriptor control buffer above. + let expected_length = + unsafe { libc::CMSG_LEN((count * size_of::()) as libc::c_uint) as usize }; + // SAFETY: count is bounded to the two-descriptor control buffer above. + let expected_space = + unsafe { libc::CMSG_SPACE((count * size_of::()) as libc::c_uint) as usize }; + let valid = length == expected_length + && available >= expected_length + && available <= expected_space + && rights.descriptors[..count] + .iter() + .all(|descriptor| *descriptor >= 0) + && (count != 2 || rights.descriptors[0] != rights.descriptors[1]); + (rights, valid) + } + + fn is_empty(&self) -> bool { + self.count == 0 + } + + fn take_pair(&mut self) -> io::Result<[OwnedFd; DESCRIPTOR_COUNT]> { + if self.count != DESCRIPTOR_COUNT { + return Err(protocol_error( + "the PTY allocation helper did not return two descriptors", + )); + } + let descriptors = self.descriptors; + self.count = 0; + // SAFETY: SCM_RIGHTS installed two distinct owned descriptors, and this guard has relinquished + // responsibility for closing them. + Ok(unsafe { + [ + OwnedFd::from_raw_fd(descriptors[0]), + OwnedFd::from_raw_fd(descriptors[1]), + ] + }) + } +} + +impl Drop for ReceivedRights { + fn drop(&mut self) { + for descriptor in &self.descriptors[..self.count] { + // SAFETY: each descriptor was installed by SCM_RIGHTS and remains owned by this guard. + unsafe { + libc::close(*descriptor); + } + } + } +} + +#[cfg(target_os = "openbsd")] +unsafe fn errno() -> libc::c_int { + // SAFETY: __errno returns the calling thread's live errno slot. + unsafe { *libc::__errno() } +} + +#[cfg(all(test, target_os = "linux"))] +unsafe fn errno() -> libc::c_int { + // SAFETY: __errno_location returns the calling thread's live errno slot. + unsafe { *libc::__errno_location() } +} + +fn protocol_error(message: &'static str) -> io::Error { + io::Error::new(io::ErrorKind::InvalidData, message) +} + +#[cfg(test)] +mod tests { + use std::{ + fs::File, + io::{Read, Write}, + os::{fd::AsRawFd, unix::net::UnixStream}, + }; + + use nix::fcntl::{FcntlArg, FdFlag, fcntl}; + + use super::*; + + #[test] + fn received_descriptors_are_installed_close_on_exec() { + let (sender, receiver) = socket_pair().unwrap(); + let (first, second) = UnixStream::pair().unwrap(); + // SAFETY: the socket and both transferred descriptors remain live through the call. + assert!(unsafe { + send_response( + sender.as_raw_fd(), + 0, + Some([first.as_raw_fd(), second.as_raw_fd()]), + ) + }); + let [first_received, second_received] = receive_response(&receiver).unwrap(); + for descriptor in [&first_received, &second_received] { + let flags = FdFlag::from_bits_truncate(fcntl(descriptor, FcntlArg::F_GETFD).unwrap()); + assert!(flags.contains(FdFlag::FD_CLOEXEC)); + } + + drop((first, second)); + let mut first_received = File::from(first_received); + let mut second_received = File::from(second_received); + first_received.write_all(b"x").unwrap(); + let mut byte = [0]; + second_received.read_exact(&mut byte).unwrap(); + assert_eq!(byte, *b"x"); + } + + #[test] + fn helper_errors_are_returned_without_descriptors() { + let (sender, receiver) = socket_pair().unwrap(); + // SAFETY: the socket remains live through the call and no descriptors are transferred. + assert!(unsafe { send_response(sender.as_raw_fd(), libc::ENXIO, None) }); + assert_eq!( + receive_response(&receiver).unwrap_err().raw_os_error(), + Some(libc::ENXIO) + ); + } +} diff --git a/src/tokio/pty/unix/openbsd.rs b/src/tokio/pty/unix/openbsd.rs new file mode 100644 index 0000000..3c3f8ce --- /dev/null +++ b/src/tokio/pty/unix/openbsd.rs @@ -0,0 +1,170 @@ +use std::{ + io, + mem::{MaybeUninit, size_of}, + os::fd::{AsRawFd, OwnedFd, RawFd}, +}; + +use nix::libc; + +use super::{PtySize, descriptor_pair, set_nonblocking, set_size, verify_close_on_exec}; + +const PTM_DEVICE: &[u8] = b"/dev/ptm\0"; +const IOC_OUT: libc::c_ulong = 0x4000_0000; +const IOCPARM_MASK: usize = 0x1fff; + +#[repr(C)] +struct PtmGet { + controller: RawFd, + slave: RawFd, + controller_name: [libc::c_char; 16], + slave_name: [libc::c_char; 16], +} + +const PTMGET: libc::c_ulong = IOC_OUT + | ((size_of::() & IOCPARM_MASK) as libc::c_ulong) << 16 + | (b't' as libc::c_ulong) << 8 + | 1; +const _: () = assert!(size_of::() == 40); + +pub(super) fn open_pty(size: PtySize) -> io::Result<(OwnedFd, OwnedFd)> { + let (parent_socket, helper_socket) = descriptor_pair::socket_pair()?; + // SAFETY: fork has no Rust-side invariants beyond separating execution by its return value. The child + // immediately enters a syscall-only helper and exits without touching shared runtime state. + let helper = unsafe { libc::fork() }; + if helper == -1 { + return Err(io::Error::last_os_error()); + } + if helper == 0 { + // SAFETY: this is the post-fork helper. It uses only stack state and descriptor syscalls before + // terminating through _exit, and never returns into the Rust runtime. + unsafe { allocate_and_send(parent_socket.as_raw_fd(), helper_socket.as_raw_fd()) } + } + + drop(helper_socket); + let response = descriptor_pair::receive_response(&parent_socket); + let helper_result = reap_helper(helper); + let [master, slave] = match response { + Ok(descriptors) => { + helper_result?; + descriptors + } + Err(error) => { + let _ = helper_result; + return Err(error); + } + }; + + // MSG_CMSG_CLOEXEC installed both descriptors atomically. Verify that invariant before making the + // master nonblocking and configuring the terminal dimensions. + verify_close_on_exec(&master)?; + verify_close_on_exec(&slave)?; + set_nonblocking(&master)?; + set_size(&master, size)?; + Ok((master, slave)) +} + +unsafe fn allocate_and_send(parent_socket: RawFd, helper_socket: RawFd) -> ! { + // SAFETY: both descriptors were inherited across fork and the helper does not use the parent end. + unsafe { + libc::close(parent_socket); + } + + // SAFETY: PTM_DEVICE is a static NUL-terminated path. + let ptm = unsafe { libc::open(PTM_DEVICE.as_ptr().cast(), libc::O_RDWR | libc::O_CLOEXEC) }; + if ptm == -1 { + // SAFETY: __errno returns the helper thread's live errno slot. + let error = unsafe { *libc::__errno() }; + // SAFETY: helper_socket remains live and no descriptors accompany this error response. + let sent = unsafe { descriptor_pair::send_response(helper_socket, error, None) }; + // SAFETY: the helper must not run Rust destructors or shared runtime teardown after fork. + unsafe { libc::_exit(if sent { 0 } else { 1 }) } + } + + let mut pair = MaybeUninit::::zeroed(); + // SAFETY: ptm is an open /dev/ptm descriptor, PTMGET encodes the exact PtmGet layout, and the output + // pointer is writable for that complete structure. + let allocated = unsafe { libc::ioctl(ptm, PTMGET, pair.as_mut_ptr()) }; + // Capture errno before close can change it. + let error = if allocated == -1 { + // SAFETY: __errno returns the helper thread's live errno slot. + unsafe { *libc::__errno() } + } else { + 0 + }; + // SAFETY: ptm is the helper's live, independently owned descriptor. + unsafe { + libc::close(ptm); + } + if allocated == -1 { + // SAFETY: helper_socket remains live and no descriptors accompany this error response. + let sent = unsafe { descriptor_pair::send_response(helper_socket, error, None) }; + // SAFETY: the helper must not run Rust destructors or shared runtime teardown after fork. + unsafe { libc::_exit(if sent { 0 } else { 1 }) } + } + + // SAFETY: PTMGET succeeded and initialized the complete structure. + let pair = unsafe { pair.assume_init() }; + if pair.controller < 0 || pair.slave < 0 || pair.controller == pair.slave { + if pair.controller >= 0 { + // SAFETY: a nonnegative controller came from successful PTMGET. + unsafe { + libc::close(pair.controller); + } + } + if pair.slave >= 0 && pair.slave != pair.controller { + // SAFETY: a distinct nonnegative slave came from successful PTMGET. + unsafe { + libc::close(pair.slave); + } + } + // SAFETY: helper_socket remains live and no descriptors accompany this error response. + let sent = unsafe { descriptor_pair::send_response(helper_socket, libc::EIO, None) }; + // SAFETY: the helper must not run Rust destructors or shared runtime teardown after fork. + unsafe { libc::_exit(if sent { 0 } else { 1 }) } + } + + // PTMGET installs both descriptors without close-on-exec. They exist only in this no-exec helper; + // SCM_RIGHTS transfers copies which recvmsg installs atomically with close-on-exec in the parent. + // SAFETY: helper_socket and both PTY descriptors remain live through this call. + let sent = unsafe { + descriptor_pair::send_response(helper_socket, 0, Some([pair.controller, pair.slave])) + }; + // SAFETY: both descriptors are independently owned by the helper. The successful send retained its + // own references in the socket message until the parent receives them. + unsafe { + libc::close(pair.controller); + libc::close(pair.slave); + libc::_exit(if sent { 0 } else { 1 }) + } +} + +fn reap_helper(helper: libc::pid_t) -> io::Result<()> { + let mut status = 0; + loop { + // SAFETY: helper is the positive PID returned by fork and status is writable for one wait status. + let waited = unsafe { libc::waitpid(helper, &mut status, 0) }; + if waited == helper { + if libc::WIFEXITED(status) && libc::WEXITSTATUS(status) == 0 { + return Ok(()); + } + return Err(io::Error::other( + "the OpenBSD PTY allocation helper exited unsuccessfully", + )); + } + if waited == -1 { + let error = io::Error::last_os_error(); + if error.kind() == io::ErrorKind::Interrupted { + continue; + } + // An application may ignore SIGCHLD or use a process-wide child reaper. In either case ECHILD + // means the short-lived helper no longer needs to be reaped here. + if error.raw_os_error() == Some(libc::ECHILD) { + return Ok(()); + } + return Err(error); + } + return Err(io::Error::other( + "waitpid returned an unexpected OpenBSD PTY helper process", + )); + } +} diff --git a/src/tokio/pty/unsupported.rs b/src/tokio/pty/unsupported.rs new file mode 100644 index 0000000..eb53c01 --- /dev/null +++ b/src/tokio/pty/unsupported.rs @@ -0,0 +1,67 @@ +use std::{ + io, + pin::Pin, + task::{Context, Poll}, +}; + +use tokio::io::{AsyncRead, AsyncWrite, ReadBuf}; + +use super::{ProviderProduct, PtySize, SpawnAttempt}; + +#[derive(Debug)] +pub(super) struct Input; + +impl AsyncWrite for Input { + fn poll_write( + self: Pin<&mut Self>, + _cx: &mut Context<'_>, + _buffer: &[u8], + ) -> Poll> { + Poll::Ready(Err(unsupported())) + } + + fn poll_flush(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll> { + Poll::Ready(Err(unsupported())) + } + + fn poll_shutdown(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll> { + Poll::Ready(Err(unsupported())) + } +} + +#[derive(Debug)] +pub(super) struct Output; + +impl AsyncRead for Output { + fn poll_read( + self: Pin<&mut Self>, + _cx: &mut Context<'_>, + _buffer: &mut ReadBuf<'_>, + ) -> Poll> { + Poll::Ready(Err(unsupported())) + } +} + +#[derive(Clone, Debug)] +pub(super) struct Resize; + +impl Resize { + pub(super) fn resize(&self, _size: PtySize) -> io::Result<()> { + Err(unsupported()) + } +} + +pub(super) fn check_available() -> io::Result<()> { + Err(unsupported()) +} + +pub(super) fn spawn(_attempt: &mut SpawnAttempt, _size: PtySize) -> io::Result { + Err(unsupported()) +} + +fn unsupported() -> io::Error { + io::Error::new( + io::ErrorKind::Unsupported, + "pseudo-terminal spawning is unsupported on this platform", + ) +} diff --git a/src/unix.rs b/src/unix.rs index 4814ce5..b1456d8 100644 --- a/src/unix.rs +++ b/src/unix.rs @@ -13,6 +13,21 @@ use crate::command::NativeCommand; const NO_PROCESS_GROUP: i32 = -1; const LEADER_PROCESS_GROUP: i32 = 0; +pub(crate) fn reset_sigmask() -> io::Result<()> { + let mut empty = std::mem::MaybeUninit::::uninit(); + // SAFETY: `empty` points to writable storage for one signal set. + if unsafe { libc::sigemptyset(empty.as_mut_ptr()) } == -1 { + return Err(io::Error::last_os_error()); + } + // SAFETY: `sigemptyset` initialized `empty`; the old mask is not requested. + let error = + unsafe { libc::pthread_sigmask(libc::SIG_SETMASK, empty.as_ptr(), ptr::null_mut()) }; + if error != 0 { + return Err(io::Error::from_raw_os_error(error)); + } + Ok(()) +} + /// Process-group setup requested for one spawn attempt. #[cfg_attr(not(feature = "process-group"), allow(dead_code))] #[derive(Clone, Copy, Debug, Eq, PartialEq)] @@ -219,18 +234,7 @@ impl Dispatcher { let process_group = self.process_group.load(Ordering::SeqCst); if reset_sigmask { - let mut empty = std::mem::MaybeUninit::::uninit(); - // SAFETY: `empty` points to writable storage for one signal set. - if unsafe { libc::sigemptyset(empty.as_mut_ptr()) } == -1 { - return Err(io::Error::last_os_error()); - } - // SAFETY: `sigemptyset` initialized `empty`; the old mask is not requested. - let error = unsafe { - libc::pthread_sigmask(libc::SIG_SETMASK, empty.as_ptr(), ptr::null_mut()) - }; - if error != 0 { - return Err(io::Error::from_raw_os_error(error)); - } + crate::unix::reset_sigmask()?; } if process_session { diff --git a/tests/tokio_pty_provider.rs b/tests/tokio_pty_provider.rs new file mode 100644 index 0000000..2016eb4 --- /dev/null +++ b/tests/tokio_pty_provider.rs @@ -0,0 +1,770 @@ +#![cfg(all( + feature = "pty", + any( + target_os = "android", + target_os = "dragonfly", + target_os = "freebsd", + target_os = "illumos", + target_os = "linux", + target_os = "macos", + target_os = "netbsd", + target_os = "openbsd", + target_os = "solaris" + ) +))] + +#[cfg(target_os = "linux")] +use std::mem::MaybeUninit; +use std::{ + io, + panic::{AssertUnwindSafe, catch_unwind, panic_any}, + sync::{Arc, Mutex}, + time::{Duration, Instant}, +}; + +use nix::libc; +use process_wrap::tokio::{ + ChildWrapper, Command, CommandWrapper, ProviderProduct, Pty, PtyController, PtySize, + SpawnAttempt, SpawnProvider, +}; +use tokio::{io::AsyncReadExt, time::timeout}; + +fn controller(child: &mut Box) -> PtyController { + child + .take_pty_controller() + .expect("a successful PTY spawn installs one controller") +} + +async fn output(child: &mut dyn ChildWrapper, controller: PtyController) -> io::Result> { + let (input, mut output, _resize) = controller.into_parts(); + drop(input); + let mut bytes = Vec::new(); + timeout(Duration::from_secs(5), async { + let (status, _) = tokio::try_join!(child.wait(), output.read_to_end(&mut bytes))?; + assert!(status.success()); + Ok::<_, io::Error>(()) + }) + .await??; + Ok(bytes) +} + +#[derive(Debug)] +struct TransparentChild(Box); + +impl ChildWrapper for TransparentChild { + fn inner(&self) -> &dyn ChildWrapper { + self.0.as_ref() + } + + fn inner_mut(&mut self) -> &mut dyn ChildWrapper { + self.0.as_mut() + } + + fn into_inner(self: Box) -> Box { + self.0 + } +} + +#[derive(Clone, Copy, Debug)] +struct Transparent; + +impl CommandWrapper for Transparent { + fn wrap_child( + &mut self, + child: Box, + _command: &Command, + ) -> io::Result> { + Ok(Box::new(TransparentChild(child))) + } +} + +#[derive(Debug)] +struct SecondTransparentChild(Box); + +impl ChildWrapper for SecondTransparentChild { + fn inner(&self) -> &dyn ChildWrapper { + self.0.as_ref() + } + + fn inner_mut(&mut self) -> &mut dyn ChildWrapper { + self.0.as_mut() + } + + fn into_inner(self: Box) -> Box { + self.0 + } +} + +#[derive(Clone, Copy, Debug)] +struct SecondTransparent; + +impl CommandWrapper for SecondTransparent { + fn wrap_child( + &mut self, + child: Box, + _command: &Command, + ) -> io::Result> { + Ok(Box::new(SecondTransparentChild(child))) + } +} + +#[derive(Debug)] +struct TerminalChild; + +impl ChildWrapper for TerminalChild { + fn inner(&self) -> &dyn ChildWrapper { + self + } + + fn inner_mut(&mut self) -> &mut dyn ChildWrapper { + self + } + + fn into_inner(self: Box) -> Box { + self + } +} + +#[tokio::test] +async fn non_pty_children_have_no_controller() -> io::Result<()> { + let mut command = Command::with_new("sh", |command| { + command.args(["-c", "exit 0"]); + }); + let mut child = command.spawn()?; + assert!(child.take_pty_controller().is_none()); + assert!(child.wait().await?.success()); + + let mut child = Box::new(TerminalChild) as Box; + assert!(child.take_pty_controller().is_none()); + Ok(()) +} + +#[tokio::test] +async fn controller_traversal_preserves_outer_wrappers_in_either_order() -> io::Result<()> { + for pty_first in [false, true] { + let mut command = Command::with_new("sh", |command| { + command.args(["-c", "printf wrapped"]); + }); + if pty_first { + command + .wrap(Pty::default()) + .wrap(Transparent) + .wrap(SecondTransparent); + } else { + command + .wrap(Transparent) + .wrap(SecondTransparent) + .wrap(Pty::default()); + } + + let mut child = command.spawn()?; + assert!(child.stdin().is_none()); + assert!(child.stdout().is_none()); + assert!(child.stderr().is_none()); + let controller = controller(&mut child); + assert!(child.take_pty_controller().is_none()); + assert_eq!(output(child.as_mut(), controller).await?, b"wrapped"); + } + Ok(()) +} + +#[derive(Debug)] +struct InspectPost { + called: Arc>, +} + +impl CommandWrapper for InspectPost { + fn post_spawn( + &mut self, + attempt: &mut SpawnAttempt, + child: &mut dyn ChildWrapper, + _command: &Command, + ) -> io::Result<()> { + assert_eq!( + attempt + .get_portable_args() + .expect("the PTY provider keeps portable intent") + .len(), + 2 + ); + assert!(child.take_pty_controller().is_none()); + *self + .called + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner) = true; + Ok(()) + } +} + +#[tokio::test] +async fn controller_is_committed_after_post_spawn_hooks() -> io::Result<()> { + let called = Arc::new(Mutex::new(false)); + let mut command = Command::with_new("sh", |command| { + command.args(["-c", "printf committed"]); + }); + command.wrap(Pty::default()).wrap(InspectPost { + called: Arc::clone(&called), + }); + + let mut child = command.spawn()?; + assert!( + *called + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner) + ); + let controller = controller(&mut child); + assert_eq!(output(child.as_mut(), controller).await?, b"committed"); + Ok(()) +} + +#[derive(Debug)] +struct InspectWrap { + called: Arc>, +} + +impl CommandWrapper for InspectWrap { + fn wrap_child( + &mut self, + mut child: Box, + _command: &Command, + ) -> io::Result> { + assert!(child.take_pty_controller().is_none()); + *self + .called + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner) = true; + Ok(child) + } +} + +#[tokio::test] +async fn controller_is_committed_after_child_wrapping_hooks() -> io::Result<()> { + let called = Arc::new(Mutex::new(false)); + let mut command = Command::with_new("sh", |command| { + command.args(["-c", "printf wrapped-commit"]); + }); + command.wrap(Pty::default()).wrap(InspectWrap { + called: Arc::clone(&called), + }); + + let mut child = command.spawn()?; + assert!( + *called + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner) + ); + let controller = controller(&mut child); + assert_eq!(output(child.as_mut(), controller).await?, b"wrapped-commit"); + Ok(()) +} + +#[derive(Debug)] +struct SpawnPostAttempt; + +impl CommandWrapper for SpawnPostAttempt { + fn post_spawn( + &mut self, + attempt: &mut SpawnAttempt, + _child: &mut dyn ChildWrapper, + _command: &Command, + ) -> io::Result<()> { + let mut probe = attempt.native_mut().spawn()?; + loop { + if probe.try_wait()?.is_some() { + return Ok(()); + } + std::thread::yield_now(); + } + } +} + +#[tokio::test] +async fn provider_setup_is_absent_from_post_spawn_native_attempt() -> io::Result<()> { + let mut command = Command::with_new("sh", |command| { + command.args(["-c", "exit 0"]); + }); + command.wrap(Pty::default()).wrap(SpawnPostAttempt); + + let mut child = command.spawn()?; + let controller = controller(&mut child); + assert!(output(child.as_mut(), controller).await?.is_empty()); + Ok(()) +} + +#[tokio::test] +async fn duplicate_pty_registration_uses_the_later_size() -> io::Result<()> { + let mut command = Command::with_new("sh", |command| { + command.args(["-c", "stty size"]); + }); + command + .wrap(Pty::new(PtySize::new(31, 97)?)) + .wrap(Pty::new(PtySize::new(42, 113)?)); + + let mut child = command.spawn()?; + let controller = controller(&mut child); + assert_eq!(output(child.as_mut(), controller).await?, b"42 113\r\n"); + Ok(()) +} + +#[derive(Debug)] +struct OtherProvider; + +impl SpawnProvider for OtherProvider { + fn spawn( + &self, + _attempt: &mut SpawnAttempt, + _command: &Command, + ) -> io::Result { + panic!("provider conflicts are rejected before spawning") + } +} + +impl CommandWrapper for OtherProvider { + fn spawn_provider(&self) -> Option<&dyn SpawnProvider> { + Some(self) + } +} + +#[test] +fn rejects_another_spawn_provider_before_callbacks() { + let mut command = Command::new("ignored"); + command.wrap(Pty::default()).wrap(OtherProvider); + let error = command.spawn().unwrap_err(); + assert_eq!(error.kind(), io::ErrorKind::InvalidInput); + assert_eq!(error.to_string(), "multiple spawn providers are registered"); +} + +#[derive(Debug)] +struct MakeAttemptNative; + +impl CommandWrapper for MakeAttemptNative { + fn pre_spawn(&mut self, attempt: &mut SpawnAttempt, _command: &Command) -> io::Result<()> { + let _ = attempt.native_mut(); + Ok(()) + } +} + +#[test] +fn rejects_opaque_base_and_attempt_state() { + let native = tokio::process::Command::new("ignored"); + let mut command = Command::from(native); + command.wrap(Pty::default()); + let error = command.spawn().unwrap_err(); + assert_eq!(error.kind(), io::ErrorKind::InvalidInput); + assert_eq!( + error.to_string(), + "a spawn provider cannot use a native-only command" + ); + + let mut command = Command::new("ignored"); + command.wrap(Pty::default()).wrap(MakeAttemptNative); + let error = command.spawn().unwrap_err(); + assert_eq!(error.kind(), io::ErrorKind::InvalidInput); + assert_eq!( + error.to_string(), + "a spawn provider cannot use a native-only spawn attempt" + ); +} + +#[test] +fn explicit_spawners_cannot_bypass_pty() { + let mut command = Command::new("ignored"); + command.wrap(Pty::default()); + let error = command + .spawn_with(|_| panic!("explicit spawner must not run")) + .unwrap_err(); + assert_eq!(error.kind(), io::ErrorKind::InvalidInput); + + let error = command + .spawn_with_child(|_| panic!("explicit child spawner must not run")) + .unwrap_err(); + assert_eq!(error.kind(), io::ErrorKind::InvalidInput); +} + +#[derive(Clone, Copy, Debug)] +enum Stage { + PostSpawn, + WrapChild, +} + +#[derive(Clone, Copy, Debug)] +enum Failure { + Error, + Panic, +} + +#[derive(Debug)] +struct FailAfterSpawn { + stage: Stage, + failure: Failure, + pid: Arc>>, +} + +impl FailAfterSpawn { + fn record(&self, child: &dyn ChildWrapper) { + *self + .pid + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner) = child.id(); + } + + fn fail(&self) -> io::Result { + match self.failure { + Failure::Error => Err(io::Error::other("fail after PTY spawn")), + Failure::Panic => panic_any("fail after PTY spawn"), + } + } +} + +impl CommandWrapper for FailAfterSpawn { + fn post_spawn( + &mut self, + _attempt: &mut SpawnAttempt, + child: &mut dyn ChildWrapper, + _command: &Command, + ) -> io::Result<()> { + if matches!(self.stage, Stage::PostSpawn) { + self.record(child); + self.fail() + } else { + Ok(()) + } + } + + fn wrap_child( + &mut self, + child: Box, + _command: &Command, + ) -> io::Result> { + if matches!(self.stage, Stage::WrapChild) { + self.record(child.as_ref()); + self.fail() + } else { + Ok(child) + } + } +} + +fn assert_reaped(pid: u32) { + let pid = i32::try_from(pid).unwrap(); + let mut status = 0; + // SAFETY: status is writable, and this only queries whether the provider transaction already + // reaped the recorded direct child. + let waited = unsafe { libc::waitpid(pid, &mut status, libc::WNOHANG) }; + assert_eq!(waited, -1); + assert_eq!( + io::Error::last_os_error().raw_os_error(), + Some(libc::ECHILD) + ); +} + +#[derive(Debug)] +struct ReapThenFail { + pid: Arc>>, +} + +impl CommandWrapper for ReapThenFail { + fn post_spawn( + &mut self, + _attempt: &mut SpawnAttempt, + child: &mut dyn ChildWrapper, + _command: &Command, + ) -> io::Result<()> { + *self + .pid + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner) = child.id(); + loop { + if child.try_wait()?.is_some() { + return Err(io::Error::other("fail after reaping PTY child")); + } + std::thread::yield_now(); + } + } +} + +#[tokio::test] +async fn transaction_observes_a_child_reaped_by_a_hook() { + let pid = Arc::new(Mutex::new(None)); + let mut command = Command::with_new("sh", |command| { + command.args(["-c", "exit 0"]); + }); + command.wrap(Pty::default()).wrap(ReapThenFail { + pid: Arc::clone(&pid), + }); + + assert_eq!( + command.spawn().unwrap_err().to_string(), + "fail after reaping PTY child" + ); + let pid = pid + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner) + .expect("the hook records the direct child before reaping it"); + assert_reaped(pid); +} + +#[derive(Debug)] +struct FailAfterDescendantStarts { + pid_file: std::path::PathBuf, + direct_pid: Arc>>, + descendant_pid: Arc>>, +} + +impl CommandWrapper for FailAfterDescendantStarts { + fn post_spawn( + &mut self, + _attempt: &mut SpawnAttempt, + child: &mut dyn ChildWrapper, + _command: &Command, + ) -> io::Result<()> { + *self + .direct_pid + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner) = child.id(); + let deadline = Instant::now() + Duration::from_secs(5); + while !self.pid_file.exists() { + if Instant::now() >= deadline { + return Err(io::Error::new( + io::ErrorKind::TimedOut, + "PTY descendant did not report its PID", + )); + } + std::thread::sleep(Duration::from_millis(10)); + } + let pid = std::fs::read_to_string(&self.pid_file)? + .trim() + .parse() + .map_err(io::Error::other)?; + *self + .descendant_pid + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner) = Some(pid); + Err(io::Error::other("fail with a live PTY descendant")) + } +} + +#[cfg(target_os = "linux")] +#[derive(Debug)] +struct FailAfterLeaderExits { + pid_file: std::path::PathBuf, + direct_pid: Arc>>, + descendant_pid: Arc>>, +} + +#[cfg(target_os = "linux")] +impl CommandWrapper for FailAfterLeaderExits { + fn post_spawn( + &mut self, + _attempt: &mut SpawnAttempt, + child: &mut dyn ChildWrapper, + _command: &Command, + ) -> io::Result<()> { + let direct_pid = child + .id() + .expect("the PTY provider reports its unreaped direct child"); + *self + .direct_pid + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner) = Some(direct_pid); + + let deadline = Instant::now() + Duration::from_secs(5); + while !self.pid_file.exists() { + if Instant::now() >= deadline { + return Err(io::Error::new( + io::ErrorKind::TimedOut, + "PTY descendant did not report its PID", + )); + } + std::thread::sleep(Duration::from_millis(10)); + } + let descendant_pid = std::fs::read_to_string(&self.pid_file)? + .trim() + .parse() + .map_err(io::Error::other)?; + *self + .descendant_pid + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner) = Some(descendant_pid); + + loop { + let mut status = MaybeUninit::::zeroed(); + // SAFETY: direct_pid names this process's direct child, status is writable, and WNOWAIT + // observes an exit without releasing the PID which anchors the PTY process group. + if unsafe { + libc::waitid( + libc::P_PID, + direct_pid as libc::id_t, + status.as_mut_ptr(), + libc::WEXITED | libc::WNOHANG | libc::WNOWAIT, + ) + } == -1 + { + return Err(io::Error::last_os_error()); + } + // SAFETY: status was zero-initialized and waitid either filled it or left si_pid as zero. + if unsafe { status.assume_init().si_pid() } + == libc::pid_t::try_from(direct_pid).map_err(io::Error::other)? + { + break; + } + if Instant::now() >= deadline { + return Err(io::Error::new( + io::ErrorKind::TimedOut, + "PTY leader did not exit before hook failure", + )); + } + std::thread::sleep(Duration::from_millis(10)); + } + + Err(io::Error::other("fail after the PTY leader exits")) + } +} + +fn assert_process_disappears(pid: i32) { + let deadline = Instant::now() + Duration::from_secs(5); + loop { + // SAFETY: signal zero only queries whether a process with this PID exists. + if unsafe { libc::kill(pid, 0) } == -1 + && io::Error::last_os_error().raw_os_error() == Some(libc::ESRCH) + { + return; + } + assert!( + Instant::now() < deadline, + "PTY descendant {pid} survived transaction rollback" + ); + std::thread::sleep(Duration::from_millis(10)); + } +} + +#[tokio::test] +async fn transaction_kills_the_live_pty_group_after_hook_failure() -> io::Result<()> { + let directory = tempfile::tempdir()?; + let pid_file = directory.path().join("descendant-pid"); + let direct_pid = Arc::new(Mutex::new(None)); + let descendant_pid = Arc::new(Mutex::new(None)); + let mut command = Command::with_new("sh", |command| { + command + .args([ + "-c", + "trap '' HUP TERM; sleep 30 & printf '%s' $! > \"$1\"; wait", + "pty-test", + ]) + .arg(&pid_file); + }); + command + .wrap(Pty::default()) + .wrap(FailAfterDescendantStarts { + pid_file, + direct_pid: Arc::clone(&direct_pid), + descendant_pid: Arc::clone(&descendant_pid), + }); + + assert_eq!( + command.spawn().unwrap_err().to_string(), + "fail with a live PTY descendant" + ); + let direct_pid = direct_pid + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner) + .expect("the hook records the direct child"); + assert_reaped(direct_pid); + let descendant_pid = descendant_pid + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner) + .expect("the hook records the descendant child"); + assert_process_disappears(descendant_pid); + Ok(()) +} + +#[cfg(target_os = "linux")] +#[tokio::test] +async fn transaction_kills_the_pty_group_before_reaping_an_exited_leader() -> io::Result<()> { + let directory = tempfile::tempdir()?; + let pid_file = directory.path().join("descendant-pid"); + let direct_pid = Arc::new(Mutex::new(None)); + let descendant_pid = Arc::new(Mutex::new(None)); + let mut command = Command::with_new("sh", |command| { + command + .args([ + "-c", + "trap '' HUP TERM; sleep 30 & printf '%s' $! > \"$1\"; exit 0", + "pty-test", + ]) + .arg(&pid_file); + }); + command.wrap(Pty::default()).wrap(FailAfterLeaderExits { + pid_file, + direct_pid: Arc::clone(&direct_pid), + descendant_pid: Arc::clone(&descendant_pid), + }); + + assert_eq!( + command.spawn().unwrap_err().to_string(), + "fail after the PTY leader exits" + ); + let direct_pid = direct_pid + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner) + .expect("the hook records the direct child"); + assert_reaped(direct_pid); + let descendant_pid = descendant_pid + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner) + .expect("the hook records the descendant child"); + assert_process_disappears(descendant_pid); + Ok(()) +} + +#[tokio::test] +async fn transaction_reaps_children_after_hook_failures() { + for stage in [Stage::PostSpawn, Stage::WrapChild] { + for failure in [Failure::Error, Failure::Panic] { + let pid = Arc::new(Mutex::new(None)); + let mut command = Command::with_new("sh", |command| { + command.args(["-c", "trap '' HUP; sleep 30"]); + }); + command.wrap(Pty::default()).wrap(FailAfterSpawn { + stage, + failure, + pid: Arc::clone(&pid), + }); + + let result = catch_unwind(AssertUnwindSafe(|| command.spawn())); + match failure { + Failure::Error => assert_eq!( + result.unwrap().unwrap_err().to_string(), + "fail after PTY spawn" + ), + Failure::Panic => assert_eq!( + *result + .expect_err("the lifecycle hook must panic") + .downcast::<&'static str>() + .unwrap(), + "fail after PTY spawn" + ), + } + let pid = pid + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner) + .expect("the failing hook records the direct child"); + assert_reaped(pid); + } + } +} + +#[tokio::test] +async fn provider_and_controller_are_reusable() -> io::Result<()> { + let mut command = Command::with_new("sh", |command| { + command.args(["-c", "printf reused"]); + }); + command.wrap(Pty::default()); + + for _ in 0..3 { + let mut child = command.spawn()?; + let controller = controller(&mut child); + assert_eq!(output(child.as_mut(), controller).await?, b"reused"); + } + Ok(()) +} diff --git a/tests/tokio_pty_unix.rs b/tests/tokio_pty_unix.rs new file mode 100644 index 0000000..b9f65d6 --- /dev/null +++ b/tests/tokio_pty_unix.rs @@ -0,0 +1,739 @@ +#![cfg(all( + feature = "pty", + any( + target_os = "android", + target_os = "dragonfly", + target_os = "freebsd", + target_os = "illumos", + target_os = "linux", + target_os = "macos", + target_os = "netbsd", + target_os = "openbsd", + target_os = "solaris" + ) +))] + +#[cfg(feature = "process-group")] +use std::any::TypeId; +#[cfg(feature = "reset-sigmask")] +use std::os::unix::process::ExitStatusExt; +use std::{io, process::ExitStatus, time::Duration}; + +#[cfg(all(feature = "kill-on-drop", feature = "process-session"))] +use nix::{ + sys::signal::{Signal, kill}, + unistd::Pid, +}; + +#[cfg(all(feature = "kill-on-drop", feature = "process-session"))] +use process_wrap::tokio::KillOnDrop; +#[cfg(feature = "process-session")] +use process_wrap::tokio::ProcessSession; +#[cfg(feature = "reset-sigmask")] +use process_wrap::tokio::ResetSigmask; +use process_wrap::tokio::{ChildWrapper, Command, Pty, PtyController, PtyOutput, PtySize}; +#[cfg(any(feature = "process-group", feature = "process-session"))] +use process_wrap::tokio::{CommandWrapper, SpawnAttempt}; +#[cfg(feature = "process-group")] +use process_wrap::tokio::{ProcessGroup, ProcessGroupChild}; +#[cfg(all(feature = "kill-on-drop", feature = "process-session"))] +use tokio::io::{AsyncBufReadExt, BufReader}; +use tokio::{ + io::{AsyncReadExt, AsyncWriteExt}, + time::{sleep, timeout}, +}; + +async fn wait_and_drain( + child: &mut dyn ChildWrapper, + output: &mut PtyOutput, +) -> io::Result<(ExitStatus, Vec)> { + let mut bytes = Vec::new(); + let status = timeout(Duration::from_secs(5), async { + let (status, _) = tokio::try_join!(child.wait(), output.read_to_end(&mut bytes))?; + Ok::<_, io::Error>(status) + }) + .await??; + Ok((status, bytes)) +} + +fn spawn_with_terminal( + command: &mut Command, + size: PtySize, +) -> io::Result<(Box, PtyController)> { + command.wrap(Pty::new(size)); + let mut child = command.spawn()?; + let controller = child + .take_pty_controller() + .expect("a successful PTY spawn installs one controller"); + Ok((child, controller)) +} + +#[tokio::test] +async fn spawns_with_terminal_fds_and_merged_output() -> io::Result<()> { + let mut command = Command::new("sh"); + command.args([ + "-c", + "test -t 0 && printf stdin-tty; test -t 1 && printf stdout-tty; test -t 2 && printf stderr-tty >&2; printf stdout; printf stderr >&2", + ]); + + let (mut child, controller) = spawn_with_terminal(&mut command, PtySize::default())?; + let (input, mut output, _resize) = controller.into_parts(); + drop(input); + let (status, bytes) = wait_and_drain(child.as_mut(), &mut output).await?; + assert!(status.success()); + assert_eq!(bytes, b"stdin-ttystdout-ttystderr-ttystdoutstderr"); + Ok(()) +} + +#[tokio::test] +async fn passes_bidirectional_control_bytes_unchanged() -> io::Result<()> { + let mut command = Command::new("sh"); + command.args([ + "-c", + "stty raw -echo; printf ready; dd bs=1 count=4 2>/dev/null", + ]); + + let (mut child, controller) = spawn_with_terminal(&mut command, PtySize::default())?; + let (mut input, mut output, _resize) = controller.into_parts(); + let mut ready = [0; 5]; + timeout(Duration::from_secs(5), output.read_exact(&mut ready)).await??; + assert_eq!(&ready, b"ready"); + + let controls = [0x00, 0x1b, 0x03, 0xff]; + input.write_all(&controls).await?; + let mut returned = [0; 4]; + timeout(Duration::from_secs(5), output.read_exact(&mut returned)).await??; + assert_eq!(returned, controls); + input.shutdown().await?; + assert!( + timeout(Duration::from_secs(5), child.wait()) + .await?? + .success() + ); + Ok(()) +} + +#[tokio::test] +async fn terminal_veof_ends_canonical_input() -> io::Result<()> { + let mut command = Command::new("sh"); + command.args(["-c", "stty -echo; printf ready; cat; printf eof"]); + + let (mut child, controller) = spawn_with_terminal(&mut command, PtySize::default())?; + let (mut input, mut output, _resize) = controller.into_parts(); + let mut ready = [0; 5]; + timeout(Duration::from_secs(5), output.read_exact(&mut ready)).await??; + assert_eq!(&ready, b"ready"); + input.write_all(b"payload\n\x04").await?; + let (status, bytes) = wait_and_drain(child.as_mut(), &mut output).await?; + assert!(status.success()); + assert_eq!(bytes, b"payload\r\neof"); + Ok(()) +} + +#[tokio::test] +async fn preserves_tracked_command_intent() -> io::Result<()> { + let directory = tempfile::tempdir()?; + let expected_directory = directory.path().canonicalize()?; + let mut command = Command::new("/bin/sh"); + command + .args([ + "-c", + "printf '%s|%s|%s|%s|%s' \"$1\" \"$VALUE\" \"${BEFORE-unset}\" \"${REMOVE-unset}\" \"$PWD\"", + "pty-test", + "argument", + ]) + .env("BEFORE", "discarded") + .env_clear() + .env("VALUE", "environment") + .env("REMOVE", "discarded") + .env_remove("REMOVE") + .current_dir(directory.path()); + + let (mut child, controller) = spawn_with_terminal(&mut command, PtySize::default())?; + let (input, mut output, _resize) = controller.into_parts(); + drop(input); + let (status, bytes) = wait_and_drain(child.as_mut(), &mut output).await?; + assert!(status.success()); + assert_eq!( + String::from_utf8(bytes).unwrap(), + format!( + "argument|environment|unset|unset|{}", + expected_directory.display() + ) + ); + Ok(()) +} + +#[tokio::test] +async fn reports_initial_size_and_sigwinch_resize() -> io::Result<()> { + let initial = PtySize::new(31, 97)?.with_pixels(640, 480); + let mut command = Command::new("sh"); + command.args([ + "-c", + "stty -echo; stty size; trap 'stty size; exit 0' WINCH; printf ready; while :; do sleep 1; done", + ]); + + let (mut child, controller) = spawn_with_terminal(&mut command, initial)?; + let (input, mut output, resize) = controller.into_parts(); + let mut initial_output = [0; 12]; + timeout( + Duration::from_secs(5), + output.read_exact(&mut initial_output), + ) + .await??; + assert_eq!(&initial_output, b"31 97\r\nready"); + + assert_eq!( + resize + .resize(PtySize { + rows: 0, + columns: 113, + pixel_width: 0, + pixel_height: 0, + }) + .unwrap_err() + .kind(), + io::ErrorKind::InvalidInput + ); + resize.resize(PtySize::new(42, 113)?)?; + let mut resized = Vec::new(); + timeout(Duration::from_secs(5), output.read_to_end(&mut resized)).await??; + assert_eq!(resized, b"42 113\r\n"); + assert!( + timeout(Duration::from_secs(5), child.wait()) + .await?? + .success() + ); + drop(input); + Ok(()) +} + +#[tokio::test] +async fn shutting_down_input_does_not_hang_up_while_output_exists() -> io::Result<()> { + let mut command = Command::new("sh"); + command.args(["-c", "stty -echo; printf ready; IFS= read -r line"]); + + let (mut child, controller) = spawn_with_terminal(&mut command, PtySize::default())?; + let (mut input, mut output, resize) = controller.into_parts(); + let mut ready = [0; 5]; + timeout(Duration::from_secs(5), output.read_exact(&mut ready)).await??; + assert_eq!(&ready, b"ready"); + + input.shutdown().await?; + assert_eq!( + input.write_all(b"closed").await.unwrap_err().kind(), + io::ErrorKind::BrokenPipe + ); + sleep(Duration::from_millis(50)).await; + assert!(child.try_wait()?.is_none()); + resize.resize(PtySize::default())?; + + drop(output); + assert_eq!( + resize.resize(PtySize::default()).unwrap_err().kind(), + io::ErrorKind::BrokenPipe + ); + timeout(Duration::from_secs(5), child.wait()).await??; + Ok(()) +} + +#[tokio::test] +async fn dropping_output_does_not_hang_up_while_input_exists() -> io::Result<()> { + let mut command = Command::new("sh"); + command.args(["-c", "stty -echo; printf ready; IFS= read -r line"]); + + let (mut child, controller) = spawn_with_terminal(&mut command, PtySize::default())?; + let (input, mut output, resize) = controller.into_parts(); + let mut ready = [0; 5]; + timeout(Duration::from_secs(5), output.read_exact(&mut ready)).await??; + drop(output); + + sleep(Duration::from_millis(50)).await; + assert!(child.try_wait()?.is_none()); + resize.resize(PtySize::default())?; + drop(input); + assert_eq!( + resize.resize(PtySize::default()).unwrap_err().kind(), + io::ErrorKind::BrokenPipe + ); + timeout(Duration::from_secs(5), child.wait()).await??; + Ok(()) +} + +#[cfg(not(target_os = "macos"))] +async fn assert_direct_child_wait_is_independent_from_descendant_output_eof( + mut command: Command, +) -> io::Result<()> { + struct ReleaseOnDrop(std::path::PathBuf); + + impl Drop for ReleaseOnDrop { + fn drop(&mut self) { + let _ = std::fs::File::create(&self.0); + } + } + + let directory = tempfile::tempdir()?; + let release = directory.path().join("release-descendant"); + let _release_on_drop = ReleaseOnDrop(release.clone()); + command + .args([ + "-c", + "trap '' HUP; (printf ready; while [ ! -e \"$1\" ]; do sleep 1; done) &", + "pty-test", + ]) + .arg(&release); + + let (mut child, controller) = spawn_with_terminal(&mut command, PtySize::default())?; + let (input, mut output, _resize) = controller.into_parts(); + drop(input); + let mut ready = [0; 5]; + timeout(Duration::from_secs(5), output.read_exact(&mut ready)).await??; + assert_eq!(&ready, b"ready"); + assert!( + timeout(Duration::from_secs(5), child.wait()) + .await?? + .success() + ); + + let mut bytes = Vec::new(); + assert!( + timeout(Duration::from_millis(100), output.read_to_end(&mut bytes)) + .await + .is_err() + ); + std::fs::File::create(&release)?; + timeout(Duration::from_secs(5), output.read_to_end(&mut bytes)).await??; + Ok(()) +} + +#[cfg(not(target_os = "macos"))] +#[tokio::test] +async fn direct_child_wait_is_independent_from_descendant_output_eof() -> io::Result<()> { + assert_direct_child_wait_is_independent_from_descendant_output_eof(Command::new("sh")).await +} + +#[cfg(all(not(target_os = "macos"), feature = "process-group"))] +#[tokio::test] +async fn process_group_wait_is_independent_from_descendant_output_eof() -> io::Result<()> { + let mut command = Command::new("sh"); + command.wrap(ProcessGroup::leader()); + assert_direct_child_wait_is_independent_from_descendant_output_eof(command).await +} + +#[tokio::test] +async fn failed_spawn_leaves_command_reusable() -> io::Result<()> { + use std::os::unix::fs::PermissionsExt; + + let directory = tempfile::tempdir()?; + let program = directory.path().join("created-after-first-spawn"); + let mut command = Command::new(&program); + assert_eq!( + spawn_with_terminal(&mut command, PtySize::default()) + .unwrap_err() + .kind(), + io::ErrorKind::NotFound + ); + + std::fs::write(&program, "#!/bin/sh\nprintf reused")?; + std::fs::set_permissions(&program, std::fs::Permissions::from_mode(0o700))?; + let (mut child, controller) = spawn_with_terminal(&mut command, PtySize::default())?; + let (input, mut output, _resize) = controller.into_parts(); + drop(input); + let (status, bytes) = wait_and_drain(child.as_mut(), &mut output).await?; + assert!(status.success()); + assert_eq!(bytes, b"reused"); + Ok(()) +} + +#[tokio::test] +async fn kill_and_start_kill_preserve_repeated_waits() -> io::Result<()> { + for wait_in_kill in [false, true] { + let mut command = Command::new("sh"); + command.args(["-c", "stty -echo; printf ready; while :; do sleep 1; done"]); + let (mut child, controller) = spawn_with_terminal(&mut command, PtySize::default())?; + let (input, mut output, _resize) = controller.into_parts(); + let mut ready = [0; 5]; + timeout(Duration::from_secs(5), output.read_exact(&mut ready)).await??; + assert_eq!(&ready, b"ready"); + drop(input); + + if wait_in_kill { + timeout(Duration::from_secs(5), Box::into_pin(child.kill())).await??; + } else { + child.start_kill()?; + timeout(Duration::from_secs(5), child.wait()).await??; + } + let status = child.try_wait()?.expect("killed child must have exited"); + assert_eq!(child.try_wait()?, Some(status)); + assert_eq!(child.wait().await?, status); + let mut bytes = Vec::new(); + timeout(Duration::from_secs(5), output.read_to_end(&mut bytes)).await??; + } + Ok(()) +} + +#[tokio::test] +async fn cancelling_wait_preserves_child_ownership() -> io::Result<()> { + let mut command = Command::new("sh"); + command.args(["-c", "stty -echo; printf ready; sleep 30"]); + let (mut child, controller) = spawn_with_terminal(&mut command, PtySize::default())?; + let (input, mut output, _resize) = controller.into_parts(); + let mut ready = [0; 5]; + timeout(Duration::from_secs(5), output.read_exact(&mut ready)).await??; + assert_eq!(&ready, b"ready"); + drop(input); + + assert!( + timeout(Duration::from_millis(50), child.wait()) + .await + .is_err() + ); + child.start_kill()?; + let status = timeout(Duration::from_secs(5), child.wait()).await??; + assert_eq!(child.try_wait()?, Some(status)); + let mut bytes = Vec::new(); + timeout(Duration::from_secs(5), output.read_to_end(&mut bytes)).await??; + Ok(()) +} + +#[cfg(any(feature = "process-group", feature = "process-session"))] +#[derive(Debug)] +struct ReapBeforeChildWrapping; + +#[cfg(any(feature = "process-group", feature = "process-session"))] +impl CommandWrapper for ReapBeforeChildWrapping { + fn post_spawn( + &mut self, + _attempt: &mut SpawnAttempt, + child: &mut dyn ChildWrapper, + _command: &Command, + ) -> io::Result<()> { + loop { + if child.try_wait()?.is_some() { + return Ok(()); + } + std::thread::yield_now(); + } + } +} + +#[cfg(feature = "process-group")] +#[tokio::test] +async fn process_group_wraps_a_pty_child_reaped_by_a_post_spawn_hook() -> io::Result<()> { + let mut command = Command::with_new("sh", |command| { + command.args(["-c", "exit 23"]); + }); + command + .wrap(ProcessGroup::leader()) + .wrap(Pty::default()) + .wrap(ReapBeforeChildWrapping); + + let mut child = command.spawn()?; + let controller = child + .take_pty_controller() + .expect("a successful PTY spawn installs one controller"); + drop(controller); + let status = child.wait().await?; + assert_eq!(status.code(), Some(23)); + assert_eq!(child.try_wait()?, Some(status)); + Ok(()) +} + +#[cfg(feature = "process-session")] +#[tokio::test] +async fn process_session_wraps_a_pty_child_reaped_by_a_post_spawn_hook() -> io::Result<()> { + let mut command = Command::with_new("sh", |command| { + command.args(["-c", "exit 29"]); + }); + command + .wrap(ProcessSession) + .wrap(Pty::default()) + .wrap(ReapBeforeChildWrapping); + + let mut child = command.spawn()?; + let controller = child + .take_pty_controller() + .expect("a successful PTY spawn installs one controller"); + drop(controller); + let status = child.wait().await?; + assert_eq!(status.code(), Some(29)); + assert_eq!(child.try_wait()?, Some(status)); + Ok(()) +} + +#[cfg(feature = "process-group")] +async fn assert_group_signal(mut command: Command) -> io::Result<()> { + command.args([ + "-c", + "stty -echo; trap '' HUP; trap 'exit 0' TERM; sleep 30 & printf ready; wait", + ]); + let (mut child, controller) = spawn_with_terminal(&mut command, PtySize::default())?; + assert_eq!(child.as_ref().type_id(), TypeId::of::()); + assert!(child.try_wait()?.is_none()); + + let (input, mut output, _resize) = controller.into_parts(); + let mut ready = [0; 5]; + timeout(Duration::from_secs(5), output.read_exact(&mut ready)).await??; + assert_eq!(&ready, b"ready"); + drop(input); + + child.signal(nix::libc::SIGTERM)?; + let status = timeout(Duration::from_secs(5), child.wait()).await??; + assert_eq!(child.try_wait()?, Some(status)); + assert_eq!(child.try_wait()?, Some(status)); + assert_eq!(child.wait().await?, status); + let mut bytes = Vec::new(); + timeout(Duration::from_secs(5), output.read_to_end(&mut bytes)).await??; + Ok(()) +} + +#[cfg(feature = "process-group")] +#[tokio::test] +async fn process_group_leader_preserves_pty_group_supervision() -> io::Result<()> { + let mut command = Command::new("sh"); + command.wrap(ProcessGroup::leader()); + assert_group_signal(command).await +} + +#[cfg(feature = "process-group")] +#[tokio::test] +async fn process_group_try_wait_supports_terminal_provider_child() -> io::Result<()> { + let mut command = Command::new("sh"); + command.args(["-c", "exit 17"]).wrap(ProcessGroup::leader()); + let (mut child, controller) = spawn_with_terminal(&mut command, PtySize::default())?; + let (input, mut output, _resize) = controller.into_parts(); + drop(input); + + let status = timeout(Duration::from_secs(5), async { + loop { + if let Some(status) = child.try_wait()? { + break Ok::<_, io::Error>(status); + } + sleep(Duration::from_millis(10)).await; + } + }) + .await??; + assert!(unsafe { child.try_inner_child_mut() }.is_none()); + assert_eq!(child.try_wait()?, Some(status)); + assert_eq!(child.wait().await?, status); + let mut bytes = Vec::new(); + timeout(Duration::from_secs(5), output.read_to_end(&mut bytes)).await??; + Ok(()) +} + +#[cfg(feature = "process-group")] +#[tokio::test] +async fn process_group_composes_when_registered_after_first_spawn() -> io::Result<()> { + let mut command = Command::new("sh"); + command.args(["-c", "printf reused"]); + + let (mut child, controller) = spawn_with_terminal(&mut command, PtySize::default())?; + let (input, mut output, _resize) = controller.into_parts(); + drop(input); + let (status, bytes) = wait_and_drain(child.as_mut(), &mut output).await?; + assert!(status.success()); + assert_eq!(bytes, b"reused"); + + command.wrap(ProcessGroup::leader()); + let (mut child, controller) = spawn_with_terminal(&mut command, PtySize::default())?; + assert_eq!(child.as_ref().type_id(), TypeId::of::()); + let (input, mut output, _resize) = controller.into_parts(); + drop(input); + let (status, bytes) = wait_and_drain(child.as_mut(), &mut output).await?; + assert!(status.success()); + assert_eq!(bytes, b"reused"); + Ok(()) +} + +#[cfg(feature = "process-session")] +#[tokio::test] +async fn process_session_preserves_pty_group_supervision() -> io::Result<()> { + let mut command = Command::new("sh"); + command.wrap(ProcessSession); + assert_group_signal(command).await +} + +#[cfg(feature = "process-group")] +#[tokio::test] +async fn rejects_attaching_a_pty_to_an_existing_group() { + for group in [ProcessGroup::attach_to(0), ProcessGroup::attach_to(42)] { + let mut command = Command::new("sh"); + command.args(["-c", "exit 0"]).wrap(group); + assert_eq!( + spawn_with_terminal(&mut command, PtySize::default()) + .unwrap_err() + .kind(), + io::ErrorKind::InvalidInput + ); + } +} + +#[cfg(feature = "process-session")] +#[tokio::test] +async fn rejects_explicit_group_and_session_in_either_order() { + for session_first in [false, true] { + let mut command = Command::new("sh"); + command.args(["-c", "exit 0"]); + if session_first { + command.wrap(ProcessSession).wrap(ProcessGroup::leader()); + } else { + command.wrap(ProcessGroup::leader()).wrap(ProcessSession); + } + assert_eq!( + spawn_with_terminal(&mut command, PtySize::default()) + .unwrap_err() + .kind(), + io::ErrorKind::InvalidInput + ); + } +} + +#[cfg(feature = "reset-sigmask")] +#[tokio::test] +async fn reset_sigmask_unblocks_signals_before_pty_setup() -> io::Result<()> { + use nix::sys::signal::{SigSet, SigmaskHow, Signal, sigprocmask}; + + let mut blocked = SigSet::empty(); + blocked.add(Signal::SIGUSR1); + let mut previous = SigSet::empty(); + sigprocmask(SigmaskHow::SIG_BLOCK, Some(&blocked), Some(&mut previous))?; + + let mut command = Command::new("sh"); + command + .args(["-c", "test -t 0 || exit 2; kill -USR1 $$; printf survived"]) + .wrap(ResetSigmask); + let spawned = spawn_with_terminal(&mut command, PtySize::default()); + sigprocmask(SigmaskHow::SIG_SETMASK, Some(&previous), None)?; + let (mut child, controller) = spawned?; + + let (input, mut output, _resize) = controller.into_parts(); + drop(input); + let status = timeout(Duration::from_secs(5), child.wait()).await??; + assert_eq!(status.signal(), Some(Signal::SIGUSR1 as i32)); + let mut bytes = Vec::new(); + timeout(Duration::from_secs(5), output.read_to_end(&mut bytes)).await??; + assert!(bytes.is_empty()); + Ok(()) +} + +#[cfg(all(feature = "kill-on-drop", feature = "process-session"))] +fn pid_alive(pid: Pid) -> bool { + !matches!(kill(pid, None), Err(nix::errno::Errno::ESRCH)) +} + +#[cfg(all(feature = "kill-on-drop", feature = "process-session"))] +struct KillPid(Option); + +#[cfg(all(feature = "kill-on-drop", feature = "process-session"))] +impl KillPid { + fn new(pid: Pid) -> Self { + Self(Some(pid)) + } + + fn disarm(&mut self) { + self.0 = None; + } +} + +#[cfg(all(feature = "kill-on-drop", feature = "process-session"))] +impl Drop for KillPid { + fn drop(&mut self) { + if let Some(pid) = self.0 { + let _ = kill(pid, Signal::SIGKILL); + } + } +} + +#[cfg(all(feature = "kill-on-drop", feature = "process-session"))] +#[tokio::test] +async fn kill_on_drop_remains_direct_child_only_with_a_pty_session() -> io::Result<()> { + let directory = tempfile::tempdir()?; + let ready = directory.path().join("descendant-ready"); + let acknowledged = directory.path().join("descendant-acknowledged"); + let mut command = Command::new("sh"); + command + .args([ + "-c", + r#"stty -echo; trap '' HUP; (trap '' HUP; trap ': > "$2"; exit 0' USR1; : > "$1"; while :; do sleep 1; done) & printf '%s:%s\n' "$$" "$!"; wait"#, + "pty-test", + ]) + .arg(&ready) + .arg(&acknowledged) + .wrap(ProcessSession) + .wrap(KillOnDrop); + let (child, controller) = spawn_with_terminal(&mut command, PtySize::default())?; + let (input, output, _resize) = controller.into_parts(); + let mut output = BufReader::new(output); + let mut line = String::new(); + timeout(Duration::from_secs(5), output.read_line(&mut line)).await??; + let (direct, descendant) = line.trim().split_once(':').unwrap(); + let direct = Pid::from_raw(direct.parse().unwrap()); + let descendant = Pid::from_raw(descendant.parse().unwrap()); + assert_eq!(child.id(), Some(direct.as_raw() as u32)); + let mut direct_cleanup = KillPid::new(direct); + let mut descendant_cleanup = KillPid::new(descendant); + assert!(pid_alive(direct)); + timeout(Duration::from_secs(5), async { + while !ready.exists() { + sleep(Duration::from_millis(10)).await; + } + }) + .await + .expect("descendant did not install its signal handler"); + + drop(input); + drop(child); + timeout(Duration::from_secs(5), async { + while pid_alive(direct) { + sleep(Duration::from_millis(10)).await; + } + }) + .await + .expect("kill-on-drop did not terminate the direct child"); + direct_cleanup.disarm(); + + kill(descendant, Signal::SIGUSR1)?; + timeout(Duration::from_secs(5), async { + while !acknowledged.exists() { + sleep(Duration::from_millis(10)).await; + } + }) + .await + .expect("descendant did not acknowledge its signal"); + timeout(Duration::from_secs(5), async { + while pid_alive(descendant) { + sleep(Duration::from_millis(10)).await; + } + }) + .await + .expect("acknowledged descendant did not exit"); + descendant_cleanup.disarm(); + Ok(()) +} + +#[test] +fn validates_character_dimensions() { + assert_eq!( + PtySize::new(0, 80).unwrap_err().kind(), + io::ErrorKind::InvalidInput + ); + assert_eq!( + PtySize::new(24, 0).unwrap_err().kind(), + io::ErrorKind::InvalidInput + ); +} + +#[test] +fn validates_initial_size_before_opening_a_pty() { + let mut command = Command::new("ignored"); + let size = PtySize { + rows: 0, + columns: 80, + pixel_width: 0, + pixel_height: 0, + }; + assert_eq!( + spawn_with_terminal(&mut command, size).unwrap_err().kind(), + io::ErrorKind::InvalidInput + ); +} diff --git a/tests/tokio_pty_unsupported.rs b/tests/tokio_pty_unsupported.rs new file mode 100644 index 0000000..b20847c --- /dev/null +++ b/tests/tokio_pty_unsupported.rs @@ -0,0 +1,64 @@ +#![cfg(all( + feature = "pty", + not(any( + target_os = "android", + target_os = "dragonfly", + target_os = "freebsd", + target_os = "illumos", + target_os = "linux", + target_os = "macos", + target_os = "netbsd", + target_os = "openbsd", + target_os = "solaris" + )) +))] + +use std::{ + io, + sync::{ + Arc, + atomic::{AtomicBool, Ordering}, + }, +}; + +use process_wrap::tokio::{Command, CommandWrapper, Pty, PtySize, SpawnAttempt}; + +#[test] +fn spawning_is_explicitly_unsupported() { + let mut command = Command::new("ignored"); + command.wrap(Pty::default()); + assert_eq!( + command.spawn().unwrap_err().kind(), + io::ErrorKind::Unsupported + ); +} + +#[derive(Debug)] +struct ObserveHook(Arc); + +impl CommandWrapper for ObserveHook { + fn pre_spawn(&mut self, _attempt: &mut SpawnAttempt, _command: &Command) -> io::Result<()> { + self.0.store(true, Ordering::SeqCst); + Ok(()) + } +} + +#[test] +fn unsupported_precedes_command_validation_and_hooks() { + let called = Arc::new(AtomicBool::new(false)); + let native = tokio::process::Command::new("ignored"); + let mut command = Command::from(native); + command + .wrap(Pty::new(PtySize { + rows: 0, + columns: 0, + pixel_width: 0, + pixel_height: 0, + })) + .wrap(ObserveHook(Arc::clone(&called))); + assert_eq!( + command.spawn().unwrap_err().kind(), + io::ErrorKind::Unsupported + ); + assert!(!called.load(Ordering::SeqCst)); +} diff --git a/tests/unix_attempt_policy.rs b/tests/unix_attempt_policy.rs index 458601a..b226d61 100644 --- a/tests/unix_attempt_policy.rs +++ b/tests/unix_attempt_policy.rs @@ -396,7 +396,7 @@ macro_rules! unix_attempt_policy_tests { } #[test] - fn attach_to_tracks_the_direct_pid_and_actual_group() { + fn attach_to_tracks_the_direct_pid_without_waiting_for_the_foreign_group() { let runtime = runtime(); let _runtime_guard = runtime.as_ref().map(tokio::runtime::Runtime::enter); let external = ExternalGroup::spawn(); @@ -415,9 +415,13 @@ macro_rules! unix_attempt_policy_tests { assert_eq!(group_child.pgid(), pgid); sleep(Duration::from_millis(200)); - assert_eq!(child.try_wait().unwrap(), None); + let status = child + .try_wait() + .unwrap() + .expect("the direct child exits independently of the foreign group"); + assert_eq!(status.code(), Some(7)); external.kill(); - assert_eq!(wait_for_exit(child.as_mut()).code(), Some(7)); + assert_eq!(wait_for_exit(child.as_mut()), status); } #[test]