diff --git a/nvx-rs/README.md b/nvx-rs/README.md index 595f88ad9..637a7f06c 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,19 @@ 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 +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. It attaches with the launch capability, waiting while the broker asks it to, @@ -915,6 +929,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/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 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(); diff --git a/nvx-rs/src/host/mod.rs b/nvx-rs/src/host/mod.rs index 950a59525..40a7be2d7 100644 --- a/nvx-rs/src/host/mod.rs +++ b/nvx-rs/src/host/mod.rs @@ -35,10 +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. -//! - [`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. [`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. @@ -103,4 +103,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/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/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..b5d388a90 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,35 @@ 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) +} + +/// 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 { + ErrorKind::TimedOut => SessionError::TimedOut, + ErrorKind::Exited => SessionError::ProcessExited, + _ => { + let Some(source) = source else { + return SessionError::Protocol(message); + }; + match source.downcast::() { + Ok(error) + if kind == ErrorKind::Protocol + && 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 +403,30 @@ 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) +} + +/// 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, 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..2b90794a4 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}; @@ -10,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 new file mode 100644 index 000000000..5ab40b105 --- /dev/null +++ b/nvx-rs/src/host/transport.rs @@ -0,0 +1,436 @@ +//! 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 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 +/// 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())), + ); + } +} + +/// Refuses an exited held VM, without relying on its numeric PID. +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(()) +} + +/// 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, + 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..2136f8fbb --- /dev/null +++ b/nvx-rs/src/host/transport/tests.rs @@ -0,0 +1,291 @@ +#![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_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(); + 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..0d075030c 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,64 @@ 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::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 +471,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 +488,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 +514,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..449ceb1f0 100644 --- a/tests/runtime-adapter/src/bin/runtime-adapter-host.rs +++ b/tests/runtime-adapter/src/bin/runtime-adapter-host.rs @@ -452,7 +452,6 @@ 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 +2783,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..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 //! @@ -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()); }