From 5a7704e96da032af7e0c262c228782f99bb32579 Mon Sep 17 00:00:00 2001 From: Enrique Saurez Date: Sat, 7 Nov 2026 15:52:18 -0800 Subject: [PATCH 1/7] host: verify VM endpoints and expose the retained transport Expose the single existing timed transport through the host transport module. Verify endpoint servers against the recorded and held VM process before authentication, and retain NVXS connection behavior while preserving causes in the crate error. Keep compatibility paths and update affected control-session and runtime-adapter callers. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> Copilot-Session: 988ad39e-5a8f-4e76-80b8-ccbe7f1f3c58 --- nvx-rs/README.md | 26 +- nvx-rs/src/host/mod.rs | 6 +- nvx-rs/src/host/nvxs.rs | 306 +------------ nvx-rs/src/host/platform/tests.rs | 2 + nvx-rs/src/host/platform/windows/endpoint.rs | 2 +- .../host/platform/windows/endpoint/tests.rs | 42 +- nvx-rs/src/host/platform/windows/tests.rs | 1 + nvx-rs/src/host/session.rs | 75 ++- nvx-rs/src/host/session/tests.rs | 1 + nvx-rs/src/host/transport.rs | 431 ++++++++++++++++++ nvx-rs/src/host/transport/tests.rs | 268 +++++++++++ nvx-rs/tests/nvxs/host.rs | 136 ++++-- .../src/bin/runtime-adapter-host.rs | 5 +- tests/runtime-adapter/src/host.rs | 39 +- tests/runtime-adapter/src/host/executions.rs | 8 +- tests/runtime-adapter/tests/host_session.rs | 85 ++-- 16 files changed, 990 insertions(+), 443 deletions(-) create mode 100644 nvx-rs/src/host/transport.rs create mode 100644 nvx-rs/src/host/transport/tests.rs diff --git a/nvx-rs/README.md b/nvx-rs/README.md index 595f88ad9..7ae95118e 100644 --- a/nvx-rs/README.md +++ b/nvx-rs/README.md @@ -135,7 +135,8 @@ changes only when the module's purpose does. | [`nvx::host::filesystem`](#nvxhostfilesystem) | The planner of the host paths that every NVX caller exports to the guest through OpenVMM's virtio-fs device | | [`nvx::host::store`](#nvxhoststore) | `SandboxStore`: the runtime records of one sandbox, which every host that supervises it shares under one lock | | [`nvx::host::instance`](#nvxhostinstance) | `Instance`: a VM in a sandbox store of its own, which the store, the supervisor, and the control session drive | -| [`nvx::host::nvxs`](#nvxhostnvxs) | `HostConnection`: the host's role of the NVXS session over a platform transport | +| [`nvx::host::nvxs`](#nvxhostnvxs) | Compatibility paths for the host NVXS connection and its retained cause details | +| [`nvx::host::transport`](#nvxhosttransport) | A verified VM endpoint connection and the blocking host NVXS driver, with the existing timed transport | | [`nvx::host::process`](#nvxhostprocess) | Verified references to OpenVMM processes, through which callers wait for them and end them | | [`nvx::host::supervisor`](#nvxhostsupervisor) | The supervisor of a sandbox's VMM, beneath the CLI, `Instance`, and external orchestrators | | [`nvx::host::template`](#nvxhosttemplate) | Legacy template capture and clone plans and store, with aliases of the snapshot metadata | @@ -908,6 +909,17 @@ instance creation or launch records, respectively. ### `nvx::host::nvxs` +Compatibility paths for `HostConnection` and its retained cause details, now +owned by [`nvx::host::transport`](#nvxhosttransport). + +### `nvx::host::transport` + +`connect_verified` opens and holds the recorded process by its boot, ID and +start time, retries while the endpoint is unavailable until the caller's +deadline, and checks its server before authentication: Unix peer credentials +on Linux and the named-pipe server process ID on Windows. An indeterminate +process check remains an error, never exit proof. + `HostConnection`: the host's role of the NVXS session over a platform transport, on which `nvx::host::session` and other host applications run their protocols. It attaches with the launch capability, waiting while the broker asks it to, @@ -915,6 +927,18 @@ sends application bytes in as many DATA records as they take, and returns each DATA payload from the guest, with deadlines; a receive that times out keeps a partial record. +These APIs return the crate's `Result`: a deadline is `TimedOut`, an exited +recorded VM `Exited`, a refused, closed, reset or broken session `Protocol`, +and another transport failure `Io`. Original I/O and protocol causes are +retained directly; attachment refusal, reset and ended-connection details +remain available as sources. + +`Transport` and `TransportHandle` re-export the existing platform types, with +one definition beneath the OS adapters. Timed reads and writes, native-handle +duplication and `unread` are unchanged: Linux reports Unix `SIOCOUTQ` charges, +and Windows reports `None`. The old `host::session::connect` remains a +compatibility wrapper. + ### `nvx::host::process` Verified process references. A `ProcessRef` names one process by its ID, its diff --git a/nvx-rs/src/host/mod.rs b/nvx-rs/src/host/mod.rs index 950a59525..261315c90 100644 --- a/nvx-rs/src/host/mod.rs +++ b/nvx-rs/src/host/mod.rs @@ -35,8 +35,9 @@ //! - [`template`] retains the legacy capture and clone plans, the [`Store`](template::Store) of //! published generations and their leases, and aliases of the snapshot metadata. Its //! [`TemplatePlan`](template::TemplatePlan) lowers an adapter's plan for a capture. -//! - [`session`] speaks NVX's control protocol, `NVXC`, over [`nvxs`], which runs the host's role -//! of the NVXS session for it and for any other application protocol, over a platform +//! - [`transport`] verifies a VM's endpoint and runs the host's role of the NVXS session. +//! [`nvxs`] retains compatibility paths. [`session`] speaks NVX's control protocol, `NVXC`, +//! over that connection for it and for any other application protocol, over a platform //! transport. [`platform`] holds the primitives of Linux and Windows hosts. Other hosts compile //! but report NVX as unavailable. //! - [`process`] names processes by verified references: their ID, start time, and host boot, @@ -103,4 +104,5 @@ mod sys; pub mod template; #[cfg(test)] pub(crate) mod tests; +pub mod transport; pub mod vm; diff --git a/nvx-rs/src/host/nvxs.rs b/nvx-rs/src/host/nvxs.rs index e1373a655..9ece5e53e 100644 --- a/nvx-rs/src/host/nvxs.rs +++ b/nvx-rs/src/host/nvxs.rs @@ -1,305 +1,3 @@ -//! The host's I/O driver of an NVXS session: [`HostConnection`] runs the host's role of -//! [`protocol::nvxs`](crate::protocol::nvxs) over a platform [`Transport`], for NVX's control -//! protocol or any other application protocol. -//! -//! [`HostConnection::attach`] sends the launch capability and waits until the broker attaches the -//! host. [`send`](HostConnection::send) writes application bytes in as many DATA records as they -//! take, and [`receive`](HostConnection::receive) returns the payload of each DATA record from the -//! guest, in order. The connection reads at most 16 KiB at a time and keeps what it read until -//! the session consumes it, besides at most one partial record, so a receive that times out loses -//! nothing and callers can poll with short deadlines. -//! -//! A failure that leaves the stream unusable ends the connection, and every later call fails with -//! [`HostConnectionError::Ended`]. That includes a send that fails or times out after it wrote -//! anything, because a record or a message that the transport took only in part cannot be -//! resumed. -//! -//! # Example -//! -//! ```no_run -//! # fn main() -> Result<(), Box> { -//! use std::time::{Duration, Instant}; -//! -//! use nvx::host::nvxs::HostConnection; -//! use nvx::host::session; -//! # let (endpoint, pid, start_time) = ("", 0, 0); -//! # let capability = [0; nvx::protocol::CAPABILITY_LEN]; -//! -//! // The VM's control endpoint, OpenVMM's process, and the capability that it was launched with. -//! let deadline = Instant::now() + Duration::from_secs(10); -//! let transport = session::connect(endpoint, pid, start_time, deadline)?; -//! let mut connection = HostConnection::attach(transport, &capability, deadline)?; -//! connection.send(b"a message of the application's protocol", Some(deadline))?; -//! let payload = connection.receive(Some(deadline))?; -//! # Ok(()) -//! # } -//! ``` +//! Compatibility paths for the host's NVXS connection, now owned by [`super::transport`]. -use std::fmt; -use std::io; -use std::time::{Duration, Instant}; - -use super::platform::Transport; -use crate::protocol::CAPABILITY_LEN; -use crate::protocol::nvxs::{ - ErrorCode, HostEvent, HostSession, Identity, RecordType, SessionError, -}; - -/// Most bytes that one read from the transport takes. -const READ_CHUNK: usize = 16 * 1024; - -/// Failure of a [`HostConnection`]. -#[derive(Debug)] -#[non_exhaustive] -pub enum HostConnectionError { - /// The deadline passed. A receive that times out keeps what arrived, and the connection stays - /// usable, as it does after a send that times out before it wrote anything. - TimedOut, - /// The broker closed the stream at a record boundary. - Closed, - /// The broker refused the host's attachment. - Rejected(ErrorCode), - /// The broker reset the session, which ended the host's attachment. The identity is the - /// session's new one. - Reset(Identity), - /// The broker violated the protocol, or the stream ended inside a record. - Protocol(SessionError), - /// The transport failed. - Io(io::Error), - /// An earlier failure ended the connection. - Ended, -} - -impl fmt::Display for HostConnectionError { - fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result { - match self { - Self::TimedOut => formatter.write_str("the NVXS operation timed out"), - Self::Closed => formatter.write_str("the broker closed the NVXS session"), - Self::Rejected(code) => { - write!(formatter, "the broker refused the NVXS attachment: {code}") - } - Self::Reset(identity) => write!( - formatter, - "the broker reset the NVXS session to epoch {}", - identity.epoch - ), - Self::Protocol(error) => error.fmt(formatter), - Self::Io(error) => write!(formatter, "NVXS I/O failed: {error}"), - Self::Ended => formatter.write_str("an earlier failure ended the NVXS connection"), - } - } -} - -impl std::error::Error for HostConnectionError { - fn source(&self) -> Option<&(dyn std::error::Error + 'static)> { - match self { - Self::Protocol(error) => Some(error), - Self::Io(error) => Some(error), - _ => None, - } - } -} - -/// The host's end of an NVXS session over a platform transport. -pub struct HostConnection { - /// Transport connected to the broker's host endpoint. - transport: Box, - /// The host's role of the session. - session: HostSession, - /// Identity that the broker attached the host to. - identity: Identity, - /// Bytes read from the transport. - input: Box<[u8]>, - /// Start of the bytes in `input` that the session has not consumed. - start: usize, - /// End of the bytes in `input`. - end: usize, - /// Records encoded for the transport. - output: Vec, - /// Whether a failure ended the connection. - ended: bool, -} - -impl HostConnection { - /// Attaches the host to the broker over `transport` with the launch `capability`, and waits - /// until `deadline` while the broker asks the host to wait for the guest. - pub fn attach( - transport: Box, - capability: &[u8; CAPABILITY_LEN], - deadline: Instant, - ) -> Result { - let mut connection = Self { - transport, - session: HostSession::new(), - // The broker's READY replaces this before the caller sees the connection. - identity: Identity { - instance_id: [0; 16], - epoch: 0, - }, - input: vec![0; READ_CHUNK].into_boxed_slice(), - start: 0, - end: 0, - output: Vec::new(), - ended: false, - }; - connection - .session - .attach(capability, &mut connection.output) - .map_err(HostConnectionError::Protocol)?; - let timeout = remaining(Some(deadline))?; - connection.write(timeout)?; - loop { - match connection.next_event(Some(deadline))? { - HostEvent::Wait => {} - HostEvent::Attached(identity) => { - connection.identity = identity; - return Ok(connection); - } - HostEvent::Rejected(code) => return Err(HostConnectionError::Rejected(code)), - // The session reports these only once the broker attached the host. - HostEvent::Data(_) => return Err(unexpected(RecordType::Data)), - HostEvent::Reset(_) => return Err(unexpected(RecordType::Reset)), - } - } - } - - /// Returns the identity that the broker attached the host to. - pub fn identity(&self) -> Identity { - self.identity - } - - /// Returns the host's role of the session. - pub fn session(&self) -> &HostSession { - &self.session - } - - /// Sends `data` to the guest by `deadline`, or without a time limit, in as many DATA records - /// as it takes, of at most [`OUTER_MAX_PAYLOAD`](crate::protocol::OUTER_MAX_PAYLOAD) bytes - /// each. - /// - /// Data that fits one record travels in one, so an application protocol that needs one frame - /// per record sends frames that fit. Empty data sends nothing. - pub fn send( - &mut self, - data: &[u8], - deadline: Option, - ) -> Result<(), HostConnectionError> { - self.check()?; - let mut rest = data; - let mut wrote = false; - while !rest.is_empty() { - let timeout = match remaining(deadline) { - Ok(timeout) => timeout, - Err(error) if wrote => return Err(self.end(error)), - Err(error) => return Err(error), - }; - self.output.clear(); - let taken = self - .session - .send(rest, &mut self.output) - .map_err(|error| self.end(HostConnectionError::Protocol(error)))?; - self.write(timeout)?; - wrote = true; - rest = &rest[taken..]; - } - Ok(()) - } - - /// Returns the payload of the next DATA record from the guest, waiting until `deadline`, or - /// without a time limit. - pub fn receive(&mut self, deadline: Option) -> Result, HostConnectionError> { - self.check()?; - let error = match self.next_event(deadline)? { - HostEvent::Data(payload) => return Ok(payload), - HostEvent::Reset(identity) => HostConnectionError::Reset(identity), - // The session reports these only while the broker attaches the host. - HostEvent::Wait => unexpected(RecordType::Wait), - HostEvent::Attached(_) => unexpected(RecordType::Ready), - HostEvent::Rejected(_) => unexpected(RecordType::Error), - }; - Err(self.end(error)) - } - - /// Returns the session's next event, reading from the transport until a record completes. - fn next_event(&mut self, deadline: Option) -> Result { - loop { - if self.start < self.end { - let (consumed, event) = self - .session - .receive(&self.input[self.start..self.end]) - .map_err(|error| self.end(HostConnectionError::Protocol(error)))?; - self.start += consumed; - match event { - Some(event) => return Ok(event), - None => continue, - } - } - let timeout = remaining(deadline)?; - match self.transport.read(&mut self.input, timeout) { - Ok(0) => { - let error = match self.session.disconnect() { - Ok(()) => HostConnectionError::Closed, - Err(error) => HostConnectionError::Protocol(error), - }; - return Err(self.end(error)); - } - Ok(count) => { - self.start = 0; - self.end = count.min(self.input.len()); - } - Err(error) if error.kind() == io::ErrorKind::Interrupted => {} - Err(error) if error.kind() == io::ErrorKind::TimedOut => { - return Err(HostConnectionError::TimedOut); - } - Err(error) => return Err(self.end(HostConnectionError::Io(error))), - } - } - } - - /// Writes the encoded records within `timeout`, ending the connection if that fails. - fn write(&mut self, timeout: Option) -> Result<(), HostConnectionError> { - self.transport - .write_all(&self.output, timeout) - .map_err(|error| { - let error = if error.kind() == io::ErrorKind::TimedOut { - HostConnectionError::TimedOut - } else { - HostConnectionError::Io(error) - }; - self.end(error) - }) - } - - /// Fails if an earlier failure ended the connection. - fn check(&self) -> Result<(), HostConnectionError> { - if self.ended { - Err(HostConnectionError::Ended) - } else { - Ok(()) - } - } - - /// Ends the connection with `error`, and returns it. - fn end(&mut self, error: HostConnectionError) -> HostConnectionError { - self.ended = true; - error - } -} - -/// Converts a deadline into the time left, failing once it has passed. -fn remaining(deadline: Option) -> Result, HostConnectionError> { - match deadline { - None => Ok(None), - Some(deadline) => deadline - .checked_duration_since(Instant::now()) - .filter(|remaining| !remaining.is_zero()) - .map(Some) - .ok_or(HostConnectionError::TimedOut), - } -} - -/// Returns the failure for an event that the session reports in no state that the connection -/// calls it in. -fn unexpected(record_type: RecordType) -> HostConnectionError { - HostConnectionError::Protocol(SessionError::Unexpected(record_type)) -} +pub use super::transport::{HostConnection, HostConnectionError}; diff --git a/nvx-rs/src/host/platform/tests.rs b/nvx-rs/src/host/platform/tests.rs index 100607c0f..6eb929811 100644 --- a/nvx-rs/src/host/platform/tests.rs +++ b/nvx-rs/src/host/platform/tests.rs @@ -5,6 +5,8 @@ pub(crate) use super::linux::tests::{ #[cfg(windows)] pub(crate) use super::windows::tests::set_access; #[cfg(windows)] +pub(crate) use super::windows::tests::{accept_control, control_server}; +#[cfg(windows)] pub(crate) use super::windows::tests::{ access, claim_released_within, inherits_file, untrusted_log_job, user_sid, with_spawn_window, }; diff --git a/nvx-rs/src/host/platform/windows/endpoint.rs b/nvx-rs/src/host/platform/windows/endpoint.rs index 9bafc42a9..5ade3b93c 100644 --- a/nvx-rs/src/host/platform/windows/endpoint.rs +++ b/nvx-rs/src/host/platform/windows/endpoint.rs @@ -386,4 +386,4 @@ impl Transport for PipeTransport { } #[cfg(test)] -mod tests; +pub(crate) mod tests; diff --git a/nvx-rs/src/host/platform/windows/endpoint/tests.rs b/nvx-rs/src/host/platform/windows/endpoint/tests.rs index 410ac39e0..42e283d69 100644 --- a/nvx-rs/src/host/platform/windows/endpoint/tests.rs +++ b/nvx-rs/src/host/platform/windows/endpoint/tests.rs @@ -5,14 +5,8 @@ use windows_sys::Win32::Storage::FileSystem::PIPE_ACCESS_DUPLEX; use super::*; -#[test] -fn a_duplicated_transport_keeps_the_verified_connection_without_reconnecting() { - let endpoint = format!( - r"\\.\pipe\nvx-transport-{}-{}", - std::process::id(), - crate::host::store::LaunchId::generate().unwrap().as_str() - ); - let name = wide(&endpoint); +pub(crate) fn control_server(endpoint: &str) -> File { + let name = wide(endpoint); // SAFETY: the name is NUL-terminated; this creates a fresh local byte pipe. let server = owned(unsafe { CreateNamedPipeW( @@ -27,21 +21,39 @@ fn a_duplicated_transport_keeps_the_verified_connection_without_reconnecting() { ) }) .unwrap(); + File::from(server) +} + +pub(crate) fn accept_control(server: &File) -> i32 { + use std::os::windows::io::{AsHandle, AsRawHandle}; + + // SAFETY: the synchronous server and its handle remain valid while it accepts. + let connected = + unsafe { ConnectNamedPipe(server.as_handle().as_raw_handle(), ptr::null_mut()) }; + if connected == 0 { + assert_eq!(last_error(), ERROR_PIPE_CONNECTED); + } + connected +} + +#[test] +fn a_duplicated_transport_keeps_the_verified_connection_without_reconnecting() { + let endpoint = format!( + r"\\.\pipe\nvx-transport-{}-{}", + std::process::id(), + crate::host::store::LaunchId::generate().unwrap().as_str() + ); + let server = control_server(&endpoint); let transport = connect_endpoint(&endpoint, std::process::id(), Duration::from_secs(2)).unwrap(); drop(transport.try_clone_handle().unwrap().unwrap()); let pipe = transport.try_clone_handle().unwrap().unwrap(); drop(transport); - // SAFETY: the synchronous server is valid; the client connected before this call. - assert_eq!( - unsafe { ConnectNamedPipe(raw(&server), ptr::null_mut()) }, - 0 - ); - assert_eq!(last_error(), ERROR_PIPE_CONNECTED); + assert_eq!(accept_control(&server), 0); // SAFETY: this creates a fresh unnamed manual-reset event. let event = owned(unsafe { CreateEventW(ptr::null(), 1, 0, ptr::null()) }).unwrap(); let mut duplicate = PipeTransport { pipe, event }; - let mut server = File::from(server); + let mut server = server; duplicate .write_all(b"request", Some(Duration::from_secs(2))) .unwrap(); diff --git a/nvx-rs/src/host/platform/windows/tests.rs b/nvx-rs/src/host/platform/windows/tests.rs index 67c72a317..281d5a8b0 100644 --- a/nvx-rs/src/host/platform/windows/tests.rs +++ b/nvx-rs/src/host/platform/windows/tests.rs @@ -19,6 +19,7 @@ use windows_sys::Win32::System::Threading::{ CreateEventW, GetCurrentProcess, INFINITE, OpenProcess, PROCESS_DUP_HANDLE, TerminateProcess, }; +pub(crate) use super::endpoint::tests::{accept_control, control_server}; use super::identity::status_error; use super::log_claim::{local_directory, log_job_id, log_job_name, log_job_processes, object_name}; pub(crate) use super::private::tests::set_access; diff --git a/nvx-rs/src/host/session.rs b/nvx-rs/src/host/session.rs index 78284430c..410ac830e 100644 --- a/nvx-rs/src/host/session.rs +++ b/nvx-rs/src/host/session.rs @@ -3,11 +3,10 @@ use std::fmt; use std::io; -use std::thread; -use std::time::{Duration, Instant}; +use std::time::Instant; -use super::nvxs::{HostConnection, HostConnectionError}; -use super::platform::{self, Transport}; +use super::process::{BootId, ProcessRef}; +use super::transport::{self, HostConnection, HostConnectionError, Transport}; use crate::protocol::nvxs::{self, RecordError, RecordType}; use crate::protocol::{ self, APP_CANCEL, APP_CAPTURE, APP_ERROR, APP_EXEC, APP_EXIT, APP_FEATURES, APP_HEADER_LEN, @@ -15,6 +14,7 @@ use crate::protocol::{ ExitCategory, GuestFeatures, HostMap, OUTER_MAX_PAYLOAD, ProtocolError, UNSUPPORTED_OPERATION, Workload, }; +use crate::{Error, ErrorKind}; /// Failure of a control session. #[derive(Debug)] @@ -80,28 +80,31 @@ pub fn connect( start_time: u64, deadline: Instant, ) -> Result, SessionError> { - loop { - let remaining = deadline.saturating_duration_since(Instant::now()); - match platform::connect_endpoint(endpoint, pid, remaining.min(Duration::from_millis(250))) { - Ok(transport) => return Ok(transport), - Err(error) if error.kind() == io::ErrorKind::PermissionDenied => { - return Err(SessionError::Refused(error)); - } - Err(error) if error.kind() == io::ErrorKind::Unsupported => { - return Err(SessionError::Io(error)); + let reference = ProcessRef { + pid, + start_time, + boot: BootId::current()?, + }; + transport::connect_verified(endpoint, &reference, deadline).map_err(connect_failure) +} + +fn connect_failure(error: Error) -> SessionError { + let (kind, message, source) = error.into_parts(); + match kind { + ErrorKind::TimedOut => SessionError::TimedOut, + ErrorKind::Exited => SessionError::ProcessExited, + _ => { + let Some(source) = source else { + return SessionError::Protocol(message); + }; + match source.downcast::() { + Ok(error) if error.kind() == io::ErrorKind::PermissionDenied => { + SessionError::Refused(*error) + } + Ok(error) => SessionError::Io(*error), + Err(source) => SessionError::Io(io::Error::other(source)), } - Err(_) => {} } - match platform::process_start_time(pid) { - Ok(Some(current)) if current == start_time => {} - // An inconclusive check keeps waiting for the endpoint until the deadline. - Err(_) => {} - Ok(_) => return Err(SessionError::ProcessExited), - } - if Instant::now() >= deadline { - return Err(SessionError::TimedOut); - } - thread::sleep(Duration::from_millis(25)); } } @@ -396,7 +399,29 @@ enum Stage { /// A violation keeps the message of the check that it fails: a malformed or oversized header, a /// payload in WAIT or READY, an empty DATA record, which carries no frame, and otherwise an /// invalid attach response or data record. A stream that ends inside a record has closed. -fn failure(error: HostConnectionError, stage: Stage) -> SessionError { +fn failure(error: Error, stage: Stage) -> SessionError { + let (kind, message, source) = error.into_parts(); + let Some(source) = source else { + return match kind { + ErrorKind::TimedOut => SessionError::TimedOut, + ErrorKind::Exited => SessionError::ProcessExited, + _ => SessionError::Protocol(message), + }; + }; + let error = match source.downcast::() { + Ok(error) => *error, + Err(source) => match source.downcast::() { + Ok(error) => HostConnectionError::Protocol(*error), + Err(source) => match source.downcast::() { + Ok(error) => HostConnectionError::Io(*error), + Err(source) => HostConnectionError::Io(io::Error::other(source)), + }, + }, + }; + connection_failure(error, stage) +} + +fn connection_failure(error: HostConnectionError, stage: Stage) -> SessionError { let violation = match error { HostConnectionError::TimedOut => return SessionError::TimedOut, HostConnectionError::Closed => return SessionError::Closed, diff --git a/nvx-rs/src/host/session/tests.rs b/nvx-rs/src/host/session/tests.rs index d1a95cdd3..6e23f915f 100644 --- a/nvx-rs/src/host/session/tests.rs +++ b/nvx-rs/src/host/session/tests.rs @@ -1,5 +1,6 @@ use std::collections::VecDeque; use std::sync::{Arc, Mutex}; +use std::time::Duration; use super::*; use crate::protocol::nvxs::{ErrorCode, Identity}; diff --git a/nvx-rs/src/host/transport.rs b/nvx-rs/src/host/transport.rs new file mode 100644 index 000000000..de9e67d56 --- /dev/null +++ b/nvx-rs/src/host/transport.rs @@ -0,0 +1,431 @@ +//! Verified connections to a VM and the host's blocking NVXS driver. +//! +//! [`connect_verified`] checks the recorded VM process and the endpoint's server before any +//! capability is sent. [`HostConnection`] runs the host's role of +//! [`protocol::nvxs`](crate::protocol::nvxs) over a platform [`Transport`], for NVX's control +//! protocol or any other application protocol. +//! +//! [`HostConnection::attach`] sends the launch capability and waits until the broker attaches the +//! host. [`send`](HostConnection::send) writes application bytes in as many DATA records as they +//! take, and [`receive`](HostConnection::receive) returns the payload of each DATA record from the +//! guest, in order. The connection reads at most 16 KiB at a time and keeps what it read until +//! the session consumes it, besides at most one partial record, so a receive that times out loses +//! nothing and callers can poll with short deadlines. +//! +//! A failure that leaves the stream unusable ends the connection, and every later call fails with +//! [`ErrorKind::Protocol`]. That includes a send that fails or times out after it wrote +//! anything, because a record or a message that the transport took only in part cannot be +//! resumed. +//! +//! # Example +//! +//! ```no_run +//! # fn main() -> Result<(), Box> { +//! use std::time::{Duration, Instant}; +//! +//! use nvx::host::transport::{self, HostConnection}; +//! # let endpoint = ""; +//! # let process = nvx::host::process::ProcessRef { +//! # pid: 1, start_time: 1, +//! # boot: nvx::host::process::BootId::current()?, +//! # }; +//! # let capability = [0; nvx::protocol::CAPABILITY_LEN]; +//! +//! // The VM's control endpoint, OpenVMM's process, and the capability that it was launched with. +//! let deadline = Instant::now() + Duration::from_secs(10); +//! let transport = transport::connect_verified(endpoint, &process, deadline)?; +//! let mut connection = HostConnection::attach(transport, &capability, deadline)?; +//! connection.send(b"a message of the application's protocol", Some(deadline))?; +//! let payload = connection.receive(Some(deadline))?; +//! # Ok(()) +//! # } +//! ``` + +use std::fmt; +use std::io; +use std::thread; +use std::time::{Duration, Instant}; + +use super::platform; +use super::process::{Process, ProcessRef}; +use crate::protocol::CAPABILITY_LEN; +use crate::protocol::nvxs::{ + ErrorCode, HostEvent, HostSession, Identity, RecordType, SessionError, +}; +use crate::{Error, ErrorKind, Result}; + +pub use super::platform::Transport; +#[cfg(any(target_os = "linux", windows))] +pub use super::platform::TransportHandle; + +/// Connects to the endpoint served by the recorded VM process, retrying while the endpoint is +/// not yet available, until `deadline`. +/// +/// Opens and holds the process with its recorded boot, ID and start time. Linux checks the Unix +/// socket's peer credentials; Windows checks the named pipe's server process ID. The held +/// process is checked again after connection, before the transport can send a capability. +/// +/// A passed deadline is [`ErrorKind::TimedOut`], an exited recorded process +/// [`ErrorKind::Exited`], a refused or broken session [`ErrorKind::Protocol`], and an +/// indeterminate process check or other transport failure [`ErrorKind::Io`]. I/O failures +/// retain their original source; a confirmed exit has a `NotFound` I/O cause. +pub fn connect_verified( + endpoint: &str, + reference: &ProcessRef, + deadline: Instant, +) -> Result> { + let process = Process::open(reference) + .map_err(|error| Error::io("cannot verify the endpoint's VM process").with_source(error))? + .ok_or_else(exited)?; + let mut unavailable = None; + loop { + check_running(&process)?; + let remaining = deadline + .checked_duration_since(Instant::now()) + .filter(|remaining| !remaining.is_zero()) + .ok_or_else(|| { + Error::timed_out("the control connection timed out").with_source( + unavailable + .take() + .unwrap_or_else(|| io::Error::from(io::ErrorKind::TimedOut)), + ) + })?; + let attempt = platform::connect_endpoint( + endpoint, + reference.pid, + remaining.min(Duration::from_millis(250)), + ); + check_running(&process)?; + match attempt { + Ok(transport) => return Ok(transport), + Err(error) + if matches!( + error.kind(), + io::ErrorKind::NotFound + | io::ErrorKind::ConnectionRefused + | io::ErrorKind::WouldBlock + | io::ErrorKind::TimedOut + | io::ErrorKind::Interrupted + ) => + { + if Instant::now() >= deadline { + return Err( + Error::timed_out("the control connection timed out").with_source(error) + ); + } + unavailable = Some(error); + } + Err(error) => { + let kind = if error.kind() == io::ErrorKind::PermissionDenied { + ErrorKind::Protocol + } else { + io_kind(&error) + }; + return Err( + Error::new(kind, "cannot connect to the control endpoint").with_source(error) + ); + } + } + thread::sleep( + Duration::from_millis(25).min(deadline.saturating_duration_since(Instant::now())), + ); + } +} + +fn check_running(process: &Process) -> Result<()> { + if process + .has_exited() + .map_err(|error| Error::io("cannot check the endpoint's VM process").with_source(error))? + { + return Err(exited()); + } + Ok(()) +} + +fn exited() -> Error { + let message = "the recorded VM process exited"; + Error::exited(message).with_source(io::Error::new(io::ErrorKind::NotFound, message)) +} + +fn io_kind(error: &io::Error) -> ErrorKind { + match error.kind() { + io::ErrorKind::TimedOut => ErrorKind::TimedOut, + io::ErrorKind::BrokenPipe + | io::ErrorKind::ConnectionAborted + | io::ErrorKind::ConnectionRefused + | io::ErrorKind::ConnectionReset + | io::ErrorKind::UnexpectedEof + | io::ErrorKind::WriteZero => ErrorKind::Protocol, + _ => ErrorKind::Io, + } +} + +/// Most bytes that one read from the transport takes. +const READ_CHUNK: usize = 16 * 1024; + +/// Retained cause details for failures of a [`HostConnection`]. +/// +/// Connection methods return the crate's [`Error`]. I/O and protocol failures keep their +/// original errors as sources; other failures keep these details as their source. +#[derive(Debug)] +#[non_exhaustive] +pub enum HostConnectionError { + /// The deadline passed. A receive that times out keeps what arrived, and the connection stays + /// usable, as it does after a send that times out before it wrote anything. + TimedOut, + /// The broker closed the stream at a record boundary. + Closed, + /// The broker refused the host's attachment. + Rejected(ErrorCode), + /// The broker reset the session, which ended the host's attachment. The identity is the + /// session's new one. + Reset(Identity), + /// The broker violated the protocol, or the stream ended inside a record. + Protocol(SessionError), + /// The transport failed. + Io(io::Error), + /// An earlier failure ended the connection. + Ended, +} + +impl fmt::Display for HostConnectionError { + fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result { + match self { + Self::TimedOut => formatter.write_str("the NVXS operation timed out"), + Self::Closed => formatter.write_str("the broker closed the NVXS session"), + Self::Rejected(code) => { + write!(formatter, "the broker refused the NVXS attachment: {code}") + } + Self::Reset(identity) => write!( + formatter, + "the broker reset the NVXS session to epoch {}", + identity.epoch + ), + Self::Protocol(error) => error.fmt(formatter), + Self::Io(error) => write!(formatter, "NVXS I/O failed: {error}"), + Self::Ended => formatter.write_str("an earlier failure ended the NVXS connection"), + } + } +} + +impl std::error::Error for HostConnectionError { + fn source(&self) -> Option<&(dyn std::error::Error + 'static)> { + match self { + Self::Protocol(error) => Some(error), + Self::Io(error) => Some(error), + _ => None, + } + } +} + +impl From for Error { + fn from(error: HostConnectionError) -> Self { + let kind = match &error { + HostConnectionError::TimedOut => ErrorKind::TimedOut, + HostConnectionError::Io(error) => io_kind(error), + _ => ErrorKind::Protocol, + }; + let result = Self::new(kind, error.to_string()); + match error { + HostConnectionError::Protocol(source) => result.with_source(source), + HostConnectionError::Io(source) => result.with_source(source), + source => result.with_source(source), + } + } +} + +/// The host's end of an NVXS session over a platform transport. +pub struct HostConnection { + /// Transport connected to the broker's host endpoint. + transport: Box, + /// The host's role of the session. + session: HostSession, + /// Identity that the broker attached the host to. + identity: Identity, + /// Bytes read from the transport. + input: Box<[u8]>, + /// Start of the bytes in `input` that the session has not consumed. + start: usize, + /// End of the bytes in `input`. + end: usize, + /// Records encoded for the transport. + output: Vec, + /// Whether a failure ended the connection. + ended: bool, +} + +impl HostConnection { + /// Attaches the host to the broker over `transport` with the launch `capability`, and waits + /// until `deadline` while the broker asks the host to wait for the guest. + pub fn attach( + transport: Box, + capability: &[u8; CAPABILITY_LEN], + deadline: Instant, + ) -> Result { + let mut connection = Self { + transport, + session: HostSession::new(), + // The broker's READY replaces this before the caller sees the connection. + identity: Identity { + instance_id: [0; 16], + epoch: 0, + }, + input: vec![0; READ_CHUNK].into_boxed_slice(), + start: 0, + end: 0, + output: Vec::new(), + ended: false, + }; + connection + .session + .attach(capability, &mut connection.output) + .map_err(HostConnectionError::Protocol)?; + let timeout = remaining(Some(deadline))?; + connection.write(timeout)?; + loop { + match connection.next_event(Some(deadline))? { + HostEvent::Wait => {} + HostEvent::Attached(identity) => { + connection.identity = identity; + return Ok(connection); + } + HostEvent::Rejected(code) => return Err(HostConnectionError::Rejected(code).into()), + // The session reports these only once the broker attached the host. + HostEvent::Data(_) => return Err(unexpected(RecordType::Data).into()), + HostEvent::Reset(_) => return Err(unexpected(RecordType::Reset).into()), + } + } + } + + /// Returns the identity that the broker attached the host to. + pub fn identity(&self) -> Identity { + self.identity + } + + /// Returns the host's role of the session. + pub fn session(&self) -> &HostSession { + &self.session + } + + /// Sends `data` to the guest by `deadline`, or without a time limit, in as many DATA records + /// as it takes, of at most [`OUTER_MAX_PAYLOAD`](crate::protocol::OUTER_MAX_PAYLOAD) bytes + /// each. + /// + /// Data that fits one record travels in one, so an application protocol that needs one frame + /// per record sends frames that fit. Empty data sends nothing. + pub fn send(&mut self, data: &[u8], deadline: Option) -> Result<()> { + self.check()?; + let mut rest = data; + let mut wrote = false; + while !rest.is_empty() { + let timeout = match remaining(deadline) { + Ok(timeout) => timeout, + Err(error) if wrote => return Err(self.end(error).into()), + Err(error) => return Err(error.into()), + }; + self.output.clear(); + let taken = self + .session + .send(rest, &mut self.output) + .map_err(|error| self.end(HostConnectionError::Protocol(error)))?; + self.write(timeout)?; + wrote = true; + rest = &rest[taken..]; + } + Ok(()) + } + + /// Returns the payload of the next DATA record from the guest, waiting until `deadline`, or + /// without a time limit. + pub fn receive(&mut self, deadline: Option) -> Result> { + self.check()?; + let error = match self.next_event(deadline)? { + HostEvent::Data(payload) => return Ok(payload), + HostEvent::Reset(identity) => HostConnectionError::Reset(identity), + // The session reports these only while the broker attaches the host. + HostEvent::Wait => unexpected(RecordType::Wait), + HostEvent::Attached(_) => unexpected(RecordType::Ready), + HostEvent::Rejected(_) => unexpected(RecordType::Error), + }; + Err(self.end(error).into()) + } + + /// Returns the session's next event, reading from the transport until a record completes. + fn next_event(&mut self, deadline: Option) -> Result { + loop { + if self.start < self.end { + let (consumed, event) = self + .session + .receive(&self.input[self.start..self.end]) + .map_err(|error| self.end(HostConnectionError::Protocol(error)))?; + self.start += consumed; + match event { + Some(event) => return Ok(event), + None => continue, + } + } + let timeout = remaining(deadline)?; + match self.transport.read(&mut self.input, timeout) { + Ok(0) => { + let error = match self.session.disconnect() { + Ok(()) => HostConnectionError::Closed, + Err(error) => HostConnectionError::Protocol(error), + }; + return Err(self.end(error)); + } + Ok(count) => { + self.start = 0; + self.end = count.min(self.input.len()); + } + Err(error) if error.kind() == io::ErrorKind::Interrupted => {} + Err(error) if error.kind() == io::ErrorKind::TimedOut => { + return Err(HostConnectionError::Io(error)); + } + Err(error) => return Err(self.end(HostConnectionError::Io(error))), + } + } + } + + /// Writes the encoded records within `timeout`, ending the connection if that fails. + fn write(&mut self, timeout: Option) -> Result<(), HostConnectionError> { + self.transport + .write_all(&self.output, timeout) + .map_err(|error| self.end(HostConnectionError::Io(error))) + } + + /// Fails if an earlier failure ended the connection. + fn check(&self) -> Result<(), HostConnectionError> { + if self.ended { + Err(HostConnectionError::Ended) + } else { + Ok(()) + } + } + + /// Ends the connection with `error`, and returns it. + fn end(&mut self, error: HostConnectionError) -> HostConnectionError { + self.ended = true; + error + } +} + +/// Converts a deadline into the time left, failing once it has passed. +fn remaining(deadline: Option) -> Result, HostConnectionError> { + match deadline { + None => Ok(None), + Some(deadline) => deadline + .checked_duration_since(Instant::now()) + .filter(|remaining| !remaining.is_zero()) + .map(Some) + .ok_or(HostConnectionError::TimedOut), + } +} + +/// Returns the failure for an event that the session reports in no state that the connection +/// calls it in. +fn unexpected(record_type: RecordType) -> HostConnectionError { + HostConnectionError::Protocol(SessionError::Unexpected(record_type)) +} + +#[cfg(test)] +mod tests; diff --git a/nvx-rs/src/host/transport/tests.rs b/nvx-rs/src/host/transport/tests.rs new file mode 100644 index 000000000..ffefab9a5 --- /dev/null +++ b/nvx-rs/src/host/transport/tests.rs @@ -0,0 +1,268 @@ +#![cfg(any(target_os = "linux", windows))] + +use std::error::Error as _; +use std::io::{Read, Write}; +use std::process::{Child, Command, Stdio}; + +use super::*; +use crate::host::process::BootId; + +#[cfg(target_os = "linux")] +type Server = std::os::unix::net::UnixListener; +#[cfg(windows)] +type Server = std::fs::File; + +fn reference(pid: u32) -> ProcessRef { + ProcessRef { + pid, + start_time: platform::process_start_time(pid).unwrap().unwrap(), + boot: BootId::current().unwrap(), + } +} + +fn endpoint(directory: &tempfile::TempDir) -> String { + #[cfg(target_os = "linux")] + { + directory + .path() + .join("control.sock") + .to_str() + .unwrap() + .to_owned() + } + #[cfg(windows)] + { + let _ = directory; + format!( + r"\\.\pipe\nvx-verified-{}-{}", + std::process::id(), + crate::host::store::LaunchId::generate().unwrap().as_str() + ) + } +} + +fn bind(endpoint: &str) -> Server { + #[cfg(target_os = "linux")] + { + Server::bind(endpoint).unwrap() + } + #[cfg(windows)] + { + platform::tests::control_server(endpoint) + } +} + +fn exchange(server: Server) { + #[cfg(target_os = "linux")] + let mut peer = server.accept().unwrap().0; + #[cfg(windows)] + let mut peer = { + platform::tests::accept_control(&server); + server + }; + let mut request = [0; 7]; + peer.read_exact(&mut request).unwrap(); + assert_eq!(&request, b"request"); + peer.write_all(b"response").unwrap(); +} + +struct Sleeper(Child); + +impl Sleeper { + fn new() -> Self { + let mut command = if cfg!(windows) { + let mut command = Command::new("ping"); + command.args(["-n", "60", "127.0.0.1"]); + command + } else { + let mut command = Command::new("sleep"); + command.arg("60"); + command + }; + command + .stdin(Stdio::null()) + .stdout(Stdio::null()) + .stderr(Stdio::null()); + Self(crate::host::tests::spawn_locked(&mut command).unwrap()) + } +} + +impl Drop for Sleeper { + fn drop(&mut self) { + if self.0.try_wait().unwrap().is_none() { + self.0.kill().unwrap(); + self.0.wait().unwrap(); + } + } +} + +#[test] +fn verified_and_legacy_connections_wait_for_the_owned_endpoint_and_keep_native_io() { + let process = reference(std::process::id()); + for legacy in [false, true] { + let directory = tempfile::tempdir().unwrap(); + let endpoint = endpoint(&directory); + let server_endpoint = endpoint.clone(); + let server = thread::spawn(move || { + thread::sleep(Duration::from_millis(50)); + exchange(bind(&server_endpoint)); + }); + let deadline = Instant::now() + Duration::from_secs(10); + let mut transport = if legacy { + crate::host::session::connect(&endpoint, process.pid, process.start_time, deadline) + .unwrap() + } else { + connect_verified(&endpoint, &process, deadline).unwrap() + }; + assert!(transport.try_clone_handle().unwrap().is_some()); + #[cfg(target_os = "linux")] + assert_eq!(transport.unread().unwrap(), Some(0)); + #[cfg(windows)] + assert_eq!(transport.unread().unwrap(), None); + transport + .write_all(b"request", Some(Duration::from_secs(10))) + .unwrap(); + let mut response = [0; 8]; + let mut read = 0; + while read < response.len() { + let count = transport + .read(&mut response[read..], Some(Duration::from_secs(10))) + .unwrap(); + assert_ne!(count, 0); + read += count; + } + assert_eq!(&response, b"response"); + server.join().unwrap(); + #[cfg(target_os = "linux")] + assert_eq!(transport.unread().unwrap(), Some(0)); + } +} + +#[test] +fn another_process_serving_the_endpoint_is_refused() { + let vm = Sleeper::new(); + let process = reference(vm.0.id()); + for legacy in [false, true] { + let directory = tempfile::tempdir().unwrap(); + let endpoint = endpoint(&directory); + let _server = bind(&endpoint); + let deadline = Instant::now() + Duration::from_secs(10); + if legacy { + let error = + crate::host::session::connect(&endpoint, process.pid, process.start_time, deadline) + .err() + .unwrap(); + assert!(matches!( + error, + crate::host::session::SessionError::Refused(_) + )); + } else { + let error = connect_verified(&endpoint, &process, deadline) + .err() + .unwrap(); + assert_eq!(error.kind(), ErrorKind::Protocol); + let source = error.source().unwrap().downcast_ref::().unwrap(); + assert_eq!(source.kind(), io::ErrorKind::PermissionDenied); + assert!(source.to_string().contains(&std::process::id().to_string())); + assert!(source.to_string().contains(&process.pid.to_string())); + } + assert!( + !Process::open(&process) + .unwrap() + .unwrap() + .has_exited() + .unwrap() + ); + } +} + +#[test] +fn endpoint_timeout_keeps_the_actual_unavailable_endpoint_error() { + let directory = tempfile::tempdir().unwrap(); + let endpoint = endpoint(&directory); + let process = reference(std::process::id()); + let error = connect_verified( + &endpoint, + &process, + Instant::now() + Duration::from_millis(80), + ) + .err() + .unwrap(); + assert_eq!(error.kind(), ErrorKind::TimedOut); + let source = error.source().unwrap().downcast_ref::().unwrap(); + assert_eq!(source.kind(), io::ErrorKind::NotFound); + assert!(source.raw_os_error().is_some()); +} + +#[test] +fn the_recorded_vm_is_checked_even_when_its_pid_serves_the_endpoint() { + let directory = tempfile::tempdir().unwrap(); + let endpoint = endpoint(&directory); + let _server = bind(&endpoint); + let mut process = reference(std::process::id()); + process.start_time += 1; + let error = connect_verified( + &endpoint, + &process, + Instant::now() + Duration::from_secs(10), + ) + .err() + .unwrap(); + assert_eq!(error.kind(), ErrorKind::Exited); + assert_eq!( + error + .source() + .unwrap() + .downcast_ref::() + .unwrap() + .kind(), + io::ErrorKind::NotFound, + ); +} + +#[test] +fn an_owned_vm_that_exited_is_reported_before_connecting() { + let directory = tempfile::tempdir().unwrap(); + let endpoint = endpoint(&directory); + let mut vm = Sleeper::new(); + let process = reference(vm.0.id()); + vm.0.kill().unwrap(); + vm.0.wait().unwrap(); + let error = connect_verified( + &endpoint, + &process, + Instant::now() + Duration::from_secs(10), + ) + .err() + .unwrap(); + assert_eq!(error.kind(), ErrorKind::Exited); + assert!(error.source().is_some()); +} + +#[test] +fn an_indeterminate_process_reference_retains_its_io_failure() { + let directory = tempfile::tempdir().unwrap(); + let endpoint = endpoint(&directory); + let process = ProcessRef { + pid: 0, + start_time: 0, + boot: BootId::current().unwrap(), + }; + let error = connect_verified( + &endpoint, + &process, + Instant::now() + Duration::from_secs(10), + ) + .err() + .unwrap(); + assert_eq!(error.kind(), ErrorKind::Io); + assert_eq!( + error + .source() + .unwrap() + .downcast_ref::() + .unwrap() + .kind(), + io::ErrorKind::InvalidInput, + ); +} diff --git a/nvx-rs/tests/nvxs/host.rs b/nvx-rs/tests/nvxs/host.rs index cda0566bc..435b63b90 100644 --- a/nvx-rs/tests/nvxs/host.rs +++ b/nvx-rs/tests/nvxs/host.rs @@ -3,12 +3,12 @@ //! that end the connection. use std::collections::VecDeque; +use std::error::Error as _; use std::io; use std::sync::{Arc, Mutex}; use std::time::{Duration, Instant}; -use nvx::host::nvxs::{HostConnection, HostConnectionError}; -use nvx::host::platform::Transport; +use nvx::host::transport::{HostConnection, HostConnectionError, Transport}; use nvx::protocol::nvxs::{ ErrorCode, GuestEvent, GuestSession, HostState, Identity, RecordParser, RecordType, SessionError, @@ -16,6 +16,7 @@ use nvx::protocol::nvxs::{ use nvx::protocol::{ CAPABILITY_LEN, OUTER_HEADER_LEN, OUTER_MAX_PAYLOAD, OUTER_MIN_RECEIVE_WINDOW, }; +use nvx::{Error, ErrorKind}; use crate::broker::{Broker, CAPABILITY, IDENTITY, record}; use crate::{Reassembler, framed, messages}; @@ -24,6 +25,10 @@ fn deadline() -> Instant { Instant::now() + Duration::from_secs(10) } +fn cause(error: &Error) -> Option<&T> { + error.source().and_then(|cause| cause.downcast_ref()) +} + /// The broker and a guest whose application echoes each message. struct World { /// Broker between the host and the guest. @@ -162,7 +167,7 @@ fn the_host_connection_waits_attaches_and_exchanges_messages() { assert_eq!(world.lock().unwrap().broker.held(), 0); // Nothing more arrives. let timed_out = connection.receive(Some(deadline())); - assert!(matches!(timed_out, Err(HostConnectionError::TimedOut))); + assert!(matches!(timed_out, Err(error) if error.kind() == ErrorKind::TimedOut)); } } @@ -251,7 +256,7 @@ fn a_receive_that_times_out_keeps_the_partial_record() { let mut connection = attach(transport); let first = connection.receive(Some(deadline())); assert!( - matches!(first, Err(HostConnectionError::TimedOut)), + matches!(&first, Err(error) if error.kind() == ErrorKind::TimedOut), "{first:?}" ); assert_eq!( @@ -260,23 +265,24 @@ fn a_receive_that_times_out_keeps_the_partial_record() { ); let again = connection.receive(None); assert!( - matches!(again, Err(HostConnectionError::TimedOut)), + matches!(&again, Err(error) if error.kind() == ErrorKind::TimedOut), "{again:?}" ); // The stream then ends at a record boundary. let closed = connection.receive(None); assert!( - matches!(closed, Err(HostConnectionError::Closed)), + matches!(&closed, Err(error) if error.kind() == ErrorKind::Protocol + && matches!(cause(error), Some(HostConnectionError::Closed))), "{closed:?}" ); let ended = connection.receive(None); assert!( - matches!(ended, Err(HostConnectionError::Ended)), + matches!(&ended, Err(error) if matches!(cause(error), Some(HostConnectionError::Ended))), "{ended:?}" ); let ended = connection.send(b"x", None); assert!( - matches!(ended, Err(HostConnectionError::Ended)), + matches!(&ended, Err(error) if matches!(cause(error), Some(HostConnectionError::Ended))), "{ended:?}" ); } @@ -287,8 +293,9 @@ fn the_host_connection_reports_refusals_and_closed_streams_while_attaching() { let (transport, written) = scripted(vec![Ok(refusal)]); let result = HostConnection::attach(transport, &CAPABILITY, deadline()); let refused = matches!( - result, - Err(HostConnectionError::Rejected(ErrorCode::Authentication)) + &result, + Err(error) if error.kind() == ErrorKind::Protocol + && matches!(cause(error), Some(HostConnectionError::Rejected(ErrorCode::Authentication))) ); assert!(refused, "{:?}", result.err()); // The attachment carried the capability. @@ -304,7 +311,7 @@ fn the_host_connection_reports_refusals_and_closed_streams_while_attaching() { let (transport, _) = scripted(Vec::new()); let result = HostConnection::attach(transport, &CAPABILITY, deadline()); assert!( - matches!(result, Err(HostConnectionError::Closed)), + matches!(&result, Err(error) if matches!(cause(error), Some(HostConnectionError::Closed))), "{:?}", result.err() ); @@ -316,7 +323,7 @@ fn the_host_connection_reports_refusals_and_closed_streams_while_attaching() { let (transport, written) = scripted(vec![ready()]); let result = HostConnection::attach(transport, &CAPABILITY, Instant::now()); assert!( - matches!(result, Err(HostConnectionError::TimedOut)), + matches!(&result, Err(error) if error.kind() == ErrorKind::TimedOut), "{:?}", result.err() ); @@ -326,7 +333,7 @@ fn the_host_connection_reports_refusals_and_closed_streams_while_attaching() { #[test] fn resets_truncations_violations_and_transport_failures_end_the_connection() { /// Checks the failure of a case. - type Check = fn(&HostConnectionError) -> bool; + type Check = fn(&Error) -> bool; let next = Identity { epoch: 2, ..IDENTITY @@ -335,27 +342,28 @@ fn resets_truncations_violations_and_transport_failures_end_the_connection() { let reset_by_peer = Err(io::ErrorKind::ConnectionReset.into()); let cases: [(io::Result>, Check); 4] = [ (Ok(record(RecordType::Reset, next, 1, &[])), |error| { - matches!(error, HostConnectionError::Reset(Identity { epoch: 2, .. })) + error.kind() == ErrorKind::Protocol + && matches!( + cause(error), + Some(HostConnectionError::Reset(Identity { epoch: 2, .. })) + ) }), (Ok(data[..20].to_vec()), |error| { - matches!( - error, - HostConnectionError::Protocol(SessionError::Truncated { buffered: 20 }) - ) + matches!(cause(error), Some(SessionError::Truncated { buffered: 20 })) }), (Ok(record(RecordType::Data, IDENTITY, 2, b"x")), |error| { matches!( - error, - HostConnectionError::Protocol(SessionError::Sequence { + cause(error), + Some(SessionError::Sequence { expected: 1, actual: 2 }) ) }), - ( - reset_by_peer, - |error| matches!(error, HostConnectionError::Io(error) if error.kind() == io::ErrorKind::ConnectionReset), - ), + (reset_by_peer, |error| { + error.kind() == ErrorKind::Protocol + && matches!(cause::(error), Some(error) if error.kind() == io::ErrorKind::ConnectionReset) + }), ]; for (read, expected) in cases { let (transport, _) = scripted(vec![ready(), read]); @@ -364,12 +372,12 @@ fn resets_truncations_violations_and_transport_failures_end_the_connection() { assert!(expected(&error), "{error:?}"); let ended = connection.receive(Some(deadline())); assert!( - matches!(ended, Err(HostConnectionError::Ended)), + matches!(&ended, Err(error) if matches!(cause(error), Some(HostConnectionError::Ended))), "{ended:?}" ); let ended = connection.send(b"x", Some(deadline())); assert!( - matches!(ended, Err(HostConnectionError::Ended)), + matches!(&ended, Err(error) if matches!(cause(error), Some(HostConnectionError::Ended))), "{ended:?}" ); } @@ -381,7 +389,7 @@ fn a_send_that_times_out_before_writing_leaves_the_connection_usable() { let mut connection = attach(transport); let result = connection.send(b"late", Some(Instant::now())); assert!( - matches!(result, Err(HostConnectionError::TimedOut)), + matches!(&result, Err(error) if error.kind() == ErrorKind::TimedOut), "{result:?}" ); // The DATA that follows takes the first sequence number. @@ -394,6 +402,67 @@ fn a_send_that_times_out_before_writing_leaves_the_connection_usable() { assert_eq!((data.sequence(), data.payload()), (0, &b"on time"[..])); } +#[test] +fn transport_timeouts_keep_the_original_read_and_write_causes() { + let (transport, _) = scripted(vec![ + ready(), + Err(io::Error::new( + io::ErrorKind::TimedOut, + "the original read failure", + )), + Ok(record(RecordType::Data, IDENTITY, 1, b"after timeout")), + ]); + let mut connection = attach(transport); + let error = connection.receive(Some(deadline())).unwrap_err(); + assert_eq!(error.kind(), ErrorKind::TimedOut); + assert_eq!( + cause::(&error).unwrap().to_string(), + "the original read failure" + ); + assert_eq!( + connection.receive(Some(deadline())).unwrap(), + b"after timeout" + ); + + let on_write: OnWrite = Box::new(|index| match index { + 0 => Ok(()), + _ => Err(io::Error::new( + io::ErrorKind::TimedOut, + "the original write failure", + )), + }); + let (transport, _) = scripted_writes(vec![ready()], on_write); + let mut connection = attach(transport); + let error = connection.send(b"x", Some(deadline())).unwrap_err(); + assert_eq!(error.kind(), ErrorKind::TimedOut); + assert_eq!( + cause::(&error).unwrap().to_string(), + "the original write failure" + ); + let error = connection.receive(None).unwrap_err(); + assert!(matches!(cause(&error), Some(HostConnectionError::Ended))); +} + +#[test] +fn other_transport_failures_keep_their_original_io_source() { + let (transport, _) = scripted(vec![ + ready(), + Err(io::Error::new( + io::ErrorKind::Other, + "the original transport failure", + )), + ]); + let mut connection = attach(transport); + let error = connection.receive(Some(deadline())).unwrap_err(); + assert_eq!(error.kind(), ErrorKind::Io); + assert_eq!( + cause::(&error).unwrap().to_string(), + "the original transport failure" + ); + let error = connection.send(b"x", None).unwrap_err(); + assert!(matches!(cause(&error), Some(HostConnectionError::Ended))); +} + #[test] fn a_send_that_fails_after_writing_ends_the_connection() { // A write that times out may have written part of a record. @@ -405,12 +474,12 @@ fn a_send_that_fails_after_writing_ends_the_connection() { let mut connection = attach(transport); let result = connection.send(b"x", Some(deadline())); assert!( - matches!(result, Err(HostConnectionError::TimedOut)), + matches!(&result, Err(error) if error.kind() == ErrorKind::TimedOut), "{result:?}" ); let ended = connection.receive(Some(deadline())); assert!( - matches!(ended, Err(HostConnectionError::Ended)), + matches!(&ended, Err(error) if matches!(cause(error), Some(HostConnectionError::Ended))), "{ended:?}" ); @@ -422,11 +491,12 @@ fn a_send_that_fails_after_writing_ends_the_connection() { let (transport, _) = scripted_writes(vec![ready()], on_write); let mut connection = attach(transport); let result = connection.send(b"x", None); - let broken = matches!(&result, Err(HostConnectionError::Io(error)) if error.kind() == io::ErrorKind::BrokenPipe); + let broken = matches!(&result, Err(error) if error.kind() == ErrorKind::Protocol + && matches!(cause::(error), Some(error) if error.kind() == io::ErrorKind::BrokenPipe)); assert!(broken, "{result:?}"); let ended = connection.receive(None); assert!( - matches!(ended, Err(HostConnectionError::Ended)), + matches!(&ended, Err(error) if matches!(cause(error), Some(HostConnectionError::Ended))), "{ended:?}" ); @@ -447,12 +517,12 @@ fn a_send_that_fails_after_writing_ends_the_connection() { *message_deadline.lock().unwrap() = Some(deadline); let result = connection.send(&vec![0; 100_000], Some(deadline)); assert!( - matches!(result, Err(HostConnectionError::TimedOut)), + matches!(&result, Err(error) if error.kind() == ErrorKind::TimedOut), "{result:?}" ); let ended = connection.send(b"x", None); assert!( - matches!(ended, Err(HostConnectionError::Ended)), + matches!(&ended, Err(error) if matches!(cause(error), Some(HostConnectionError::Ended))), "{ended:?}" ); // The attachment and the message's first record went out. diff --git a/tests/runtime-adapter/src/bin/runtime-adapter-host.rs b/tests/runtime-adapter/src/bin/runtime-adapter-host.rs index b6a88e20a..ca3713e1b 100644 --- a/tests/runtime-adapter/src/bin/runtime-adapter-host.rs +++ b/tests/runtime-adapter/src/bin/runtime-adapter-host.rs @@ -446,13 +446,13 @@ use std::path::{Path, PathBuf}; use std::process::ExitCode; use std::time::{Duration, Instant}; +use nvx::ErrorKind; use nvx::activation::{MachineIdMode, StoreGeneration}; use nvx::exec::{Cause, Key, Limits, Outcome, Status, Stream, Timers, Window}; use nvx::host::artifacts::{Artifacts, absolute}; use nvx::host::filesystem::ExportPlan; use nvx::host::instance::generate_capability; use nvx::host::launch::{DEFAULT_MEMORY_MIB, LaunchSpec}; -use nvx::host::nvxs::HostConnectionError; use nvx::host::report::{self, Report}; use nvx::host::sandbox::SandboxPlan; use nvx::host::session; @@ -2784,7 +2784,8 @@ fn unanswered( deadline: Instant, ) -> (Evidence, nvx::Error) { const PROBLEM: &str = "the guest agent did not answer once the clone's sandbox started"; - let timed_out = matches!(error, ClientError::Session(HostConnectionError::TimedOut)); + let timed_out = + matches!(&error, ClientError::Session(error) if error.kind() == ErrorKind::TimedOut); if timed_out || Instant::now() >= deadline { return ( Evidence::Timeout, diff --git a/tests/runtime-adapter/src/host.rs b/tests/runtime-adapter/src/host.rs index 9c601e917..28db62a7c 100644 --- a/tests/runtime-adapter/src/host.rs +++ b/tests/runtime-adapter/src/host.rs @@ -43,15 +43,16 @@ mod executions; mod map_reply; mod order_error; +use std::error::Error as _; use std::fmt; use std::time::Instant; use nvx::activation::ReadyActivation; -use nvx::host::nvxs::{HostConnection, HostConnectionError}; -use nvx::host::platform::Transport; use nvx::host::template::CloneControl; +use nvx::host::transport::{HostConnection, HostConnectionError, Transport}; use nvx::protocol::nvxs::Identity; use nvx::protocol::{CAPABILITY_LEN, HostMap}; +use nvx::{Error, ErrorKind}; use crate::frame::{BodyError, Decoder, FrameError, Message}; use crate::probe::{Facts, Probe, ProbeReply}; @@ -66,7 +67,7 @@ pub use order_error::OrderError; #[non_exhaustive] pub enum ClientError { /// The control session failed, or its deadline passed. - Session(HostConnectionError), + Session(Error), /// A malformed message: one that the guest agent sent, or one that the host program asked to /// send, which the client did not send. Frame(FrameError), @@ -119,8 +120,8 @@ impl std::error::Error for ClientError { } } -impl From for ClientError { - fn from(error: HostConnectionError) -> Self { +impl From for ClientError { + fn from(error: Error) -> Self { Self::Session(error) } } @@ -146,9 +147,9 @@ impl From for ClientError { /// the body. /// /// A reset of the session ends the attachment, and the partial message with it: the session's next -/// epoch starts with a new message, in a new attachment. `nvx::host::nvxs` reports a reset that -/// the broker announces as [`HostConnectionError::Reset`], and the end of the stream, with which -/// OpenVMM's broker ends an attachment instead, as [`HostConnectionError::Closed`]. +/// epoch starts with a new message, in a new attachment. `nvx::host::transport` reports both as +/// [`ErrorKind::Protocol`], retaining [`HostConnectionError::Reset`] or +/// [`HostConnectionError::Closed`] as the cause. /// /// A failure ends the client, and every later call fails with [`ClientError::Ended`]: a failure of /// the session, a send that fails or times out, because the session cannot resume a message that @@ -245,8 +246,8 @@ impl Client { // The decoder took the whole payload, so the next one replaces it. self.payload = match self.connection.receive(deadline) { Ok(payload) => payload, - Err(HostConnectionError::TimedOut) => { - return Err(HostConnectionError::TimedOut.into()); + Err(error) if error.kind() == ErrorKind::TimedOut => { + return Err(error.into()); } Err(error) => return Err(self.fail(error.into())), }; @@ -606,13 +607,17 @@ impl CloneControl for CloneOrder<'_> { } fn lost(&self, error: &OrderError) -> bool { + let OrderError::Client(ClientError::Session(error)) = error else { + return false; + }; matches!( - error, - OrderError::Client(ClientError::Session( - HostConnectionError::Reset(_) - | HostConnectionError::Closed - | HostConnectionError::Io(_) - )) - ) + error + .source() + .and_then(|cause| cause.downcast_ref::()), + Some(HostConnectionError::Reset(_) | HostConnectionError::Closed) + ) || (error.kind() != ErrorKind::TimedOut + && error + .source() + .is_some_and(|cause| cause.is::())) } } diff --git a/tests/runtime-adapter/src/host/executions.rs b/tests/runtime-adapter/src/host/executions.rs index 0a25a5c62..aa11e9a19 100644 --- a/tests/runtime-adapter/src/host/executions.rs +++ b/tests/runtime-adapter/src/host/executions.rs @@ -5,8 +5,8 @@ use std::collections::BTreeMap; use std::fmt; use std::time::Instant; +use nvx::ErrorKind; use nvx::exec::{Key, Timers, Unit, Violation}; -use nvx::host::nvxs::HostConnectionError; use super::{Client, ClientError}; use crate::exec::{GuestMessage, HostExecution, MAX_EXECUTIONS, Start, name}; @@ -226,7 +226,7 @@ impl Executions { .fold(deadline, Instant::min); match client.receive(Some(wake)) { Ok(message) => self.dispatch(&message, Instant::now())?, - Err(ClientError::Session(HostConnectionError::TimedOut)) => {} + Err(ClientError::Session(error)) if error.kind() == ErrorKind::TimedOut => {} Err(error) => return Err(error.into()), } } @@ -290,7 +290,9 @@ impl Executions { /// once the send's deadline, the run's, passed. fn failed(error: ClientError) -> ExecutionsError { match error { - ClientError::Session(HostConnectionError::TimedOut) => ExecutionsError::TimedOut, + ClientError::Session(error) if error.kind() == ErrorKind::TimedOut => { + ExecutionsError::TimedOut + } error => ExecutionsError::Client(error), } } diff --git a/tests/runtime-adapter/tests/host_session.rs b/tests/runtime-adapter/tests/host_session.rs index e52e2a68f..83d5e7a47 100644 --- a/tests/runtime-adapter/tests/host_session.rs +++ b/tests/runtime-adapter/tests/host_session.rs @@ -22,6 +22,7 @@ #![cfg(feature = "host")] use std::collections::{BTreeMap, VecDeque}; +use std::error::Error as _; use std::io; use std::sync::{Arc, Condvar, Mutex, MutexGuard}; use std::thread::{self, JoinHandle}; @@ -42,6 +43,7 @@ use nvx::protocol::{ CAPABILITY_LEN, HostMap, MAX_HOST_MAPPINGS, MAX_MAP_PATH_BYTES, OUTER_HEADER_LEN, OUTER_MAX_PAYLOAD, OUTER_MIN_RECEIVE_WINDOW, }; +use nvx::{Error, ErrorKind}; use runtime_adapter::GUEST_BUILD; use runtime_adapter::exec::{GuestMessage, HostMessage, MAX_EXECUTIONS, Start}; use runtime_adapter::frame::{ @@ -56,6 +58,10 @@ use runtime_adapter::root::{ MapFailure, Ready, RootConfig, RootFailure, RootReply, RootStage, decode_maps, encode_maps, }; +fn cause(error: &Error) -> Option<&T> { + error.source().and_then(|cause| cause.downcast_ref()) +} + /// Launch capability that the test broker accepts. const CAPABILITY: [u8; CAPABILITY_LEN] = [0xa5; CAPABILITY_LEN]; @@ -668,8 +674,8 @@ fn the_client_reassembles_messages_from_partial_and_shared_records() { // Nothing more arrives. let timed_out = client.receive(Some(Instant::now() + SHORT)); let nothing = matches!( - timed_out, - Err(ClientError::Session(HostConnectionError::TimedOut)) + &timed_out, + Err(ClientError::Session(error)) if error.kind() == ErrorKind::TimedOut ); assert!(nothing, "{timed_out:?}"); } @@ -693,8 +699,8 @@ fn a_receive_that_times_out_keeps_the_partial_message() { broker.write(&rest[..20]); let timed_out = client.receive(Some(Instant::now() + SHORT)); let kept = matches!( - timed_out, - Err(ClientError::Session(HostConnectionError::TimedOut)) + &timed_out, + Err(ClientError::Session(error)) if error.kind() == ErrorKind::TimedOut ); assert!(kept, "{timed_out:?}"); @@ -720,8 +726,8 @@ fn a_send_that_fails_or_times_out_ends_the_client_with_what_it_holds() { let echo = Message::Echo(vec![0; 1000]); let timed_out = client.send(&echo, Some(Instant::now() + SHORT)); let late = matches!( - timed_out, - Err(ClientError::Session(HostConnectionError::TimedOut)) + &timed_out, + Err(ClientError::Session(error)) if error.kind() == ErrorKind::TimedOut ); assert!(late, "{timed_out:?}"); assert_ended(&mut client); @@ -734,8 +740,8 @@ fn a_send_that_fails_or_times_out_ends_the_client_with_what_it_holds() { let failed = client.send(&Message::Ping, Some(deadline())); let broken = matches!( &failed, - Err(ClientError::Session(HostConnectionError::Io(error))) - if error.kind() == io::ErrorKind::BrokenPipe + Err(ClientError::Session(error)) if error.kind() == ErrorKind::Protocol + && matches!(cause::(error), Some(error) if error.kind() == io::ErrorKind::BrokenPipe) ); assert!(broken, "{failed:?}"); assert_ended(&mut client); @@ -754,8 +760,9 @@ fn a_reset_or_a_disconnect_in_the_middle_of_a_message_drops_it() { broker.send(RecordType::Reset, &[]); let reset = client.receive(Some(deadline())); let ended = matches!( - reset, - Err(ClientError::Session(HostConnectionError::Reset(identity))) if identity == next + &reset, + Err(ClientError::Session(error)) + if matches!(cause(error), Some(HostConnectionError::Reset(identity)) if *identity == next) ); assert!(ended, "{reset:?}"); assert_ended(&mut client); @@ -773,8 +780,9 @@ fn a_reset_or_a_disconnect_in_the_middle_of_a_message_drops_it() { broker.close(); let closed = client.receive(Some(deadline())); let ended = matches!( - closed, - Err(ClientError::Session(HostConnectionError::Closed)) + &closed, + Err(ClientError::Session(error)) + if matches!(cause(error), Some(HostConnectionError::Closed)) ); assert!(ended, "{closed:?}"); assert_ended(&mut client); @@ -796,10 +804,9 @@ fn a_reset_or_a_disconnect_in_the_middle_of_a_message_drops_it() { broker.close(); let truncated = client.receive(Some(deadline())); let ended = matches!( - truncated, - Err(ClientError::Session(HostConnectionError::Protocol( - SessionError::Truncated { buffered: 30 } - ))) + &truncated, + Err(ClientError::Session(error)) + if matches!(cause(error), Some(SessionError::Truncated { buffered: 30 })) ); assert!(ended, "{truncated:?}"); assert_ended(&mut client); @@ -826,8 +833,8 @@ fn the_client_waits_for_the_guest_and_reports_refusals_while_attaching() { broker.send(RecordType::Wait, &[]); let timed_out = joined(attach(transport, Instant::now() + SHORT)); let late = matches!( - timed_out, - Err(ClientError::Session(HostConnectionError::TimedOut)) + &timed_out, + Err(ClientError::Session(error)) if error.kind() == ErrorKind::TimedOut ); assert!(late, "{timed_out:?}"); @@ -843,10 +850,9 @@ fn the_client_waits_for_the_guest_and_reports_refusals_while_attaching() { broker.send(RecordType::Error, &code); let refused = joined(attaching); let rejected = matches!( - refused, - Err(ClientError::Session(HostConnectionError::Rejected( - ErrorCode::Authentication - ))) + &refused, + Err(ClientError::Session(error)) + if matches!(cause(error), Some(HostConnectionError::Rejected(ErrorCode::Authentication))) ); assert!(rejected, "after {waits} WAIT records: {refused:?}"); } @@ -858,8 +864,9 @@ fn the_client_waits_for_the_guest_and_reports_refusals_while_attaching() { broker.close(); let refused = joined(attaching); let closed = matches!( - refused, - Err(ClientError::Session(HostConnectionError::Closed)) + &refused, + Err(ClientError::Session(error)) + if matches!(cause(error), Some(HostConnectionError::Closed)) ); assert!(closed, "{refused:?}"); } @@ -1016,7 +1023,7 @@ fn the_client_fails_on_records_that_the_protocol_does_not_allow() { let refused = joined(attaching); let failed = matches!( &refused, - Err(ClientError::Session(HostConnectionError::Protocol(error))) if *error == expected + Err(ClientError::Session(error)) if cause::(error) == Some(&expected) ); assert!(failed, "{name}: {refused:?}"); } @@ -1137,7 +1144,7 @@ fn the_client_fails_on_records_that_the_protocol_does_not_allow() { let refused = client.receive(Some(deadline())); let failed = matches!( &refused, - Err(ClientError::Session(HostConnectionError::Protocol(error))) if *error == expected + Err(ClientError::Session(error)) if cause::(error) == Some(&expected) ); assert!(failed, "{name}: {refused:?}"); assert_ended(&mut client); @@ -1421,8 +1428,8 @@ fn requests_check_their_answers_and_end_the_client_when_they_fail() { assert_eq!(broker.messages(1), [Message::Ping]); let (mut client, result) = joined(requesting); let late = matches!( - result, - Err(ClientError::Session(HostConnectionError::TimedOut)) + &result, + Err(ClientError::Session(error)) if error.kind() == ErrorKind::TimedOut ); assert!(late, "{result:?}"); broker.send_data(&Message::Pong.encode().unwrap(), OUTER_MAX_PAYLOAD); @@ -2089,10 +2096,9 @@ fn only_a_session_that_ended_is_lost_to_a_clone_s_order() { ); let error = joined(attaching).unwrap_err(); let rejected = matches!( - error, - OrderError::Client(ClientError::Session(HostConnectionError::Rejected( - ErrorCode::Authentication - ))) + &error, + OrderError::Client(ClientError::Session(error)) + if matches!(cause(error), Some(HostConnectionError::Rejected(ErrorCode::Authentication))) ); assert!(rejected && not_lost(&error), "{error:?}"); @@ -2103,8 +2109,8 @@ fn only_a_session_that_ended_is_lost_to_a_clone_s_order() { thread::spawn(move || clone_order().attach(transport, &CAPABILITY, Instant::now() + SHORT)); let error = joined(attaching).unwrap_err(); let late = matches!( - error, - OrderError::Client(ClientError::Session(HostConnectionError::TimedOut)) + &error, + OrderError::Client(ClientError::Session(error)) if error.kind() == ErrorKind::TimedOut ); assert!(late && not_lost(&error), "{error:?}"); @@ -2139,8 +2145,8 @@ fn only_a_session_that_ended_is_lost_to_a_clone_s_order() { let (mut client, activated) = joined(activation); let error = activated.unwrap_err(); let late = matches!( - error, - OrderError::Client(ClientError::Session(HostConnectionError::TimedOut)) + &error, + OrderError::Client(ClientError::Session(error)) if error.kind() == ErrorKind::TimedOut ); assert!(late && not_lost(&error), "{error:?}"); assert_ended(&mut client); @@ -2566,10 +2572,9 @@ fn the_driver_ends_its_run_on_a_message_that_breaks_the_rules() { broker.close(); let (_, ran) = joined(running); let closed = matches!( - ran, - Err(ExecutionsError::Client(ClientError::Session( - HostConnectionError::Closed - ))) + &ran, + Err(ExecutionsError::Client(ClientError::Session(error))) + if matches!(cause(error), Some(HostConnectionError::Closed)) ); assert!(closed, "{:?}", ran.err()); } From e3b6efefbe90719d9a6edecf6b7f718147d8578c Mon Sep 17 00:00:00 2001 From: Enrique Saurez Date: Sat, 7 Nov 2026 16:07:55 -0800 Subject: [PATCH 2/7] host: retain indeterminate process failures in the connect wrapper Keep process-query permission failures as I/O errors instead of endpoint refusals. Cover verified process exit while waiting for an endpoint and preserve the source of the legacy failure. Document the internal helpers required by the repository lint. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> Copilot-Session: 988ad39e-5a8f-4e76-80b8-ccbe7f1f3c58 --- nvx-rs/src/host/session.rs | 7 ++++++- nvx-rs/src/host/session/tests.rs | 14 ++++++++++++++ nvx-rs/src/host/transport.rs | 3 +++ nvx-rs/src/host/transport/tests.rs | 23 +++++++++++++++++++++++ nvx-rs/tests/nvxs/host.rs | 5 +---- 5 files changed, 47 insertions(+), 5 deletions(-) diff --git a/nvx-rs/src/host/session.rs b/nvx-rs/src/host/session.rs index 410ac830e..b5d388a90 100644 --- a/nvx-rs/src/host/session.rs +++ b/nvx-rs/src/host/session.rs @@ -88,6 +88,7 @@ pub fn connect( transport::connect_verified(endpoint, &reference, deadline).map_err(connect_failure) } +/// Preserves the existing error type for callers of the temporary endpoint wrapper. fn connect_failure(error: Error) -> SessionError { let (kind, message, source) = error.into_parts(); match kind { @@ -98,7 +99,10 @@ fn connect_failure(error: Error) -> SessionError { return SessionError::Protocol(message); }; match source.downcast::() { - Ok(error) if error.kind() == io::ErrorKind::PermissionDenied => { + Ok(error) + if kind == ErrorKind::Protocol + && error.kind() == io::ErrorKind::PermissionDenied => + { SessionError::Refused(*error) } Ok(error) => SessionError::Io(*error), @@ -421,6 +425,7 @@ fn failure(error: Error, stage: Stage) -> SessionError { connection_failure(error, stage) } +/// Maps retained connection cause details to the legacy NVXC failure at `stage`. fn connection_failure(error: HostConnectionError, stage: Stage) -> SessionError { let violation = match error { HostConnectionError::TimedOut => return SessionError::TimedOut, diff --git a/nvx-rs/src/host/session/tests.rs b/nvx-rs/src/host/session/tests.rs index 6e23f915f..2b90794a4 100644 --- a/nvx-rs/src/host/session/tests.rs +++ b/nvx-rs/src/host/session/tests.rs @@ -11,6 +11,20 @@ use crate::protocol::{ const INSTANCE: [u8; 16] = [7; 16]; +#[test] +fn the_legacy_connect_wrapper_keeps_indeterminate_process_failures_as_io() { + let error = Error::io("cannot verify the VM process").with_source(io::Error::new( + io::ErrorKind::PermissionDenied, + "the original process query failure", + )); + let error = connect_failure(error); + let SessionError::Io(source) = error else { + panic!("an indeterminate process check is not an endpoint refusal: {error}"); + }; + assert_eq!(source.kind(), io::ErrorKind::PermissionDenied); + assert_eq!(source.to_string(), "the original process query failure"); +} + /// What a test shares with a [`Scripted`] transport. #[derive(Default)] struct Wire { diff --git a/nvx-rs/src/host/transport.rs b/nvx-rs/src/host/transport.rs index de9e67d56..d27dd6979 100644 --- a/nvx-rs/src/host/transport.rs +++ b/nvx-rs/src/host/transport.rs @@ -132,6 +132,7 @@ pub fn connect_verified( } } +/// Refuses an exited held VM, without relying on its numeric PID. fn check_running(process: &Process) -> Result<()> { if process .has_exited() @@ -142,11 +143,13 @@ fn check_running(process: &Process) -> Result<()> { Ok(()) } +/// Describes the verified absence of the recorded VM process. fn exited() -> Error { let message = "the recorded VM process exited"; Error::exited(message).with_source(io::Error::new(io::ErrorKind::NotFound, message)) } +/// Classifies transport failures separately from broken-session conditions. fn io_kind(error: &io::Error) -> ErrorKind { match error.kind() { io::ErrorKind::TimedOut => ErrorKind::TimedOut, diff --git a/nvx-rs/src/host/transport/tests.rs b/nvx-rs/src/host/transport/tests.rs index ffefab9a5..2136f8fbb 100644 --- a/nvx-rs/src/host/transport/tests.rs +++ b/nvx-rs/src/host/transport/tests.rs @@ -239,6 +239,29 @@ fn an_owned_vm_that_exited_is_reported_before_connecting() { assert!(error.source().is_some()); } +#[test] +fn an_owned_vm_exit_interrupts_waiting_for_its_endpoint() { + let directory = tempfile::tempdir().unwrap(); + let endpoint = endpoint(&directory); + let mut vm = Sleeper::new(); + let process = reference(vm.0.id()); + let ending = thread::spawn(move || { + thread::sleep(Duration::from_millis(30)); + vm.0.kill().unwrap(); + vm.0.wait().unwrap(); + }); + let error = connect_verified( + &endpoint, + &process, + Instant::now() + Duration::from_secs(10), + ) + .err() + .unwrap(); + ending.join().unwrap(); + assert_eq!(error.kind(), ErrorKind::Exited); + assert!(error.source().is_some()); +} + #[test] fn an_indeterminate_process_reference_retains_its_io_failure() { let directory = tempfile::tempdir().unwrap(); diff --git a/nvx-rs/tests/nvxs/host.rs b/nvx-rs/tests/nvxs/host.rs index 435b63b90..0d075030c 100644 --- a/nvx-rs/tests/nvxs/host.rs +++ b/nvx-rs/tests/nvxs/host.rs @@ -447,10 +447,7 @@ fn transport_timeouts_keep_the_original_read_and_write_causes() { fn other_transport_failures_keep_their_original_io_source() { let (transport, _) = scripted(vec![ ready(), - Err(io::Error::new( - io::ErrorKind::Other, - "the original transport failure", - )), + Err(io::Error::other("the original transport failure")), ]); let mut connection = attach(transport); let error = connection.receive(Some(deadline())).unwrap_err(); From 028502ffbfd190f84032d192da694d0e85c9cbf4 Mon Sep 17 00:00:00 2001 From: Enrique Saurez Date: Sat, 7 Nov 2026 16:27:16 -0800 Subject: [PATCH 3/7] host: record the verified transport public API Record the host transport entry point, verified connector and retained NVXS connection on Linux and Windows. The existing platform transport and compatibility paths remain available; no unrelated public surface changes. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> Copilot-Session: 988ad39e-5a8f-4e76-80b8-ccbe7f1f3c58 --- nvx-rs/public-api/linux.txt | 26 ++++++++++++++++++++++++++ nvx-rs/public-api/windows.txt | 26 ++++++++++++++++++++++++++ 2 files changed, 52 insertions(+) diff --git a/nvx-rs/public-api/linux.txt b/nvx-rs/public-api/linux.txt index d7e079c64..aaf095e64 100644 --- a/nvx-rs/public-api/linux.txt +++ b/nvx-rs/public-api/linux.txt @@ -3574,6 +3574,32 @@ field nvx::host::template::TimeContract::tsc_frequency_hz field nvx::host::template::TimeContract::tsc_tolerance_ppm fn nvx::host::template::reattach_clone fn nvx::host::template::status_evidence +mod nvx::host::transport +struct nvx::host::transport::HostConnection +method nvx::host::transport::HostConnection::attach +method nvx::host::transport::HostConnection::identity +method nvx::host::transport::HostConnection::receive +method nvx::host::transport::HostConnection::send +method nvx::host::transport::HostConnection::session +enum nvx::host::transport::HostConnectionError +variant nvx::host::transport::HostConnectionError::Closed +variant nvx::host::transport::HostConnectionError::Ended +variant nvx::host::transport::HostConnectionError::Io +field nvx::host::transport::HostConnectionError::Io::0 +variant nvx::host::transport::HostConnectionError::Protocol +field nvx::host::transport::HostConnectionError::Protocol::0 +variant nvx::host::transport::HostConnectionError::Rejected +field nvx::host::transport::HostConnectionError::Rejected::0 +variant nvx::host::transport::HostConnectionError::Reset +field nvx::host::transport::HostConnectionError::Reset::0 +variant nvx::host::transport::HostConnectionError::TimedOut +trait nvx::host::transport::Transport +method nvx::host::transport::Transport::read +method nvx::host::transport::Transport::try_clone_handle +method nvx::host::transport::Transport::unread +method nvx::host::transport::Transport::write_all +type nvx::host::transport::TransportHandle +fn nvx::host::transport::connect_verified mod nvx::host::vm struct nvx::host::vm::Capability method nvx::host::vm::Capability::as_bytes diff --git a/nvx-rs/public-api/windows.txt b/nvx-rs/public-api/windows.txt index bba6c1433..c7c7f3334 100644 --- a/nvx-rs/public-api/windows.txt +++ b/nvx-rs/public-api/windows.txt @@ -2272,6 +2272,32 @@ field nvx::host::template::TimeContract::tsc_frequency_hz field nvx::host::template::TimeContract::tsc_tolerance_ppm fn nvx::host::template::reattach_clone fn nvx::host::template::status_evidence +mod nvx::host::transport +struct nvx::host::transport::HostConnection +method nvx::host::transport::HostConnection::attach +method nvx::host::transport::HostConnection::identity +method nvx::host::transport::HostConnection::receive +method nvx::host::transport::HostConnection::send +method nvx::host::transport::HostConnection::session +enum nvx::host::transport::HostConnectionError +variant nvx::host::transport::HostConnectionError::Closed +variant nvx::host::transport::HostConnectionError::Ended +variant nvx::host::transport::HostConnectionError::Io +field nvx::host::transport::HostConnectionError::Io::0 +variant nvx::host::transport::HostConnectionError::Protocol +field nvx::host::transport::HostConnectionError::Protocol::0 +variant nvx::host::transport::HostConnectionError::Rejected +field nvx::host::transport::HostConnectionError::Rejected::0 +variant nvx::host::transport::HostConnectionError::Reset +field nvx::host::transport::HostConnectionError::Reset::0 +variant nvx::host::transport::HostConnectionError::TimedOut +trait nvx::host::transport::Transport +method nvx::host::transport::Transport::read +method nvx::host::transport::Transport::try_clone_handle +method nvx::host::transport::Transport::unread +method nvx::host::transport::Transport::write_all +type nvx::host::transport::TransportHandle +fn nvx::host::transport::connect_verified mod nvx::host::vm struct nvx::host::vm::Capability method nvx::host::vm::Capability::as_bytes From e1153a11e76db871f22e13ba7f98d1b4fea00ab0 Mon Sep 17 00:00:00 2001 From: Enrique Saurez Date: Sat, 7 Nov 2026 16:57:41 -0800 Subject: [PATCH 4/7] SL2.11a: align transport namespace documentation Correct the NVXC layering description and the runtime adapter client module path after moving the retained host driver. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> Copilot-Session: 988ad39e-5a8f-4e76-80b8-ccbe7f1f3c58 --- nvx-rs/src/host/mod.rs | 9 ++++----- tests/runtime-adapter/src/host.rs | 6 +++--- 2 files changed, 7 insertions(+), 8 deletions(-) diff --git a/nvx-rs/src/host/mod.rs b/nvx-rs/src/host/mod.rs index 261315c90..40a7be2d7 100644 --- a/nvx-rs/src/host/mod.rs +++ b/nvx-rs/src/host/mod.rs @@ -35,11 +35,10 @@ //! - [`template`] retains the legacy capture and clone plans, the [`Store`](template::Store) of //! published generations and their leases, and aliases of the snapshot metadata. Its //! [`TemplatePlan`](template::TemplatePlan) lowers an adapter's plan for a capture. -//! - [`transport`] verifies a VM's endpoint and runs the host's role of the NVXS session. -//! [`nvxs`] retains compatibility paths. [`session`] speaks NVX's control protocol, `NVXC`, -//! over that connection for it and for any other application protocol, over a platform -//! transport. [`platform`] holds the primitives of Linux and Windows hosts. Other hosts compile -//! but report NVX as unavailable. +//! - [`session`] speaks NVX's control protocol, `NVXC`, over [`transport`], which verifies a VM's +//! endpoint and runs the host's role of the NVXS session for it and for any other application +//! protocol, over a platform transport. [`nvxs`] retains compatibility paths. [`platform`] holds +//! the primitives of Linux and Windows hosts. Other hosts compile but report NVX as unavailable. //! - [`process`] names processes by verified references: their ID, start time, and host boot, //! through which a pidfd or a process handle holds them, so that waits sleep until a process //! exits and termination reaches only it, whether or not the caller started it. diff --git a/tests/runtime-adapter/src/host.rs b/tests/runtime-adapter/src/host.rs index 28db62a7c..3df2178f5 100644 --- a/tests/runtime-adapter/src/host.rs +++ b/tests/runtime-adapter/src/host.rs @@ -1,8 +1,8 @@ //! The adapter's host program: a client of the guest agent that attaches to the host's end of the //! control session between the host and the guest, and exchanges the adapter's own -//! [framing](crate::frame) with the guest agent, through `nvx::host::nvxs` alone, over any platform -//! transport, and the driver of the [executions](crate::exec) that the host program runs over it, -//! [`Executions`]. +//! [framing](crate::frame) with the guest agent, through `nvx::host::transport` alone, over any +//! platform transport, and the driver of the [executions](crate::exec) that the host program runs +//! over it, [`Executions`]. //! //! # The host's order //! From e64dc23ecac00048fa7448957a61639be7cd9639 Mon Sep 17 00:00:00 2001 From: Enrique Saurez Date: Sat, 7 Nov 2026 17:15:48 -0800 Subject: [PATCH 5/7] SL2.11a: reuse the adapter host error kind import Remove the redundant ErrorKind import exposed by the host-only adapter build, retaining the existing grouped crate import. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> Copilot-Session: 988ad39e-5a8f-4e76-80b8-ccbe7f1f3c58 --- tests/runtime-adapter/src/bin/runtime-adapter-host.rs | 1 - 1 file changed, 1 deletion(-) diff --git a/tests/runtime-adapter/src/bin/runtime-adapter-host.rs b/tests/runtime-adapter/src/bin/runtime-adapter-host.rs index ca3713e1b..449ceb1f0 100644 --- a/tests/runtime-adapter/src/bin/runtime-adapter-host.rs +++ b/tests/runtime-adapter/src/bin/runtime-adapter-host.rs @@ -446,7 +446,6 @@ use std::path::{Path, PathBuf}; use std::process::ExitCode; use std::time::{Duration, Instant}; -use nvx::ErrorKind; use nvx::activation::{MachineIdMode, StoreGeneration}; use nvx::exec::{Cause, Key, Limits, Outcome, Status, Stream, Timers, Window}; use nvx::host::artifacts::{Artifacts, absolute}; From 7654ec4ee3ff7d4b7c5a4c8fc70cd2637b92f549 Mon Sep 17 00:00:00 2001 From: Enrique Saurez Date: Sun, 8 Nov 2026 21:54:17 -0800 Subject: [PATCH 6/7] host: verify original peers before Linux control authentication Reuse the socket peer pidfd check before returning a control transport so a transferred listener cannot authenticate an exited creator's reused PID. Reject unsupported or denied peer verification without a numeric-PID fallback. Cover transferred listeners and denied SO_PEERPIDFD queries with Linux regressions that verify their subprocesses actually ran. Document that all Linux host control sessions now require Linux 6.5 or a backport; the guest kernel and OpenVMM pin are unchanged. Refs https://github.com/microsoft/nvx/pull/1406#discussion_r4509406405 Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> Copilot-Session: ebc0c25b-c373-4a83-8598-86152db0cce1 --- nvx-rs/README.md | 6 +- nvx-rs/src/host/platform/linux/endpoint.rs | 11 +- nvx-rs/src/host/platform/linux/tests.rs | 137 +++++++++++++++++++++ nvx-rs/src/host/transport.rs | 6 +- 4 files changed, 152 insertions(+), 8 deletions(-) diff --git a/nvx-rs/README.md b/nvx-rs/README.md index 7ae95118e..637a7f06c 100644 --- a/nvx-rs/README.md +++ b/nvx-rs/README.md @@ -917,8 +917,10 @@ owned by [`nvx::host::transport`](#nvxhosttransport). `connect_verified` opens and holds the recorded process by its boot, ID and start time, retries while the endpoint is unavailable until the caller's deadline, and checks its server before authentication: Unix peer credentials -on Linux and the named-pipe server process ID on Windows. An indeterminate -process check remains an error, never exit proof. +and the original socket peer's live pidfd on Linux, and the named-pipe server +process ID on Windows. Linux requires `SO_PEERPIDFD` (Linux 6.5 or a backport) +and refuses unverifiable peers rather than falling back to a numeric PID. +An indeterminate process check remains an error, never exit proof. `HostConnection`: the host's role of the NVXS session over a platform transport, on which `nvx::host::session` and other host applications run their protocols. diff --git a/nvx-rs/src/host/platform/linux/endpoint.rs b/nvx-rs/src/host/platform/linux/endpoint.rs index 83850a45d..dc29d19f6 100644 --- a/nvx-rs/src/host/platform/linux/endpoint.rs +++ b/nvx-rs/src/host/platform/linux/endpoint.rs @@ -189,14 +189,14 @@ fn check_original_peer(connection: &UnixStream) -> io::Result<()> { ) => { return Err(io::Error::new( io::ErrorKind::Unsupported, - "restore-ready peer authentication requires SO_PEERPIDFD", + "socket peer authentication requires SO_PEERPIDFD", )); } // Before Linux 6.16, a reaped peer yields EINVAL rather than a readable pidfd. Err(sys::Errno::EINVAL | sys::Errno::ESRCH | sys::Errno::ENODATA) => { return Err(io::Error::new( io::ErrorKind::PermissionDenied, - "the original restore-ready socket peer exited", + "the original socket peer exited", )); } result => result?, @@ -209,11 +209,11 @@ fn check_original_peer(connection: &UnixStream) -> io::Result<()> { if events.intersects(PollFlags::POLLIN | PollFlags::POLLHUP) { return Err(io::Error::new( io::ErrorKind::PermissionDenied, - "the original restore-ready socket peer exited", + "the original socket peer exited", )); } Err(io::Error::other(format!( - "the restore-ready peer pidfd reported unexpected events {events:?}" + "the socket peer pidfd reported unexpected events {events:?}" ))) } @@ -248,6 +248,8 @@ fn set_ready_socket_mode(socket_path: &Path) -> io::Result<()> { /// Connects to a control endpoint served by the process `expected_pid`. /// +/// Requires a live original socket peer through `SO_PEERPIDFD`, not just its cached numeric PID. +/// /// The connection attempt never blocks: a listener whose backlog is full yields /// [`io::ErrorKind::WouldBlock`], and the caller retries within its deadline. pub(crate) fn connect_endpoint( @@ -265,6 +267,7 @@ pub(crate) fn connect_endpoint( ), )); } + check_original_peer(&stream)?; Ok(Box::new(UnixTransport { stream })) } diff --git a/nvx-rs/src/host/platform/linux/tests.rs b/nvx-rs/src/host/platform/linux/tests.rs index 8b12359ed..d977adcae 100644 --- a/nvx-rs/src/host/platform/linux/tests.rs +++ b/nvx-rs/src/host/platform/linux/tests.rs @@ -52,6 +52,143 @@ fn a_duplicated_transport_keeps_the_verified_connection_without_reconnecting() { assert_eq!(peer.read(&mut response).unwrap(), 0); } +#[test] +fn a_transferred_listener_cannot_authenticate_its_exited_creator() { + use std::os::fd::AsFd; + use std::os::unix::net::UnixListener; + + let child = "NVX_TEST_TRANSFERRED_CONTROL_LISTENER"; + let endpoint = "NVX_TEST_TRANSFERRED_CONTROL_ENDPOINT"; + let verified = "NVX_STALE_CONTROL_PEER_REJECTED"; + if let Some(pid) = std::env::var_os(child) { + let pid: u32 = pid.to_str().unwrap().parse().unwrap(); + let endpoint = std::env::var(endpoint).unwrap(); + let listener = UnixListener::from(std::io::stdin().as_fd().try_clone_to_owned().unwrap()); + let probe = crate::host::sys::connect_nonblocking(&endpoint).unwrap(); + assert_eq!(crate::host::sys::peer_pid(&probe).unwrap(), pid); + drop(listener.accept().unwrap()); + drop(probe); + // Match the stale creator's PID without changing the host's PID allocator. + let error = connect_endpoint(&endpoint, pid, Duration::from_secs(2)) + .err() + .expect("a stale listener's numeric PID must not authenticate it"); + assert_eq!(error.kind(), std::io::ErrorKind::PermissionDenied); + assert!(error.to_string().contains("original socket peer exited")); + let (mut peer, _) = listener.accept().unwrap(); + peer.set_read_timeout(Some(Duration::from_secs(2))).unwrap(); + assert_eq!(peer.read(&mut [0]).unwrap(), 0); + println!("{verified} pid={pid}"); + return; + } + + let directory = tempfile::tempdir().unwrap(); + let path = directory.path().join("control.sock"); + let mut command = Command::new("python3"); + command + .args([ + "-c", + r#"import array, os, socket, sys +receiver, sender = socket.socketpair() +creator = os.fork() +if creator == 0: + receiver.close() + listener = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) + listener.bind(sys.argv[1]) + listener.listen() + sender.sendmsg([b"L"], [(socket.SOL_SOCKET, socket.SCM_RIGHTS, + array.array("i", [listener.fileno()]))]) + os._exit(0) +sender.close() +data, messages, flags, address = receiver.recvmsg(1, socket.CMSG_SPACE(array.array("i").itemsize)) +assert data == b"L" and len(messages) == 1 +assert messages[0][:2] == (socket.SOL_SOCKET, socket.SCM_RIGHTS) +descriptors = array.array("i") +descriptors.frombytes(messages[0][2]) +assert len(descriptors) == 1 +assert os.waitpid(creator, 0) == (creator, 0) +os.dup2(descriptors[0], 0) +if descriptors[0] != 0: + os.close(descriptors[0]) +receiver.close() +os.environ["NVX_TEST_TRANSFERRED_CONTROL_LISTENER"] = str(creator) +os.execv(sys.argv[2], [sys.argv[2], "--exact", sys.argv[3], "--nocapture", "--test-threads=1"]) +"#, + ]) + .arg(&path) + .arg(std::env::current_exe().unwrap()) + .arg( + "host::platform::linux::tests::a_transferred_listener_cannot_authenticate_its_exited_creator", + ) + .env(endpoint, &path) + .stdin(Stdio::null()) + .stdout(Stdio::piped()) + .stderr(Stdio::piped()); + let output = crate::host::tests::spawn_locked(&mut command) + .unwrap() + .wait_with_output() + .unwrap(); + assert!( + output.status.success() && String::from_utf8_lossy(&output.stdout).contains(verified), + "{}\n{}", + String::from_utf8_lossy(&output.stdout), + String::from_utf8_lossy(&output.stderr) + ); +} + +#[test] +fn a_control_connection_fails_closed_without_original_peer_pidfds() { + use std::os::unix::net::UnixListener; + + let child = "NVX_TEST_CONTROL_PIDFD_DENIAL"; + let Some(errno) = std::env::var_os(child) else { + for errno in [nix::libc::ENOPROTOOPT, nix::libc::EPERM, nix::libc::EINVAL] { + let mut command = Command::new(std::env::current_exe().unwrap()); + command + .args([ + "--exact", + "host::platform::linux::tests::a_control_connection_fails_closed_without_original_peer_pidfds", + "--nocapture", + ]) + .env(child, errno.to_string()) + .stdin(Stdio::null()) + .stdout(Stdio::piped()) + .stderr(Stdio::piped()); + crate::host::sys::tests::refuse_peer_pidfd_on_spawn(&mut command, errno); + let output = crate::host::tests::spawn_locked(&mut command) + .unwrap() + .wait_with_output() + .unwrap(); + assert!( + output.status.success() + && String::from_utf8_lossy(&output.stdout) + .contains(&format!("NVX_CONTROL_PEER_PIDFD_REFUSED {errno}")), + "{}\n{}", + String::from_utf8_lossy(&output.stdout), + String::from_utf8_lossy(&output.stderr) + ); + } + return; + }; + let errno: i32 = errno.to_str().unwrap().parse().unwrap(); + let directory = tempfile::tempdir().unwrap(); + let path = directory.path().join("control.sock"); + let _listener = UnixListener::bind(&path).unwrap(); + let error = connect_endpoint( + path.to_str().unwrap(), + std::process::id(), + Duration::from_secs(2), + ) + .err() + .expect("an unverifiable peer must not authenticate by numeric PID alone"); + let expected = if errno == nix::libc::EINVAL { + std::io::ErrorKind::PermissionDenied + } else { + std::io::ErrorKind::Unsupported + }; + assert_eq!(error.kind(), expected); + println!("NVX_CONTROL_PEER_PIDFD_REFUSED {errno}"); +} + #[test] fn a_wait_sleeps_a_remainder_below_a_millisecond_instead_of_polling_a_millisecond() { assert_eq!(wait_step(Duration::ZERO), (Duration::ZERO, Duration::ZERO)); diff --git a/nvx-rs/src/host/transport.rs b/nvx-rs/src/host/transport.rs index d27dd6979..5ab40b105 100644 --- a/nvx-rs/src/host/transport.rs +++ b/nvx-rs/src/host/transport.rs @@ -62,8 +62,10 @@ pub use super::platform::TransportHandle; /// not yet available, until `deadline`. /// /// Opens and holds the process with its recorded boot, ID and start time. Linux checks the Unix -/// socket's peer credentials; Windows checks the named pipe's server process ID. The held -/// process is checked again after connection, before the transport can send a capability. +/// socket's peer credentials and original peer pidfd, requiring `SO_PEERPIDFD` (Linux 6.5 or a +/// backport) without falling back to a numeric PID. Windows checks the named pipe's server +/// process ID. The held process is checked again after connection, before the transport can +/// send a capability. /// /// A passed deadline is [`ErrorKind::TimedOut`], an exited recorded process /// [`ErrorKind::Exited`], a refused or broken session [`ErrorKind::Protocol`], and an From 81f58751f77787bc3fc9e6abfe4a3cfe12696be1 Mon Sep 17 00:00:00 2001 From: Enrique Saurez Date: Sun, 8 Nov 2026 23:20:03 -0800 Subject: [PATCH 7/7] host: preserve unsupported Linux control-peer diagnostics Return permanent SO_PEERPIDFD verification failures immediately from the CLI session connector, retaining the existing I/O diagnostic. Keep the retry behavior for unavailable endpoints and Windows named pipes. Add unavailable and denied peer-pidfd subprocess regressions that verify the diagnostic is retained without polling or retrying. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> Copilot-Session: ebc0c25b-c373-4a83-8598-86152db0cce1 --- nvx-rs/src/host/cli/session.rs | 12 +++--- nvx-rs/src/host/cli/session/tests.rs | 58 ++++++++++++++++++++++++++++ 2 files changed, 65 insertions(+), 5 deletions(-) diff --git a/nvx-rs/src/host/cli/session.rs b/nvx-rs/src/host/cli/session.rs index 142a258bf..4e0a6aabb 100644 --- a/nvx-rs/src/host/cli/session.rs +++ b/nvx-rs/src/host/cli/session.rs @@ -6,7 +6,8 @@ //! //! - [`Session::connect`] retries the endpoint every 25 ms until its timeout, then attaches with //! the capability within a fresh timeout. On Windows, it gives up at once on the errors that -//! Python does not retry. +//! Python does not retry. On Linux, unsupported peer verification fails at once, retaining +//! its diagnostic instead of reporting that the endpoint did not appear. //! - Each operation bounds its request and its response by one timeout. A transfer that times //! out fails with `managed control request timed out` while it sends and `managed control //! response timed out` while it receives. @@ -247,10 +248,11 @@ impl Session { match platform::connect_endpoint(endpoint, pid, slice) { Ok(transport) => break transport, Err(error) => { - if cfg!(windows) - && error - .raw_os_error() - .is_some_and(|code| raised_at_once(code as u32)) + if (cfg!(target_os = "linux") && error.kind() == io::ErrorKind::Unsupported) + || (cfg!(windows) + && error + .raw_os_error() + .is_some_and(|code| raised_at_once(code as u32))) { return Err(Exit::from_os_error(error)); } diff --git a/nvx-rs/src/host/cli/session/tests.rs b/nvx-rs/src/host/cli/session/tests.rs index e54951b33..270f75a55 100644 --- a/nvx-rs/src/host/cli/session/tests.rs +++ b/nvx-rs/src/host/cli/session/tests.rs @@ -992,6 +992,64 @@ mod endpoints { ); } + #[test] + #[cfg(target_os = "linux")] + fn unsupported_peer_verification_fails_without_retrying() { + use std::process::{Command, Stdio}; + + let child = "NVX_TEST_CLI_CONTROL_PIDFD_DENIAL"; + let Some(errno) = std::env::var_os(child) else { + for errno in [nix::libc::ENOPROTOOPT, nix::libc::EPERM, nix::libc::EACCES] { + let mut command = Command::new(std::env::current_exe().unwrap()); + command + .args([ + "--exact", + "host::cli::session::tests::endpoints::unsupported_peer_verification_fails_without_retrying", + "--nocapture", + ]) + .env(child, errno.to_string()) + .stdin(Stdio::null()) + .stdout(Stdio::piped()) + .stderr(Stdio::piped()); + crate::host::sys::tests::refuse_peer_pidfd_on_spawn(&mut command, errno); + let output = crate::host::tests::spawn_locked(&mut command) + .unwrap() + .wait_with_output() + .unwrap(); + assert!( + output.status.success() + && String::from_utf8_lossy(&output.stdout) + .contains(&format!("NVX_CLI_CONTROL_PIDFD_REFUSED {errno}")), + "{}\n{}", + String::from_utf8_lossy(&output.stdout), + String::from_utf8_lossy(&output.stderr) + ); + } + return; + }; + let errno: i32 = errno.to_str().unwrap().parse().unwrap(); + let directory = tempfile::tempdir().unwrap(); + let path = directory.path().join("control.sock"); + let _listener = UnixListener::bind(&path).unwrap(); + let mut checks = 0; + let failure = Session::connect_unless_exited( + path.to_str().unwrap(), + std::process::id(), + &CAPABILITY, + 60.0, + || { + checks += 1; + checks == 2 + }, + ) + .err() + .expect("unsupported peer verification must fail before authentication"); + assert_eq!(failure.code(), Some(ErrorCode::Io)); + assert!(failure.message().unwrap().contains("SO_PEERPIDFD")); + assert_eq!(checks, 0, "a permanent verification failure must not retry"); + println!("NVX_CLI_CONTROL_PIDFD_REFUSED {errno}"); + } + #[test] fn waits_for_an_endpoint_end_once_openvmm_exits() { let directory = tempfile::tempdir().unwrap();