nvx-rs: expose optional native control transport handles

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:

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<Option<TransportHandle>> {
+        self.stream.try_clone().map(|stream| Some(stream.into()))
+    }
+
     fn read(&mut self, buffer: &mut [u8], timeout: Option<Duration>) -> io::Result<usize> {
         self.stream.set_read_timeout(socket_timeout(timeout))?;
         loop {

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));

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<Option<TransportHandle>> {
+        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))]

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<Duration>) -> io::Result<usize> {
+            unreachable!("cloning a handle does not read the transport")
+        }
+
+        fn write_all(&mut self, _: &[u8], _: Option<Duration>) -> io::Result<()> {
+            unreachable!("cloning a handle does not write the transport")
+        }
+    }
+
+    let transport: Box<dyn Transport> = 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() {

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<String> {
@@ -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<Option<TransportHandle>> {
+        self.pipe.try_clone().map(Some)
+    }
+
     fn read(&mut self, buffer: &mut [u8], timeout: Option<Duration>) -> io::Result<usize> {
         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;

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);
+}