diff --git a/nvx-rs/README.md b/nvx-rs/README.md index 99495c993..7f1a119d7 100644 --- a/nvx-rs/README.md +++ b/nvx-rs/README.md @@ -989,6 +989,14 @@ which is zero once OpenVMM read everything. A named pipe tells its writer nothing of the kind, so on Windows, and for a transport that does not tell, it is `None`. +On Linux and Windows, `Transport::try_clone_handle` optionally returns an +owned native handle to the same peer-verified connection, not a reconnect: +an `OwnedFd` for the Unix socket or an `OwnedHandle` for the overlapped pipe. +An adapter must drop the blocking transport before using or registering the +duplicate with an I/O runtime, or changing the shared blocking mode. The +default is `Ok(None)`, so transports without native handles need not implement +it, and NVX does not depend on an asynchronous I/O runtime. + ### `nvx::host::sandbox` The launch profile of `nvx-rs sandbox`, with native input validation: diff --git a/nvx-rs/src/host/platform/linux/endpoint.rs b/nvx-rs/src/host/platform/linux/endpoint.rs index e68304ed7..f0d027671 100644 --- a/nvx-rs/src/host/platform/linux/endpoint.rs +++ b/nvx-rs/src/host/platform/linux/endpoint.rs @@ -10,7 +10,7 @@ use std::path::Path; use std::thread; use std::time::{Duration, Instant}; -use crate::host::platform::{Transport, time_left}; +use crate::host::platform::{Transport, TransportHandle, time_left}; use crate::host::sys; /// Returns the control endpoint OpenVMM should listen on for a sandbox. @@ -236,6 +236,10 @@ fn map_timeout(error: io::Error) -> io::Error { } impl Transport for UnixTransport { + fn try_clone_handle(&self) -> io::Result> { + self.stream.try_clone().map(|stream| Some(stream.into())) + } + fn read(&mut self, buffer: &mut [u8], timeout: Option) -> io::Result { self.stream.set_read_timeout(socket_timeout(timeout))?; loop { diff --git a/nvx-rs/src/host/platform/linux/tests.rs b/nvx-rs/src/host/platform/linux/tests.rs index 8ea241b8c..8b12359ed 100644 --- a/nvx-rs/src/host/platform/linux/tests.rs +++ b/nvx-rs/src/host/platform/linux/tests.rs @@ -13,6 +13,45 @@ pub(crate) use super::process::{check_identity, start_time_from_stat}; use super::*; use crate::host::platform::Transport; +#[test] +fn a_duplicated_transport_keeps_the_verified_connection_without_reconnecting() { + use std::io::Write; + use std::os::unix::net::UnixListener; + + let directory = tempfile::tempdir().unwrap(); + let path = directory.path().join("control.sock"); + let listener = UnixListener::bind(&path).unwrap(); + let transport = connect_endpoint( + path.to_str().unwrap(), + std::process::id(), + Duration::from_secs(2), + ) + .unwrap(); + let (mut peer, _) = listener.accept().unwrap(); + drop(listener); + fs::remove_file(&path).unwrap(); + // Neither dropping another duplicate nor removing the endpoint closes this connection. + drop(transport.try_clone_handle().unwrap().unwrap()); + let handle = transport.try_clone_handle().unwrap().unwrap(); + drop(transport); + let mut duplicate = UnixStream::from(handle); + let timeout = Some(Duration::from_secs(2)); + duplicate.set_read_timeout(timeout).unwrap(); + duplicate.set_write_timeout(timeout).unwrap(); + peer.set_read_timeout(timeout).unwrap(); + peer.set_write_timeout(timeout).unwrap(); + duplicate.write_all(b"request").unwrap(); + let mut request = [0; 7]; + peer.read_exact(&mut request).unwrap(); + assert_eq!(&request, b"request"); + peer.write_all(b"response").unwrap(); + let mut response = [0; 8]; + duplicate.read_exact(&mut response).unwrap(); + assert_eq!(&response, b"response"); + drop(duplicate); + assert_eq!(peer.read(&mut response).unwrap(), 0); +} + #[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/mod.rs b/nvx-rs/src/host/platform/mod.rs index 6ad8f507e..bcaf56946 100644 --- a/nvx-rs/src/host/platform/mod.rs +++ b/nvx-rs/src/host/platform/mod.rs @@ -45,6 +45,17 @@ pub fn probe_hypervisor(hypervisor: Hypervisor) -> Result<()> { /// Byte stream to an OpenVMM control endpoint. pub trait Transport: Send { + /// Duplicates the connected endpoint for an adapter that uses native asynchronous I/O. + /// + /// Returns `Ok(None)` by default for transports without a native handle. The duplicate + /// owns a handle to the same already peer-verified connection, not a new connection. + /// The caller must drop this transport before using the duplicate or registering it with + /// an I/O runtime, and must not change the shared blocking mode while this transport lives. + #[cfg(any(target_os = "linux", windows))] + fn try_clone_handle(&self) -> io::Result> { + Ok(None) + } + /// Reads at least one byte, waiting at most `timeout` (`None` waits indefinitely). /// /// Returns `Ok(0)` at end of stream and an [`io::ErrorKind::TimedOut`] error on timeout. @@ -69,6 +80,14 @@ pub trait Transport: Send { } } +/// An owned duplicate of a verified control socket connection. +#[cfg(target_os = "linux")] +pub type TransportHandle = std::os::fd::OwnedFd; + +/// An owned duplicate of a verified, overlapped control pipe connection. +#[cfg(windows)] +pub type TransportHandle = std::os::windows::io::OwnedHandle; + /// Returns the time left until `deadline`, `None` without one, or an [`io::ErrorKind::TimedOut`] /// error once it has passed. #[cfg(any(target_os = "linux", windows))] diff --git a/nvx-rs/src/host/platform/tests.rs b/nvx-rs/src/host/platform/tests.rs index 163098843..d44a79b5f 100644 --- a/nvx-rs/src/host/platform/tests.rs +++ b/nvx-rs/src/host/platform/tests.rs @@ -6,6 +6,25 @@ pub(crate) use super::linux::tests::{ pub(crate) use super::windows::tests::set_access; use super::*; +#[cfg(any(target_os = "linux", windows))] +#[test] +fn a_transport_without_a_native_handle_needs_no_clone_implementation() { + struct CustomTransport; + + impl Transport for CustomTransport { + fn read(&mut self, _: &mut [u8], _: Option) -> io::Result { + unreachable!("cloning a handle does not read the transport") + } + + fn write_all(&mut self, _: &[u8], _: Option) -> io::Result<()> { + unreachable!("cloning a handle does not write the transport") + } + } + + let transport: Box = Box::new(CustomTransport); + assert!(transport.try_clone_handle().unwrap().is_none()); +} + #[cfg(any(target_os = "linux", windows))] #[test] fn host_identities_are_read_as_openvmm_reads_them() { diff --git a/nvx-rs/src/host/platform/windows/endpoint.rs b/nvx-rs/src/host/platform/windows/endpoint.rs index 2efb85dd2..736721baa 100644 --- a/nvx-rs/src/host/platform/windows/endpoint.rs +++ b/nvx-rs/src/host/platform/windows/endpoint.rs @@ -28,7 +28,7 @@ use windows_sys::Win32::System::Threading::{ use super::private::{CurrentUser, with_owner_only_attributes}; use super::{last_error, owned, raw, wide}; -use crate::host::platform::{Transport, time_left}; +use crate::host::platform::{Transport, TransportHandle, time_left}; /// Returns a fresh named-pipe endpoint for OpenVMM to listen on. pub(crate) fn control_endpoint(_socket_path: &Path) -> io::Result { @@ -332,6 +332,10 @@ fn completion_error(error: u32, transferred: usize, timed_out: bool) -> io::Resu } impl Transport for PipeTransport { + fn try_clone_handle(&self) -> io::Result> { + self.pipe.try_clone().map(Some) + } + fn read(&mut self, buffer: &mut [u8], timeout: Option) -> io::Result { let length = u32::try_from(buffer.len()).unwrap_or(u32::MAX); let data = buffer.as_mut_ptr(); @@ -361,3 +365,6 @@ impl Transport for PipeTransport { Ok(()) } } + +#[cfg(test)] +mod tests; diff --git a/nvx-rs/src/host/platform/windows/endpoint/tests.rs b/nvx-rs/src/host/platform/windows/endpoint/tests.rs new file mode 100644 index 000000000..410ac39e0 --- /dev/null +++ b/nvx-rs/src/host/platform/windows/endpoint/tests.rs @@ -0,0 +1,62 @@ +use std::fs::File; +use std::io::{Read, Write}; + +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); + // SAFETY: the name is NUL-terminated; this creates a fresh local byte pipe. + let server = owned(unsafe { + CreateNamedPipeW( + name.as_ptr(), + PIPE_ACCESS_DUPLEX | FILE_FLAG_FIRST_PIPE_INSTANCE, + PIPE_TYPE_BYTE | PIPE_READMODE_BYTE | PIPE_WAIT | PIPE_REJECT_REMOTE_CLIENTS, + 1, + 4096, + 4096, + 0, + ptr::null(), + ) + }) + .unwrap(); + 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); + // 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); + duplicate + .write_all(b"request", Some(Duration::from_secs(2))) + .unwrap(); + let mut request = [0; 7]; + server.read_exact(&mut request).unwrap(); + assert_eq!(&request, b"request"); + server.write_all(b"response").unwrap(); + let mut response = [0; 8]; + assert_eq!( + duplicate + .read(&mut response, Some(Duration::from_secs(2))) + .unwrap(), + response.len() + ); + assert_eq!(&response, b"response"); + drop(duplicate); + assert_eq!(server.read(&mut response).unwrap(), 0); +}