diff --git a/nvx-rs/README.md b/nvx-rs/README.md index d12fac71e..3e25da9ff 100644 --- a/nvx-rs/README.md +++ b/nvx-rs/README.md @@ -876,6 +876,14 @@ cancellation and exit fail every pending phase. Failure stops no process and writes no record. A matching Ready ends at `Done`, without sending mappings or Start. Terminal results are offered once and absorb later inputs. +NVX's own binaries drive the machine with a blocking driver, private to the +crate, beside the control session. It checks that the restore-ready client and +the control server are the VM's held process, sends the same activate message, +the record and the workload's identity, in at most three sessions, and allows +60 seconds to Ready, then 30 more for the mapping table and the workload's +Start. The caller's cancellation and the VM's exit end any pending accept, read +or write. + `ReadyListener` owns a restore-ready endpoint: a mode-0600 Unix socket in the private VM directory on Linux, or a fresh first-instance named pipe with an owner-only protected access control list on Windows. Its public `receive` diff --git a/nvx-rs/src/host/mod.rs b/nvx-rs/src/host/mod.rs index a41a2dcbe..cecb93c0f 100644 --- a/nvx-rs/src/host/mod.rs +++ b/nvx-rs/src/host/mod.rs @@ -55,7 +55,9 @@ //! restore from a [`Snapshot`](snapshot::Snapshot) hard-links the snapshot's OpenVMM files into //! the VM directory and binds its restore-ready listener before any spawn. //! - [`restore`] orders caller-driven activation to Ready without I/O or a clock, and listens for -//! OpenVMM's restore-ready event from the held VM process. +//! OpenVMM's restore-ready event from the held VM process. The private `restore_driver` (feature +//! `cli`) drives it over the [`session`] for NVX's blocking callers, through Ready, the mapping +//! table and the workload's start. //! - `cli` (feature `cli`) is the `nvx-rs` command-line interface, which runs microVMs and //! sandboxes, looks up and downloads artifacts, and compiles egress policies. //! @@ -90,6 +92,8 @@ pub mod platform; pub mod process; pub mod report; pub mod restore; +#[cfg(feature = "cli")] +mod restore_driver; pub mod runtime; pub mod sandbox; pub mod seal; diff --git a/nvx-rs/src/host/platform/linux/endpoint.rs b/nvx-rs/src/host/platform/linux/endpoint.rs index 7b407462c..b088a9e40 100644 --- a/nvx-rs/src/host/platform/linux/endpoint.rs +++ b/nvx-rs/src/host/platform/linux/endpoint.rs @@ -278,6 +278,16 @@ pub(crate) fn transport_from_handle(handle: TransportHandle) -> io::Result io::Result<()> { + match nix::sys::socket::shutdown(handle.as_raw_fd(), nix::sys::socket::Shutdown::Both) { + Ok(()) | Err(sys::Errno::ENOTCONN) => Ok(()), + Err(error) => Err(error.into()), + } +} + /// Returns the process that listens on a control endpoint, or `None` if nothing listens. pub(crate) fn endpoint_server_pid(endpoint: &str) -> io::Result> { endpoint_server_pid_until(endpoint, Instant::now() + Duration::from_secs(5)) diff --git a/nvx-rs/src/host/platform/linux/mod.rs b/nvx-rs/src/host/platform/linux/mod.rs index 988ced72d..8dffecfa0 100644 --- a/nvx-rs/src/host/platform/linux/mod.rs +++ b/nvx-rs/src/host/platform/linux/mod.rs @@ -18,7 +18,9 @@ pub(crate) use self::endpoint::{ ReadyListener, connect_endpoint, control_endpoint, endpoint_server_pid, listen_ready, }; #[cfg(feature = "cli")] -pub(crate) use self::endpoint::{endpoint_server_pid_until, transport_from_handle}; +pub(crate) use self::endpoint::{ + endpoint_server_pid_until, interrupt_transfers, transport_from_handle, +}; pub use self::identity::file_identity; pub(crate) use self::identity::{ Directory, file_id, is_deleted, link_count, opened_file_id, opened_file_identity, diff --git a/nvx-rs/src/host/platform/windows/endpoint.rs b/nvx-rs/src/host/platform/windows/endpoint.rs index dae07c09b..d76e3ab91 100644 --- a/nvx-rs/src/host/platform/windows/endpoint.rs +++ b/nvx-rs/src/host/platform/windows/endpoint.rs @@ -245,6 +245,21 @@ pub(crate) fn transport_from_handle(pipe: TransportHandle) -> io::Result io::Result<()> { + // SAFETY: the handle is an open pipe handle of the caller's, and no OVERLAPPED is named. + if unsafe { CancelIoEx(raw(handle), ptr::null()) } == 0 { + let error = last_error(); + if error != windows_sys::Win32::Foundation::ERROR_NOT_FOUND { + return Err(io::Error::from_raw_os_error(error as i32)); + } + } + Ok(()) +} + /// Returns the process that serves a named-pipe endpoint, or `None` if the pipe does not exist. /// /// OpenVMM keeps one pipe instance for its whole lifetime, so a missing pipe means no OpenVMM diff --git a/nvx-rs/src/host/platform/windows/mod.rs b/nvx-rs/src/host/platform/windows/mod.rs index a8333febb..2003f8d41 100644 --- a/nvx-rs/src/host/platform/windows/mod.rs +++ b/nvx-rs/src/host/platform/windows/mod.rs @@ -26,7 +26,9 @@ pub(crate) use self::endpoint::{ ReadyListener, connect_endpoint, control_endpoint, endpoint_server_pid, listen_ready, }; #[cfg(feature = "cli")] -pub(crate) use self::endpoint::{endpoint_server_pid_until, transport_from_handle}; +pub(crate) use self::endpoint::{ + endpoint_server_pid_until, interrupt_transfers, transport_from_handle, +}; pub use self::identity::file_identity; pub(crate) use self::identity::{ Directory, file_id, is_deleted, link_count, opened_file_id, opened_file_identity, diff --git a/nvx-rs/src/host/restore_driver.rs b/nvx-rs/src/host/restore_driver.rs new file mode 100644 index 000000000..9b819950e --- /dev/null +++ b/nvx-rs/src/host/restore_driver.rs @@ -0,0 +1,499 @@ +//! The blocking driver of a restore over NVX's control protocol, for NVX's own blocking callers. +//! +//! [`restore`] drives a [`RestoreMachine`] through a launched VM's restore-ready event and +//! control sessions until the guest's Ready, then sends the VM's mapping table and starts its +//! workload, and returns the live [`ControlSession`]. The restore-ready client and the control +//! server must be the VM's held process. Each session carries the same activate message: the +//! activation record and the workload's identity. A lost session is replaced, in at most +//! [`MAX_SESSIONS`] sessions in all. +//! +//! Every pending accept, read and write races the step's deadline, the caller's [`Cancel`] and +//! the VM's verified exit. The restore-ready endpoint and the control endpoint are waited for in +//! slices of at most [`POLL`], between which the restore checks both; while a control session +//! transfers, a watcher thread checks them and, once either happens, ends the session's pending +//! transfers. A failure leaves the VM's process, its outputs and its restore-ready listener to the +//! caller. + +use std::fmt; +use std::io; +use std::path::Path; +use std::sync::OnceLock; +use std::sync::atomic::{AtomicBool, Ordering}; +use std::sync::mpsc::{self, RecvTimeoutError}; +#[cfg(any(target_os = "linux", windows))] +use std::sync::{Mutex, PoisonError}; +use std::thread; +use std::time::{Duration, Instant}; + +#[cfg(any(target_os = "linux", windows))] +use super::platform; +use super::process::Process; +use super::report; +use super::restore::{ + ReadyListener, RestoreAction, RestoreError, RestoreEvidence, RestoreInput, RestoreMachine, + RestorePhase, +}; +use super::session::{self, ControlSession, SessionError}; +use super::transport::Transport; +use super::vm::{Capability, Launched, VmOutputs}; +use crate::ErrorKind; +use crate::activation::ActivationRecord; +use crate::protocol::{self, APP_EXEC, HostMap, workload::WorkloadIdentity}; + +/// How long a restore may take, from the driver's start to the guest's Ready. +const READY_BUDGET: Duration = Duration::from_secs(60); +/// How long the mapping table and the workload's Start may take, from the guest's Ready. +const START_BUDGET: Duration = Duration::from_secs(30); +/// The most control sessions in which a restore sends its activate message. +const MAX_SESSIONS: usize = 3; +/// The longest that a wait goes without checking the caller's cancellation and the VM's exit, +/// and the pause after a lost session, in which the VMM's console notices the loss and an +/// exiting VM shows its exit. +const POLL: Duration = Duration::from_millis(50); +/// How long a VMM that closed its restore-ready endpoint without the event may take to exit, as +/// OpenVMM does when it cannot start its VM worker. +const CLOSED_EXIT_WAIT: Duration = Duration::from_secs(10); +/// OpenVMM's restore-ready event. +const READY_EVENT: &[u8] = b"OPENVMM_RESTORE_READY_V1\n"; + +/// The caller's cancellation of a restore, which the restore observes within [`POLL`] once set. +pub(crate) type Cancel = AtomicBool; + +/// The workload that a restored VM runs, and the identity that it runs as. +pub(crate) struct Workload<'a> { + /// The identity that the activate message carries, as which nvx-init sets up the workload. + pub(crate) identity: WorkloadIdentity, + /// The workload that Start runs. + pub(crate) exec: protocol::Workload<'a>, + /// The nonzero request ID of Start, which the workload's events on the returned session + /// repeat. + pub(crate) request_id: u64, +} + +/// A failed restore. The VM's process, outputs and restore-ready listener remain the caller's. +#[derive(Debug)] +pub(crate) enum ControlRestoreError { + /// The restore failed before the guest's Ready, with the machine's phase, evidence and + /// source. + Machine(RestoreError), + /// The mapping table failed after the guest's Ready. + Maps(SessionError), + /// The workload's Start failed after the mapping table. + Start(SessionError), +} + +impl fmt::Display for ControlRestoreError { + fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result { + match self { + Self::Machine(error) => error.fmt(formatter), + Self::Maps(error) => write!(formatter, "the restored VM's mappings failed: {error}"), + Self::Start(error) => write!(formatter, "the restored VM's Start failed: {error}"), + } + } +} + +impl std::error::Error for ControlRestoreError { + fn source(&self) -> Option<&(dyn std::error::Error + 'static)> { + match self { + Self::Machine(error) => Some(error), + Self::Maps(error) | Self::Start(error) => Some(error), + } + } +} + +/// Restores the launched VM: drives its restore to the guest's Ready, in 60 seconds from the +/// call, then sends `maps` and starts `workload`, in 30 seconds from Ready, and returns the +/// session in which the workload's events arrive, no longer subject to `cancel`. +/// +/// `maps` is the mapping table that `record` announces. A failure stops nothing: the caller keeps +/// the VM's process, its outputs and its restore-ready listener, and decides what to do with +/// them. +#[cfg_attr( + not(test), + expect( + dead_code, + reason = "SL3.8a's `snapshot restore` restores sandboxes through it" + ) +)] +pub(crate) fn restore( + launched: &mut Launched, + record: &ActivationRecord, + workload: &Workload<'_>, + maps: &[HostMap], + cancel: &Cancel, +) -> Result { + let deadline = Instant::now() + READY_BUDGET; + let Launched { + process, + outputs, + capability, + ready, + .. + } = launched; + let watch = Watch { + process, + cancel, + interrupt: OnceLock::new(), + #[cfg(any(target_os = "linux", windows))] + connection: Mutex::new(None), + }; + let result = thread::scope(|scope| { + // Dropping the sender, also as a panic unwinds, ends the watcher. + let (_running, stopped) = mpsc::channel::<()>(); + scope.spawn(|| watch.run(stopped)); + let session = activate( + &watch, + ready.as_mut(), + outputs, + capability, + record, + workload, + deadline, + )?; + start(&watch, session, record, workload, maps) + }); + // The watcher may have ended the session's transfers after the last of them. + match watch.interrupted() { + Some(interrupt) if result.is_ok() => Err(ControlRestoreError::Start(interrupt.into())), + _ => result, + } +} + +/// Drives the machine to the guest's Ready, and returns the session in which it came. +fn activate( + watch: &Watch<'_>, + mut listener: Option<&mut ReadyListener>, + outputs: &VmOutputs, + capability: &Capability, + record: &ActivationRecord, + workload: &Workload<'_>, + deadline: Instant, +) -> Result { + let mut machine = RestoreMachine::new(record); + machine.input(RestoreInput::Spawned(watch.process.reference().clone())); + let payload = + protocol::encode_activate_payload(record, &workload.identity).unwrap_or_else(|error| { + machine.input(RestoreInput::OperationFailed { + phase: RestorePhase::Activate, + evidence: RestoreEvidence::InvalidSettings, + source: Some(Box::new(error)), + }); + Vec::new() + }); + let mut session = None; + let mut sessions = 0; + loop { + let action = machine + .next_action() + .expect("each step gives the machine an input"); + let (phase, result) = match action { + RestoreAction::Fail(mut error) => { + // The VM ran before the driver did, whatever the machine saw of it. + error.guest_may_have_run = true; + return Err(ControlRestoreError::Machine(error)); + } + RestoreAction::Done => return Ok(session.expect("Ready came in a session")), + RestoreAction::AwaitReadyEvent => ( + RestorePhase::ReadyEvent, + ready_event(watch, listener.as_deref_mut(), deadline), + ), + RestoreAction::Connect => { + session = None; + watch.release(); + let result = connect(watch, &outputs.endpoint, deadline).and_then(|transport| { + sessions += 1; + watch.register(&*transport)?; + let attached = + ControlSession::attach(transport, capability.as_bytes(), deadline)?; + session = Some(attached); + Ok(RestoreInput::Connected) + }); + (RestorePhase::Connect, result) + } + RestoreAction::SendActivate(_) => { + let attached = session.as_mut().expect("activation follows a connection"); + let result = attached.activate(&payload, deadline); + (RestorePhase::Activate, result.map(RestoreInput::GuestReady)) + } + }; + machine.input(match result { + Ok(input) => input, + Err(error) => watch.failed(&outputs.outcome, phase, error, sessions, deadline), + }); + } +} + +/// Sends the mapping table and starts the workload in the session in which the guest was ready. +fn start( + watch: &Watch<'_>, + mut session: ControlSession, + record: &ActivationRecord, + workload: &Workload<'_>, + maps: &[HostMap], +) -> Result { + let deadline = Instant::now() + START_BUDGET; + let announced = record.configuration().mappings; + if maps.len() != announced { + return Err(ControlRestoreError::Maps(SessionError::Protocol(format!( + "the activation record announces {announced} mappings, not {}", + maps.len() + )))); + } + if !maps.is_empty() { + watch + .check() + .and_then(|()| session.map_host_paths(maps, deadline)) + .map_err(|error| ControlRestoreError::Maps(watch.cause(error, deadline)))?; + } + let payload = protocol::encode_exec_payload(&workload.exec) + .map_err(|error| ControlRestoreError::Start(error.into()))?; + watch + .check() + .and_then(|()| session.send_request(APP_EXEC, workload.request_id, &payload, deadline)) + .map_err(|error| ControlRestoreError::Start(watch.cause(error, deadline)))?; + Ok(session) +} + +/// Receives the restore-ready event from the VM's held process. +fn ready_event( + watch: &Watch<'_>, + listener: Option<&mut ReadyListener>, + deadline: Instant, +) -> Result { + let listener = listener.ok_or_else(|| { + SessionError::Protocol("the launch has no restore-ready listener".to_owned()) + })?; + let mut event = Vec::new(); + while event.len() < READY_EVENT.len() { + let slice = watch.slice(deadline)?; + let open = listener + .receive(watch.process, &mut event, READY_EVENT.len() + 1, slice) + .map_err(|error| match error.kind() { + ErrorKind::Exited => SessionError::ProcessExited, + ErrorKind::Protocol => SessionError::Refused(io::Error::other(error)), + _ => SessionError::Io(io::Error::other(error)), + })?; + if !open { + watch.pause(deadline.min(Instant::now() + CLOSED_EXIT_WAIT))?; + break; + } + } + Ok(RestoreInput::ReadyEvent(event)) +} + +/// Connects to the VM's control endpoint, which its held process must serve. +fn connect( + watch: &Watch<'_>, + endpoint: &str, + deadline: Instant, +) -> Result, SessionError> { + let process = watch.process.reference(); + loop { + let slice = watch.slice(deadline)?; + match session::connect( + endpoint, + process.pid, + process.start_time, + Instant::now() + slice, + ) { + Err(SessionError::TimedOut) => {} + result => return result, + } + } +} + +/// Why a restore stopped waiting. +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +enum Interrupt { + /// The caller cancelled the restore. + Cancelled, + /// The VM's held process exited. + Exited, +} + +impl From for SessionError { + fn from(interrupt: Interrupt) -> Self { + match interrupt { + Interrupt::Cancelled => Self::Io(io::Error::new( + io::ErrorKind::Interrupted, + "the caller cancelled the restore", + )), + Interrupt::Exited => Self::ProcessExited, + } + } +} + +/// Races a restore's pending operations against the caller's cancellation and the VM's exit. +struct Watch<'a> { + /// The VM's held process. + process: &'a Process, + /// The caller's cancellation. + cancel: &'a Cancel, + /// The first interrupt observed, which stands. + interrupt: OnceLock, + /// A duplicate of the current session's connection, whose transfers an interrupt ends. + #[cfg(any(target_os = "linux", windows))] + connection: Mutex>, +} + +impl Watch<'_> { + /// Returns the restore's interrupt, once the caller cancelled it or its VM verifiably exited. + fn poll(&self) -> Option { + if self.cancel.load(Ordering::Acquire) { + let _ = self.interrupt.set(Interrupt::Cancelled); + } else if self.process.has_exited().unwrap_or(false) { + let _ = self.interrupt.set(Interrupt::Exited); + } + self.interrupted() + } + + /// Returns the interrupt observed so far, without checking for another. + fn interrupted(&self) -> Option { + self.interrupt.get().copied() + } + + /// Fails once the restore was interrupted. + fn check(&self) -> Result<(), SessionError> { + self.poll() + .map_or(Ok(()), |interrupt| Err(interrupt.into())) + } + + /// Returns the interrupt that ended a step with `error`, if any, after giving a lost session's + /// VM until [`POLL`] or `deadline` to show its exit, since a VM that exits ends its sessions + /// before its exit shows. + fn interrupt_behind(&self, error: &SessionError, deadline: Instant) -> Option { + if session_lost(error) { + let _ = self.pause(deadline.min(Instant::now() + POLL)); + } + self.poll() + } + + /// Returns the interrupt that ended a step with `error` as its failure, or `error`. + fn cause(&self, error: SessionError, deadline: Instant) -> SessionError { + self.interrupt_behind(&error, deadline) + .map_or(error, SessionError::from) + } + + /// Returns how long the next wait until `deadline` may take, or why the step ends. + fn slice(&self, deadline: Instant) -> Result { + self.check()?; + deadline + .checked_duration_since(Instant::now()) + .filter(|left| !left.is_zero()) + .map(|left| left.min(POLL)) + .ok_or(SessionError::TimedOut) + } + + /// Waits until `until`, unless the restore is interrupted first. + fn pause(&self, until: Instant) -> Result<(), SessionError> { + loop { + match self.slice(until) { + Ok(slice) => thread::sleep(slice), + Err(SessionError::TimedOut) => return Ok(()), + Err(error) => return Err(error), + } + } + } + + /// Lets an interrupt end `transport`'s pending transfers, in place of the previous session's. + fn register(&self, transport: &dyn Transport) -> Result<(), SessionError> { + #[cfg(any(target_os = "linux", windows))] + { + let handle = transport.try_clone_handle()?; + *self + .connection + .lock() + .unwrap_or_else(PoisonError::into_inner) = handle; + } + #[cfg(not(any(target_os = "linux", windows)))] + let _ = transport; + Ok(()) + } + + /// Forgets the previous session's connection, so that it closes with its session. + fn release(&self) { + #[cfg(any(target_os = "linux", windows))] + self.connection + .lock() + .unwrap_or_else(PoisonError::into_inner) + .take(); + } + + /// Checks every [`POLL`] until `stopped` disconnects, and once the restore is interrupted, ends + /// the session's pending transfers, at each check, since a Windows transfer that starts later + /// would not end. + fn run(&self, stopped: mpsc::Receiver<()>) { + while let Err(RecvTimeoutError::Timeout) = stopped.recv_timeout(POLL) { + #[cfg(any(target_os = "linux", windows))] + if self.poll().is_some() + && let Some(connection) = &*self + .connection + .lock() + .unwrap_or_else(PoisonError::into_inner) + { + // A failure leaves the transfer to its deadline. + let _ = platform::interrupt_transfers(connection); + } + } + } + + /// Turns a step's failure into the machine's input: the caller's cancellation, the VM's + /// verified exit with its outcome report, a lost session that another may replace, or a final + /// failure with its evidence. + fn failed( + &self, + outcome: &Path, + phase: RestorePhase, + error: SessionError, + sessions: usize, + deadline: Instant, + ) -> RestoreInput { + let lost = phase != RestorePhase::ReadyEvent && session_lost(&error); + match (self.interrupt_behind(&error, deadline), error) { + (Some(Interrupt::Cancelled), _) => RestoreInput::Cancelled, + (Some(Interrupt::Exited), _) | (None, SessionError::ProcessExited) => { + match report::read(outcome) { + Ok(report) => RestoreInput::Exited(report), + Err(error) => RestoreInput::OperationFailed { + phase, + evidence: RestoreEvidence::ReportUnreadable, + source: Some(Box::new(error)), + }, + } + } + (None, _) if lost && sessions < MAX_SESSIONS => RestoreInput::SessionLost, + (None, SessionError::TimedOut) => RestoreInput::Expired, + (None, error) => RestoreInput::OperationFailed { + phase, + evidence: match error { + SessionError::Refused(_) => RestoreEvidence::PeerMismatch, + SessionError::Protocol(_) | SessionError::AuthenticationFailed => { + RestoreEvidence::Malformed + } + _ if lost => RestoreEvidence::Reset, + _ => RestoreEvidence::Io, + }, + source: Some(Box::new(error)), + }, + } + } +} + +/// Returns whether `error` lost its session: a reset, or the end of its stream or transport. +fn session_lost(error: &SessionError) -> bool { + match error { + SessionError::Closed | SessionError::Reset => true, + SessionError::Io(error) => matches!( + error.kind(), + io::ErrorKind::BrokenPipe + | io::ErrorKind::ConnectionAborted + | io::ErrorKind::ConnectionReset + | io::ErrorKind::UnexpectedEof + | io::ErrorKind::WriteZero + ), + _ => false, + } +} + +#[cfg(test)] +mod tests; diff --git a/nvx-rs/src/host/restore_driver/tests.rs b/nvx-rs/src/host/restore_driver/tests.rs new file mode 100644 index 000000000..f32cc7dfa --- /dev/null +++ b/nvx-rs/src/host/restore_driver/tests.rs @@ -0,0 +1,695 @@ +#![cfg(any(target_os = "linux", windows))] + +use std::ffi::OsString; +use std::fs; +use std::path::PathBuf; + +use super::*; +use crate::activation::{ + Configuration, ExpectedResources, Export, ExportKind, Image, MachineIdMode, OverlayMode, + ReadyActivation, RestoreSection, TransactionId, Upper, View, +}; +use crate::host::process::Opened; +use crate::host::session::ExecEvent; +use crate::protocol::{ExitCategory, WorkloadEnvironment}; + +/// A stand-in for OpenVMM and its guest, which serves the control endpoint, sends the +/// restore-ready event, attaches the host with the capability of its standard input, and answers +/// NVXC requests as `case` asks, with `activation` in its Ready. It records each request in +/// `trace`, and the step at which it stalls in `stage`. +const PEER: &str = r#" +import json, os, socket, struct, sys, time +case, endpoint, ready, trace, stage, activation = sys.argv[1:] +capability = sys.stdin.buffer.read() +def stall(value): + with open(stage, 'w', encoding='ascii') as out: out.write(value) + time.sleep(3600) + sys.exit(0) +if os.name == 'nt': + import ctypes, msvcrt + from ctypes import wintypes + kernel = ctypes.WinDLL('kernel32', use_last_error=True) + kernel.CreateNamedPipeW.argtypes = [wintypes.LPCWSTR] + [wintypes.DWORD] * 6 + [ctypes.c_void_p] + kernel.CreateNamedPipeW.restype = wintypes.HANDLE + kernel.ConnectNamedPipe.argtypes = [wintypes.HANDLE, ctypes.c_void_p] + kernel.ConnectNamedPipe.restype = wintypes.BOOL + kernel.CreateFileW.argtypes = [wintypes.LPCWSTR, wintypes.DWORD, wintypes.DWORD, + ctypes.c_void_p, wintypes.DWORD, wintypes.DWORD, wintypes.HANDLE] + kernel.CreateFileW.restype = wintypes.HANDLE + def owned(handle): + if handle == ctypes.c_void_p(-1).value: raise ctypes.WinError(ctypes.get_last_error()) + return os.fdopen(msvcrt.open_osfhandle(handle, os.O_BINARY), 'r+b', buffering=0) + def bind(): + # The first instance of a duplex pipe for local clients, with small buffers. + handle = kernel.CreateNamedPipeW(endpoint.replace('/', '\\'), 0x80003, 8, 1, 4096, 4096, + 0, None) + return handle, owned(handle) + def accept(server): + handle, stream = server + if not kernel.ConnectNamedPipe(handle, None) and ctypes.get_last_error() != 535: + raise ctypes.WinError(ctypes.get_last_error()) + return stream + def send_ready(data): + stream = owned(kernel.CreateFileW(ready.replace('/', '\\'), 0x40000000, 0, None, 3, 0, + None)) + stream.write(data) + return stream +else: + def bind(): + listener = socket.socket(socket.AF_UNIX) + listener.bind(endpoint) + listener.listen() + return listener + def accept(server): + return server.accept()[0].makefile('rwb', buffering=0) + def send_ready(data): + peer = socket.socket(socket.AF_UNIX) + peer.connect(ready) + peer.sendall(data) + return peer +def exact(stream, count): + data = b'' + while len(data) < count: + part = stream.read(count - len(data)) + if not part: raise EOFError() + data += part + return data +def record(stream, kind, epoch, sequence, payload=b''): + stream.write(struct.pack('<4sHBB16sQQI', b'NVXS', 1, kind, 0, bytes([7]) * 16, epoch, + sequence, len(payload)) + payload) +def reply(stream, kind, rid, status, payload, epoch, sequence): + frame = struct.pack('<4sBBHQiI', b'NVXC', 1, kind, 0, rid, status, len(payload)) + payload + record(stream, 5, epoch, sequence, frame) +server = None if case == 'connect' else bind() +if case == 'stall': + held = accept(server) + stall(case) +if case == 'ready-accept': stall(case) +if case == 'ready-read': + held = send_ready(b'O') + stall(case) +if case != 'control-only': send_ready(b'OPENVMM_RESTORE_READY_V1\n').close() +if case == 'connect': stall(case) +for session in range(1, 4): + if os.name == 'nt' and session > 1: server = bind() + stream = accept(server) + try: + header = exact(stream, 44) + assert header[:4] == b'NVXS' and header[6] == 2 + assert exact(stream, struct.unpack(' PathBuf { + let name = if cfg!(windows) { + "python.exe" + } else { + "python3" + }; + std::env::split_paths(&std::env::var_os("PATH").unwrap()) + .map(|directory| directory.join(name)) + .find(|path| path.is_file() && !path.to_string_lossy().contains("WindowsApps")) + .expect("Python is on PATH") +} + +fn record() -> ActivationRecord { + ActivationRecord::new( + Some(RestoreSection { + transaction: TransactionId::from_bytes([1; 16]), + }), + Configuration { + image: Image::Erofs { uuid: None }, + upper: Upper::Tmpfs { + max_memory_percent: 50, + }, + overlay: OverlayMode::TrustedLowers, + export: Some(Export { + kind: ExportKind::Aggregate, + }), + mappings: 1, + network: None, + view: View { + hostname: b"restore-test".to_vec(), + machine_id: MachineIdMode::Generated, + }, + resources: ExpectedResources { + cpus: 2, + memory: 1 << 30, + }, + }, + ) + .unwrap() +} + +fn workload() -> Workload<'static> { + Workload { + identity: WorkloadIdentity::DEFAULT, + exec: protocol::Workload { + argv: vec!["/bin/true".to_owned()], + timeout_ms: 0, + cwd: None, + environment: WorkloadEnvironment::Default, + }, + request_id: 19, + } +} + +fn maps() -> Vec { + vec![HostMap { + source: "entry".to_owned(), + target: "/mapped".to_owned(), + read_only: true, + }] +} + +fn hex(bytes: &[u8]) -> String { + bytes.iter().map(|byte| format!("{byte:02x}")).collect() +} + +/// Holds `process` again, independently of its launch. +fn reopen(process: &Process) -> Process { + let Opened::Running(process) = Process::open(process.reference()).unwrap() else { + panic!("the process runs"); + }; + process +} + +/// Waits until a [`PEER`] whose stage file is `file` stalls at `stage`. +fn wait_for(file: &Path, stage: &str) { + let deadline = Instant::now() + Duration::from_secs(20); + while fs::read_to_string(file).ok().as_deref() != Some(stage) { + assert!(Instant::now() < deadline, "the peer did not reach {stage}"); + thread::sleep(Duration::from_millis(10)); + } +} + +/// A launch whose VM is a stand-in that the test ends when it drops the launch. +struct Vm { + launched: Launched, + directory: tempfile::TempDir, +} + +impl Vm { + /// Spawns `command`, or the stand-in VMM without one, as a VM launched in a directory of its + /// own, and binds its restore-ready listener first when `restore` asks for one. + fn spawn( + restore: bool, + command: impl FnOnce(&Path, &VmOutputs, &str) -> Vec, + ) -> Self { + let (directory, vm) = crate::host::vm_directory::tests::vm_dir(); + let outputs = vm.outputs(false).unwrap(); + let ready = restore.then(|| ReadyListener::bind(&vm).unwrap()); + let endpoint = ready.as_ref().map_or("", |ready| ready.endpoint()); + let arguments = command(directory.path(), &outputs, endpoint); + let command = (!arguments.is_empty()).then(|| (python(), arguments)); + Self { + launched: Launched::stand_in(&vm, outputs, ready, command), + directory, + } + } +} + +impl Drop for Vm { + fn drop(&mut self) { + let _ = self.launched.process.kill(); + let _ = self + .launched + .process + .wait(Instant::now() + Duration::from_secs(10)); + } +} + +/// A [`PEER`], launched as a restored VM. +struct Peer { + vm: Vm, + stage: PathBuf, + trace: PathBuf, +} + +impl Peer { + fn start(case: &str) -> Self { + let vm = Vm::spawn(true, |directory, outputs, ready| { + let activation = hex(&ReadyActivation::of(&record()).encode()); + ["-c", PEER, case, &outputs.endpoint, ready] + .map(OsString::from) + .into_iter() + .chain(["trace", "stage"].map(|name| directory.join(name).into_os_string())) + .chain([OsString::from(activation)]) + .collect() + }); + Self { + stage: vm.directory.path().join("stage"), + trace: vm.directory.path().join("trace"), + vm, + } + } + + /// Returns each NVXC request that the peer received, in order. + fn requests(&self) -> Vec { + fs::read_to_string(&self.trace) + .unwrap_or_default() + .lines() + .map(|line| serde_json::from_str(line).unwrap()) + .collect() + } + + /// Returns the kinds of [`requests`](Self::requests). + fn kinds(&self) -> Vec { + self.requests() + .iter() + .map(|request| request["kind"].as_u64().unwrap()) + .collect() + } +} + +fn machine(error: ControlRestoreError) -> RestoreError { + match error { + ControlRestoreError::Machine(error) => error, + error => panic!("not a failure of the machine: {error:?}"), + } +} + +#[test] +fn a_lost_session_is_replaced_with_the_same_activate_message_and_the_live_session_returns() { + let mut peer = Peer::start("reconnect"); + let cancel = Cancel::new(false); + let workload = workload(); + let mut session = restore( + &mut peer.vm.launched, + &record(), + &workload, + &maps(), + &cancel, + ) + .unwrap(); + // The returned session no longer observes the restore's cancellation. + cancel.store(true, Ordering::Release); + let deadline = Some(Instant::now() + Duration::from_secs(5)); + assert_eq!( + session + .next_exec_event(workload.request_id, deadline) + .unwrap(), + ExecEvent::Stdout(b"restored".to_vec()) + ); + assert_eq!( + session + .next_exec_event(workload.request_id, deadline) + .unwrap(), + ExecEvent::Exit { + category: ExitCategory::Exit, + status: 0 + } + ); + let requests = peer.requests(); + let activate = hex(&protocol::encode_activate_payload(&record(), &workload.identity).unwrap()); + let sessions: Vec<_> = requests + .iter() + .filter(|request| request["kind"] == 8) + .map(|request| { + assert_eq!(request["payload"], activate.as_str()); + request["session"].as_u64().unwrap() + }) + .collect(); + assert_eq!(sessions, [1, 2, 3]); + assert_eq!(peer.kinds(), [8, 8, 8, 6, 2]); + assert_eq!(requests[4]["id"], workload.request_id); + assert!(!peer.vm.launched.process.has_exited().unwrap()); + assert!(peer.vm.launched.ready.is_some()); +} + +#[test] +fn a_third_lost_session_and_a_malformed_ready_are_final() { + for (case, phase, evidence, requests) in [ + ( + "session-limit", + RestorePhase::Activate, + RestoreEvidence::Reset, + 3, + ), + ( + "malformed", + RestorePhase::Ready, + RestoreEvidence::Malformed, + 1, + ), + ] { + let mut peer = Peer::start(case); + let error = machine( + restore( + &mut peer.vm.launched, + &record(), + &workload(), + &maps(), + &Cancel::new(false), + ) + .err() + .unwrap(), + ); + assert_eq!(error.phase, phase, "{case}"); + assert_eq!(error.evidence, evidence, "{case}"); + assert!(error.guest_may_have_run, "{case}"); + assert_eq!(peer.requests().len(), requests, "{case}"); + assert!(!peer.vm.launched.process.has_exited().unwrap(), "{case}"); + } +} + +#[test] +fn neither_an_unverified_restore_ready_client_nor_control_server_gets_the_activation() { + for ready_mismatch in [true, false] { + let mut peer = Peer::start(if ready_mismatch { + "reconnect" + } else { + "connect" + }); + let mut other = Peer::start("control-only"); + if ready_mismatch { + std::mem::swap( + &mut peer.vm.launched.process, + &mut other.vm.launched.process, + ); + } else { + peer.vm.launched.outputs.endpoint = other.vm.launched.outputs.endpoint.clone(); + } + let error = machine( + restore( + &mut peer.vm.launched, + &record(), + &workload(), + &maps(), + &Cancel::new(false), + ) + .err() + .unwrap(), + ); + assert_eq!(error.evidence, RestoreEvidence::PeerMismatch); + assert_eq!( + error.phase, + if ready_mismatch { + RestorePhase::ReadyEvent + } else { + RestorePhase::Connect + } + ); + assert!(peer.requests().is_empty() && other.requests().is_empty()); + } +} + +/// How a test ends a restore's pending operation. +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +enum End { + Deadline, + Cancel, + Exit, +} + +/// The cases of [`PEER`] that stall a restore's pending accept or read, with the phase that +/// fails, or none for the mapping table. +const STALLS: [(&str, Option); 6] = [ + ("ready-accept", Some(RestorePhase::ReadyEvent)), + ("ready-read", Some(RestorePhase::ReadyEvent)), + ("connect", Some(RestorePhase::Connect)), + ("attach-read", Some(RestorePhase::Connect)), + ("activate-read", Some(RestorePhase::Activate)), + ("maps-read", None), +]; + +/// Restores against a peer that stalls at `case`, ends the restore as `end` says once the peer +/// stalls, and checks how and when the restore fails. +fn check_ended(case: &str, phase: Option, end: End) { + let mut peer = Peer::start(case); + let cancel = Cancel::new(false); + let stage = peer.stage.clone(); + let vm = reopen(&peer.vm.launched.process); + let began = Instant::now(); + let error = thread::scope(|scope| { + if end != End::Deadline { + let cancel = &cancel; + scope.spawn(move || { + wait_for(&stage, case); + thread::sleep(Duration::from_millis(50)); + if end == End::Cancel { + cancel.store(true, Ordering::Release); + } else { + vm.kill().unwrap(); + } + }); + } + restore( + &mut peer.vm.launched, + &record(), + &workload(), + &maps(), + &cancel, + ) + .err() + .unwrap() + }); + let elapsed = began.elapsed(); + if end == End::Deadline { + // Ready has 60 seconds from the call, and the mapping table and Start 30 from Ready. + let budget = Duration::from_secs(if phase.is_some() { 60 } else { 30 }); + assert!( + budget <= elapsed && elapsed < budget + Duration::from_secs(30), + "{case}: {elapsed:?}, {error:?}" + ); + } else { + assert!( + elapsed < Duration::from_secs(10), + "{case}/{end:?}: {elapsed:?}, {error:?}" + ); + } + match (phase, error) { + (Some(phase), ControlRestoreError::Machine(error)) => { + assert_eq!(error.phase, phase, "{case}/{end:?}"); + let evidence = match end { + End::Deadline => RestoreEvidence::Expired, + End::Cancel => RestoreEvidence::Cancelled, + End::Exit => RestoreEvidence::Exited, + }; + assert_eq!(error.evidence, evidence, "{case}/{end:?}"); + } + (None, ControlRestoreError::Maps(error)) => match (end, error) { + (End::Deadline, SessionError::TimedOut) | (End::Exit, SessionError::ProcessExited) => {} + (End::Cancel, SessionError::Io(error)) + if error.kind() == io::ErrorKind::Interrupted => {} + (end, error) => panic!("{case}/{end:?}: {error:?}"), + }, + (_, error) => panic!("{case}/{end:?}: {error:?}"), + } + if end != End::Exit { + assert!(!peer.vm.launched.process.has_exited().unwrap(), "{case}"); + } +} + +#[test] +fn ready_has_sixty_seconds_and_start_thirty_from_ready_for_each_pending_accept_and_read() { + // The cases wait out their budgets together. + thread::scope(|scope| { + for (case, phase) in STALLS { + scope.spawn(move || check_ended(case, phase, End::Deadline)); + } + }); +} + +#[test] +fn cancellation_ends_each_pending_accept_and_read() { + for (case, phase) in STALLS { + check_ended(case, phase, End::Cancel); + } +} + +#[test] +fn the_verified_exit_of_the_vm_ends_each_pending_accept_and_read() { + for (case, phase) in STALLS { + check_ended(case, phase, End::Exit); + } +} + +#[test] +fn the_deadline_cancellation_and_the_verified_exit_end_a_pending_write() { + for end in [End::Deadline, End::Cancel, End::Exit] { + // The peer never reads what the restore writes, so a large write stays pending. + let peer = Peer::start("stall"); + let vm = Vm::spawn(false, |_, _, _| Vec::new()); + let process = &vm.launched.process; + let cancel = Cancel::new(false); + let watch = Watch { + process, + cancel: &cancel, + interrupt: OnceLock::new(), + connection: Mutex::new(None), + }; + let server = peer.vm.launched.process.reference(); + let mut transport = session::connect( + &peer.vm.launched.outputs.endpoint, + server.pid, + server.start_time, + Instant::now() + Duration::from_secs(10), + ) + .unwrap(); + watch.register(&*transport).unwrap(); + let timeout = Duration::from_millis(if end == End::Deadline { 500 } else { 20_000 }); + let began = Instant::now(); + let error = thread::scope(|scope| { + let (_running, stopped) = mpsc::channel::<()>(); + scope.spawn(|| watch.run(stopped)); + if end != End::Deadline { + let (stage, cancel) = (&peer.stage, &cancel); + scope.spawn(move || { + wait_for(stage, "stall"); + thread::sleep(Duration::from_millis(200)); + if end == End::Cancel { + cancel.store(true, Ordering::Release); + } else { + process.kill().unwrap(); + } + }); + } + transport + .write_all(&vec![7; 16 << 20], Some(timeout)) + .unwrap_err() + }); + let elapsed = began.elapsed(); + let interrupt = match end { + End::Deadline => { + assert_eq!(error.kind(), io::ErrorKind::TimedOut); + assert!(elapsed >= timeout, "{elapsed:?}"); + None + } + End::Cancel => Some(Interrupt::Cancelled), + End::Exit => Some(Interrupt::Exited), + }; + assert!(elapsed < Duration::from_secs(10), "{end:?}: {elapsed:?}"); + assert_eq!(watch.interrupted(), interrupt); + } +} + +#[test] +fn mapping_and_start_failures_are_told_apart_from_the_machines() { + for case in ["maps-error", "start-error", "maps-count"] { + let mut peer = Peer::start(case); + let mut workload = workload(); + if case == "start-error" { + // EXEC refuses a workload without a program. + workload.exec.argv.clear(); + } + let maps = if case == "maps-count" { + Vec::new() + } else { + maps() + }; + let error = restore( + &mut peer.vm.launched, + &record(), + &workload, + &maps, + &Cancel::new(false), + ) + .err() + .unwrap(); + match (case, &error) { + ("maps-error" | "maps-count", ControlRestoreError::Maps(SessionError::Protocol(_))) + | ("start-error", ControlRestoreError::Start(SessionError::Protocol(_))) => {} + (case, error) => panic!("{case}: {error:?}"), + } + assert!(!peer.kinds().contains(&2), "{case}"); + assert!(!peer.vm.launched.process.has_exited().unwrap(), "{case}"); + } + // A restore without an export sends no mapping table, and starts its workload at once. + let mut configuration = record().configuration().clone(); + configuration.export = None; + configuration.mappings = 0; + let record = ActivationRecord::new(record().restore().cloned(), configuration).unwrap(); + let mut peer = Peer::start("reconnect"); + // The session stays open until the peer answered its Start. + let _session = restore( + &mut peer.vm.launched, + &record, + &workload(), + &[], + &Cancel::new(false), + ) + .unwrap(); + wait_for(&peer.stage, "started"); + assert_eq!(peer.kinds(), [8, 8, 8, 2]); +} + +#[test] +fn a_cold_record_and_a_root_workload_fail_before_the_vm_is_contacted() { + let cold = ActivationRecord::new(None, record().configuration().clone()).unwrap(); + let mut root = workload(); + root.identity = WorkloadIdentity { + uid: 0, + gid: 0, + create_account: false, + }; + for (record, workload, phase, evidence) in [ + ( + cold, + workload(), + RestorePhase::ReadyEvent, + RestoreEvidence::InvalidActivate, + ), + ( + record(), + root, + RestorePhase::Activate, + RestoreEvidence::InvalidSettings, + ), + ] { + let mut peer = Peer::start("connect"); + let error = machine( + restore( + &mut peer.vm.launched, + &record, + &workload, + &maps(), + &Cancel::new(false), + ) + .err() + .unwrap(), + ); + assert_eq!((error.phase, error.evidence), (phase, evidence)); + assert!(error.guest_may_have_run); + assert!(peer.requests().is_empty()); + assert!(!peer.vm.launched.process.has_exited().unwrap()); + } +} diff --git a/nvx-rs/src/host/session.rs b/nvx-rs/src/host/session.rs index ecfc9b2cc..74b23fe57 100644 --- a/nvx-rs/src/host/session.rs +++ b/nvx-rs/src/host/session.rs @@ -185,6 +185,28 @@ impl ControlSession { ) } + /// Sends nvx-init's activate message, whose `payload` carries a restore's activation record + /// and workload identity, and returns the activation that the guest's Ready reports. + #[cfg(feature = "cli")] + pub(crate) fn activate( + &mut self, + payload: &[u8], + deadline: Instant, + ) -> Result { + let request_id = request_id()?; + self.send_app(protocol::APP_ACTIVATE, request_id, payload, Some(deadline))?; + match self.read_reply(request_id, Some(deadline))? { + (APP_READY, 0, ready) => { + crate::activation::ReadyActivation::decode(&ready).map_err(|error| { + SessionError::Protocol(format!("the guest's Ready is invalid: {error}")) + }) + } + _ => Err(SessionError::Protocol( + "the guest agent did not activate the restore".to_owned(), + )), + } + } + /// Asks the guest agent which control features its image provides. /// /// An agent that predates the request refuses it as an unsupported operation, which means diff --git a/nvx-rs/src/host/session/tests.rs b/nvx-rs/src/host/session/tests.rs index 2b90794a4..e1753b39a 100644 --- a/nvx-rs/src/host/session/tests.rs +++ b/nvx-rs/src/host/session/tests.rs @@ -624,6 +624,41 @@ fn capture_refuses_invalid_acknowledgements() { } } +#[cfg(feature = "cli")] +#[test] +fn activate_sends_its_payload_and_returns_the_activation_that_ready_reports() { + let (mut session, requests) = attached_recording(|kind, id| { + assert_eq!(kind, protocol::APP_ACTIVATE); + encode_app(APP_READY, id, 0, &[[1].as_slice(), &[9; 16]].concat()) + }); + let ready = session.activate(b"activation", deadline()).unwrap(); + assert_eq!( + ready.transaction, + Some(crate::activation::TransactionId::from_bytes([9; 16])) + ); + assert_eq!( + *requests.lock().unwrap(), + [(protocol::APP_ACTIVATE, b"activation".to_vec())] + ); +} + +#[cfg(feature = "cli")] +#[test] +fn activate_refuses_a_reply_other_than_a_valid_ready() { + let answers: [fn(u8, u64) -> Vec; 4] = [ + |_, id| encode_app(APP_ERROR, id, 22, b"invalid-request"), + |_, id| encode_app(APP_READY, id ^ 1, 0, &[0]), + |_, id| encode_app(APP_READY, id, 1, &[0]), + |_, id| encode_app(APP_READY, id, 0, &[2]), + ]; + for answer in answers { + let error = attached(answer) + .activate(b"activation", deadline()) + .unwrap_err(); + assert!(matches!(error, SessionError::Protocol(_)), "{error:?}"); + } +} + #[test] fn features_report_what_the_guest_provides() { let mut session = attached(|kind, id| { diff --git a/nvx-rs/src/host/vm/tests.rs b/nvx-rs/src/host/vm/tests.rs index 53b42f697..c434c6b8f 100644 --- a/nvx-rs/src/host/vm/tests.rs +++ b/nvx-rs/src/host/vm/tests.rs @@ -28,6 +28,27 @@ use crate::uuid::Uuid; #[cfg(any(target_os = "linux", windows))] mod reattach; +#[cfg(all(feature = "cli", any(target_os = "linux", windows)))] +impl Launched { + /// Spawns `command`, or the stand-in VMM without one, as a cold launch in `dir` spawns + /// OpenVMM, with these outputs and restore-ready listener. + pub(crate) fn stand_in( + dir: &VmDir, + outputs: VmOutputs, + ready: Option, + command: Option<(PathBuf, Vec)>, + ) -> Self { + let mut prepared = stand_in(dir); + prepared.outputs = outputs; + prepared.ready = ready; + if let Some((program, arguments)) = command { + prepared.program = program; + prepared.arguments = arguments; + } + spawn(prepared).unwrap() + } +} + #[test] fn the_reserved_names_are_the_documented_ones_and_leave_the_openvmm_prefix_to_openvmm() { let names = [ diff --git a/nvx-rs/src/protocol/app.rs b/nvx-rs/src/protocol/app.rs index 2feb2c3ff..872522658 100644 --- a/nvx-rs/src/protocol/app.rs +++ b/nvx-rs/src/protocol/app.rs @@ -64,11 +64,12 @@ pub(crate) const APP_CAPTURE: u8 = 7; #[cfg_attr( not(any( test, + feature = "cli", all(feature = "guest", target_os = "linux", target_arch = "x86_64") )), expect( dead_code, - reason = "SL4.3 sends activation requests to nvx-init in SL7.7b" + reason = "only the CLI's restore driver sends it, and only nvx-init reads it" ) )] pub(crate) const APP_ACTIVATE: u8 = 8; @@ -188,8 +189,11 @@ pub(crate) fn decode_capture_payload(payload: &[u8]) -> Result