From 38d7b9bfe5bfb2f9169cf4aa1927b0836b1ce0e2 Mon Sep 17 00:00:00 2001 From: Enrique Saurez Date: Tue, 20 Oct 2026 09:40:26 -0700 Subject: [PATCH] nvx-rs: run the control session on the shared NVXS session's host role ControlSession parsed the control endpoint's NVXS records itself, and tracked the broker's identity and sequence numbers inline with NVXC decoding. Make it the NVXC adapter over the host role of the shared session, nvx::host::nvxs::HostConnection: each NVXC frame travels in one DATA record, and a failure keeps its variant and, for a record with one fault, its message, so that the CLI reports what it reported before. The shared session checks more than the control session did: the sequence numbers of WAIT and READY, the code that an ERROR carries, and the identity and sequence number of a RESET. A failure that leaves the stream unusable, such as a reset, a record that the protocol does not allow, or a write that times out, now ends the session, and later calls fail. OpenVMM's broker refuses a capability by closing the stream, and never sends the host a RESET. The fake OpenVMM refuses one with an ERROR record, and the managed agent's loopback broker resets the host; both now send these records as the shared session requires: an ERROR carries its code, and a RESET to the host carries the session's next epoch at the host leg's next sequence number. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> Copilot-Session: 086f45a6-c39c-4704-85ef-1fa7f8b8b672 --- nvx-rs/README.md | 4 +- nvx-rs/src/guest/managed_agent/loopback.rs | 7 +- nvx-rs/src/host/cli/session.rs | 4 +- nvx-rs/src/host/cli/session/tests.rs | 4 +- nvx-rs/src/host/mod.rs | 4 +- nvx-rs/src/host/session.rs | 581 ++++++++++++++------- nvx-rs/src/testing/fake_openvmm.rs | 13 +- 7 files changed, 408 insertions(+), 209 deletions(-) diff --git a/nvx-rs/README.md b/nvx-rs/README.md index 824db9f77..fd089d2d4 100644 --- a/nvx-rs/README.md +++ b/nvx-rs/README.md @@ -60,9 +60,9 @@ nvx-rs/ | `nvx::host::filesystem` | The `ExportPlan` of host paths through OpenVMM's aggregate virtio-fs device, the host-mapping table, the reserved guest targets, and `guest_path` and `resolve_guest_path` (also at `nvx::`) | | `nvx::host::store` | `SandboxStore`: the runtime records of one sandbox in a private directory with one lock, which every host that supervises the sandbox shares. The `RuntimeRecord` names the running VM: the `ProcessRef` of its VMM, its control endpoint, and the `RuntimeIdentity` that it was launched with, the identities of its OpenVMM, kernel, and initramfs from install or staging metadata and its boot and control contract versions, which decide whether another build of the host may reattach to it. The `LaunchJournal` of a launch holds a marker written before the spawn, the VMM's `ProcessRef` right after it, and an inherited claim on the VMM's log, which proves that a launch that never published its endpoint has exited, so that recovery ends an interrupted launch and signals only a VMM that it holds through a verified `Process`. Records are JSON in one versioned format, and records in any other format fail rather than being migrated | | `nvx::host::instance` | `Instance`: a VM in a directory of its own. It starts OpenVMM with a fresh capability and waits until the guest is ready and has mounted the mapped paths, reconciles recorded state, recovers interrupted launches, runs workloads and pumps their output, and stops the VM with a forced fallback | -| `nvx::host::nvxs` | `HostConnection`: the host's role of the NVXS session over a platform transport. It attaches with the launch capability, waiting while the broker asks it to, 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 | +| `nvx::host::nvxs` | `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, 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 | | `nvx::host::process` | The detached launch of OpenVMM, and verified process references. A `ProcessRef` names one process by its ID, its start time, and the `BootId` of the host boot during which it started. `Process::open` checks a reference and holds its process through a pidfd on Linux or a process handle on Windows, whether or not the caller started it, so that waits sleep until the process exits and termination never reaches a process that reuses its ID. A reference from an earlier boot names an exited process, and a process whose identity this user cannot check, such as another user's, makes `Process::open` fail rather than report an exit. Linux hosts without pidfds poll the reference instead, and cannot terminate the process | -| `nvx::host::session`, `platform`, `report` | The authenticated control session, Linux and Windows primitives, and OpenVMM's outcome report | +| `nvx::host::session`, `platform`, `report` | The authenticated control session, which carries NVX's control protocol, `NVXC`, one frame per DATA record of a `HostConnection`; Linux and Windows primitives; and OpenVMM's outcome report | | `nvx::host::sandbox` | The launch profile of `nvx.py sandbox`, ported from the one that `scripts/nvx_tools/sandbox.py` held: `SandboxLaunch`, with the EROFS distro `SandboxLayer`, the one layer that a sandbox takes, the ext4 scratch disk, the entrypoint, arguments, hostname, workload identity, and resource limits, and its `--microvm-sandbox-block` and `--microvm-workload-identity` arguments and kernel command line, within the 1024-byte budget that OpenVMM leaves it. `mount`: `SandboxMount` live shares of host directories and files, their guest targets and the reserved ones, their access policies and owner, the aggregate of several shares, and the checks of their host paths. `options`: the checks that `nvx.py sandbox` and `nvx.py run` apply to their options: the attribution of `--mount-deny`, `--mount-allow`, and `--mount-write` to shares, the systemd policy, the targets of `nvx.py run` shares, and the managed exec environment; see [Python parity](#python-parity) | | `nvx::testing::fake_openvmm` | A fake OpenVMM that serves the control console and runs scripted workloads without a VM, for direct launches and for the managed and one-shot sandboxes of `nvx-rs sandbox` and `nvx.py sandbox` | | `nvx::host::cli` | Feature `cli`: the `nvx-rs` executable; see [The nvx-rs CLI](#the-nvx-rs-cli) | diff --git a/nvx-rs/src/guest/managed_agent/loopback.rs b/nvx-rs/src/guest/managed_agent/loopback.rs index b5f5827e3..389f5397c 100644 --- a/nvx-rs/src/guest/managed_agent/loopback.rs +++ b/nvx-rs/src/guest/managed_agent/loopback.rs @@ -141,10 +141,13 @@ impl BrokerCore { .epoch .checked_add(1) .ok_or_else(|| io::Error::from(Errno::EOVERFLOW))?; - self.host_output.clear(); self.receive_credit = 0; self.reset_guest()?; - self.queue_host(OUTER_RESET, 0, &[]) + // The RESET ends the host's attachment with the session's new identity, after the + // records that the host has not read, at the next sequence number of the host's leg. + let sequence = self.host_send_sequence; + self.host_send_sequence = self.host_send_sequence.wrapping_add(1); + self.queue_host(OUTER_RESET, sequence, &[]) } fn reset_guest(&mut self) -> io::Result<()> { diff --git a/nvx-rs/src/host/cli/session.rs b/nvx-rs/src/host/cli/session.rs index 27d70032a..3cacb01a8 100644 --- a/nvx-rs/src/host/cli/session.rs +++ b/nvx-rs/src/host/cli/session.rs @@ -487,8 +487,8 @@ impl Session { /// Returns Python's message for a violation of the protocol that `nvx::host::session` reports. /// -/// The two sessions check the same records in the same order, and their messages differ only -/// where Python folds a case into a broader one. +/// The two sessions report the same violations of records that have one fault, and their +/// messages differ only where Python folds a case into a broader one. fn protocol_message(message: String) -> String { match message.as_str() { "control attach response carried a payload" => { diff --git a/nvx-rs/src/host/cli/session/tests.rs b/nvx-rs/src/host/cli/session/tests.rs index 8ea9b84f9..109d0221c 100644 --- a/nvx-rs/src/host/cli/session/tests.rs +++ b/nvx-rs/src/host/cli/session/tests.rs @@ -510,7 +510,9 @@ fn session_reset_is_not_a_closed_endpoint() { // closure with which OpenVMM shuts down. let (mut session, worker) = session_with(|guest| { let _ = guest.request(); - let reset = encode_outer(OUTER_RESET, &INSTANCE, 1, 0, &[]).unwrap(); + // The reset starts the session's next epoch, at the next sequence number of the host's + // leg. + let reset = encode_outer(OUTER_RESET, &INSTANCE, 2, guest.sequence + 1, &[]).unwrap(); guest.stream.write_all(&reset, None).unwrap(); let _ = guest.record(); }); diff --git a/nvx-rs/src/host/mod.rs b/nvx-rs/src/host/mod.rs index 1a26c2377..aecd50489 100644 --- a/nvx-rs/src/host/mod.rs +++ b/nvx-rs/src/host/mod.rs @@ -14,8 +14,8 @@ //! - [`store`] keeps the runtime records of a sandbox in a private directory with one lock: the //! runtime record of its VM, with the VM's runtime identity, and the journal of its launch, //! through which an interrupted launch is recovered without signaling an unverified process. -//! - [`session`] speaks the authenticated control protocol, and [`nvxs`] runs the host's role of -//! the NVXS session, which carries it and any other application protocol, over a platform +//! - [`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. //! - [`process`] launches OpenVMM processes. Its verified process references name a process by diff --git a/nvx-rs/src/host/session.rs b/nvx-rs/src/host/session.rs index 38edcc8aa..219567c0c 100644 --- a/nvx-rs/src/host/session.rs +++ b/nvx-rs/src/host/session.rs @@ -1,17 +1,19 @@ -//! Host side of an authenticated control session with a sandbox's guest agent. +//! Host side of an authenticated control session with a sandbox's guest agent: NVX's control +//! protocol, `NVXC`, over the host's role of the shared NVXS session, [`HostConnection`]. use std::fmt; use std::io; use std::thread; use std::time::{Duration, Instant}; +use super::nvxs::{HostConnection, HostConnectionError}; use super::platform::{self, Transport}; +use crate::protocol::nvxs::{self, RecordError, RecordType}; use crate::protocol::{ self, APP_CANCEL, APP_ERROR, APP_EXEC, APP_EXIT, APP_FEATURES, APP_HEADER_LEN, APP_MAPS, APP_PING, APP_READY, APP_STDERR, APP_STDOUT, APP_STOP, APP_STOPPED, CAPABILITY_LEN, - ExitCategory, GuestFeatures, HostMap, OUTER_DATA, OUTER_ERROR, OUTER_HEADER_LEN, - OUTER_HOST_ATTACH, OUTER_MAX_PAYLOAD, OUTER_READY, OUTER_RESET, OUTER_WAIT, OuterHeader, - ProtocolError, UNSUPPORTED_OPERATION, Workload, + ExitCategory, GuestFeatures, HostMap, OUTER_MAX_PAYLOAD, ProtocolError, UNSUPPORTED_OPERATION, + Workload, }; /// Failure of a control session. @@ -127,68 +129,29 @@ pub enum ExecEvent { }, } -/// Authenticated control session. +/// Authenticated control session: `NVXC` frames, each in one DATA record of the host's NVXS +/// connection to OpenVMM's broker. /// /// Received bytes are buffered until a whole record is available, so a read that times out -/// loses nothing and callers can poll with short deadlines. +/// loses nothing and callers can poll with short deadlines. A failure that leaves the stream +/// unusable, such as a reset or a record that the protocol does not allow, ends the session, and +/// later calls fail. pub struct ControlSession { - /// Platform transport connected to OpenVMM's control endpoint. - transport: Box, - /// Bytes received beyond the last complete outer record. - received: Vec, - /// Random instance identity assigned by the broker during attachment. - instance_id: [u8; 16], - /// Broker connection epoch assigned during attachment. - epoch: u64, - /// Sequence number of the next host-to-guest outer record. - send_sequence: u64, - /// Sequence number expected on the next guest-to-host outer record. - receive_sequence: u64, + /// The host's NVXS connection to OpenVMM's control endpoint. + connection: HostConnection, } impl ControlSession { - /// Authenticates with the capability OpenVMM received at launch. + /// Authenticates with the capability OpenVMM received at launch, and waits until `deadline` + /// while OpenVMM's broker waits for the guest to acknowledge the session. pub fn attach( transport: Box, capability: &[u8; CAPABILITY_LEN], deadline: Instant, ) -> Result { - let mut session = Self { - transport, - received: Vec::new(), - instance_id: [0; 16], - epoch: 0, - send_sequence: 0, - receive_sequence: 0, - }; - let record = protocol::encode_outer(OUTER_HOST_ATTACH, &[0; 16], 0, 0, capability)?; - session - .transport - .write_all(&record, remaining(Some(deadline))?)?; - loop { - let (header, payload) = session.read_outer(Some(deadline))?; - if !payload.is_empty() { - return Err(SessionError::Protocol( - "control attach response carried a payload".to_owned(), - )); - } - match header.record_type { - // Another client holds the broker's single host slot. - OUTER_WAIT => continue, - OUTER_ERROR => return Err(SessionError::AuthenticationFailed), - OUTER_READY if header.instance_id != [0; 16] && header.epoch != 0 => { - session.instance_id = header.instance_id; - session.epoch = header.epoch; - session.receive_sequence = header.sequence.wrapping_add(1); - return Ok(session); - } - _ => { - return Err(SessionError::Protocol( - "control endpoint returned an invalid attach response".to_owned(), - )); - } - } - } + let connection = HostConnection::attach(transport, capability, deadline) + .map_err(|error| failure(error, Stage::Attach))?; + Ok(Self { connection }) } /// Waits until the guest agent acknowledges readiness. @@ -321,7 +284,7 @@ impl ControlSession { self.read_app(deadline) } - /// Encodes and sends one application frame in the next sequenced outer record. + /// Encodes and sends one application frame in the next DATA record. fn send_app( &mut self, kind: u8, @@ -330,38 +293,27 @@ impl ControlSession { deadline: Option, ) -> Result<(), SessionError> { let frame = protocol::encode_app(kind, request_id, 0, payload); - let record = protocol::encode_outer( - OUTER_DATA, - &self.instance_id, - self.epoch, - self.send_sequence, - &frame, - )?; - self.transport.write_all(&record, remaining(deadline)?)?; - self.send_sequence = self.send_sequence.wrapping_add(1); - Ok(()) + // The guest agent takes each DATA record as one frame, so a frame must fit one. + if frame.len() > OUTER_MAX_PAYLOAD { + return Err(SessionError::Protocol( + "control payload exceeds the outer protocol limit".to_owned(), + )); + } + self.connection + .send(&frame, deadline) + .map_err(|error| failure(error, Stage::Data)) } - /// Reads and validates the next application frame, advancing the receive sequence. + /// Reads and decodes the next application frame, the payload of the next DATA record. fn read_app( &mut self, deadline: Option, ) -> Result<(u8, u64, i32, Vec), SessionError> { - let (header, frame) = self.read_outer(deadline)?; - if header.record_type == OUTER_RESET { - return Err(SessionError::Reset); - } - if header.record_type != OUTER_DATA - || header.instance_id != self.instance_id - || header.epoch != self.epoch - || header.sequence != self.receive_sequence - { - return Err(SessionError::Protocol( - "control endpoint returned an invalid data record".to_owned(), - )); - } - self.receive_sequence = self.receive_sequence.wrapping_add(1); - let frame = protocol::decode_app(&frame)?; + let payload = self + .connection + .receive(deadline) + .map_err(|error| failure(error, Stage::Data))?; + let frame = protocol::decode_app(&payload)?; Ok(( frame.kind, frame.request_id, @@ -405,57 +357,67 @@ impl ControlSession { } Ok(()) } +} - /// Buffers transport bytes until one complete outer record can be returned. - fn read_outer( - &mut self, - deadline: Option, - ) -> Result<(OuterHeader, Vec), SessionError> { - loop { - if self.received.len() >= OUTER_HEADER_LEN { - let header = protocol::decode_outer_header( - self.received[..OUTER_HEADER_LEN] - .try_into() - .expect("the slice has the header length"), - )?; - let end = OUTER_HEADER_LEN + header.length; - if self.received.len() >= end { - let payload = self.received[OUTER_HEADER_LEN..end].to_vec(); - self.received.drain(..end); - return Ok((header, payload)); - } - } - self.fill(deadline)?; - } - } - - /// Appends the next bytes from the transport to the receive buffer. - fn fill(&mut self, deadline: Option) -> Result<(), SessionError> { - let mut chunk = [0u8; 16 * 1024]; - loop { - match self.transport.read(&mut chunk, remaining(deadline)?) { - Ok(0) => return Err(SessionError::Closed), - Ok(count) => { - self.received.extend_from_slice(&chunk[..count]); - return Ok(()); - } - Err(error) if error.kind() == io::ErrorKind::Interrupted => {} - Err(error) => return Err(error.into()), - } - } - } +/// Stage of a control session at which a failure arrived, which names a protocol violation. +#[derive(Clone, Copy)] +enum Stage { + /// The host attaches. + Attach, + /// Application frames flow. + Data, } -/// Converts a deadline into the time left, failing once it has passed. -fn remaining(deadline: Option) -> Result, SessionError> { - match deadline { - None => Ok(None), - Some(deadline) => deadline - .checked_duration_since(Instant::now()) - .filter(|remaining| !remaining.is_zero()) - .map(Some) - .ok_or(SessionError::TimedOut), - } +/// Converts a failure of the host's NVXS connection at `stage` into the session's failure. +/// +/// 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 { + let violation = match error { + HostConnectionError::TimedOut => return SessionError::TimedOut, + HostConnectionError::Closed => return SessionError::Closed, + HostConnectionError::Rejected(_) => return SessionError::AuthenticationFailed, + HostConnectionError::Reset(_) => return SessionError::Reset, + HostConnectionError::Io(error) => return error.into(), + HostConnectionError::Ended => { + return SessionError::Protocol( + "the control session ended after an earlier failure".to_owned(), + ); + } + HostConnectionError::Protocol(violation) => violation, + }; + let message = match (violation, stage) { + (nvxs::SessionError::Truncated { .. }, _) => return SessionError::Closed, + ( + nvxs::SessionError::Record( + RecordError::Magic(_) | RecordError::Version(_) | RecordError::Flags(_), + ), + _, + ) => "control endpoint returned an invalid outer record", + (nvxs::SessionError::Record(RecordError::Length { length, .. }), _) + if length > OUTER_MAX_PAYLOAD => + { + "control endpoint returned an oversized outer record" + } + ( + nvxs::SessionError::Record(RecordError::Length { + record_type: RecordType::Wait | RecordType::Ready, + .. + }), + Stage::Attach, + ) => "control attach response carried a payload", + ( + nvxs::SessionError::Record(RecordError::Length { + record_type: RecordType::Data, + length: 0, + }), + Stage::Data, + ) => "control endpoint returned a truncated application frame", + (_, Stage::Attach) => "control endpoint returned an invalid attach response", + (_, Stage::Data) => "control endpoint returned an invalid data record", + }; + SessionError::Protocol(message.to_owned()) } /// Returns the error category that the payload of an `APP_ERROR` response names, truncated to 64 @@ -482,32 +444,66 @@ mod tests { use std::sync::{Arc, Mutex}; use super::*; + use crate::protocol::nvxs::{ErrorCode, Identity}; use crate::protocol::{ - WorkloadEnvironment, decode_app, decode_outer_header, encode_app, encode_outer, + OUTER_DATA, OUTER_ERROR, OUTER_HEADER_LEN, OUTER_HOST_ATTACH, OUTER_READY, OUTER_RESET, + OUTER_WAIT, WorkloadEnvironment, decode_app, decode_outer_header, encode_app, encode_outer, }; const INSTANCE: [u8; 16] = [7; 16]; - /// Transport that replays scripted guest records and records host writes. - struct Scripted { + /// What a test shares with a [`Scripted`] transport. + #[derive(Default)] + struct Wire { + /// Bytes for the host to read. input: VecDeque, - output: Arc>>, + /// Whether the stream ends once the host read `input`. + closed: bool, + /// Bytes that the host wrote. + output: Vec, + } + + /// Transport of a scripted guest, whose records the test supplies at any time. A read + /// without input times out, or ends the stream once the test closed it. + #[derive(Clone, Default)] + struct Scripted(Arc>); + + impl Scripted { + /// Supplies records for the host to read. + fn supply(&self, records: &[Vec]) { + self.0.lock().unwrap().input.extend(records.concat()); + } + + /// Ends the stream after the records that the test supplied. + fn close(&self) { + self.0.lock().unwrap().closed = true; + } + + /// Returns the bytes that the host wrote. + fn output(&self) -> Vec { + self.0.lock().unwrap().output.clone() + } } impl Transport for Scripted { fn read(&mut self, buffer: &mut [u8], _timeout: Option) -> io::Result { - if self.input.is_empty() { - return Err(io::ErrorKind::TimedOut.into()); + let mut wire = self.0.lock().unwrap(); + if wire.input.is_empty() { + return if wire.closed { + Ok(0) + } else { + Err(io::ErrorKind::TimedOut.into()) + }; } - let count = buffer.len().min(self.input.len()); - for byte in &mut buffer[..count] { - *byte = self.input.pop_front().unwrap(); + let count = buffer.len().min(wire.input.len()); + for (byte, input) in buffer.iter_mut().zip(wire.input.drain(..count)) { + *byte = input; } Ok(count) } fn write_all(&mut self, data: &[u8], _timeout: Option) -> io::Result<()> { - self.output.lock().unwrap().extend_from_slice(data); + self.0.lock().unwrap().output.extend_from_slice(data); Ok(()) } } @@ -517,19 +513,33 @@ mod tests { encode_outer(record_type, &instance, epoch, sequence, payload).unwrap() } - fn scripted(records: &[Vec]) -> (Box, Arc>>) { - let output = Arc::new(Mutex::new(Vec::new())); - let transport = Scripted { - input: records.concat().into(), - output: Arc::clone(&output), - }; - (Box::new(transport), output) + /// Returns `record` with `bytes` at `offset`. + fn patched(mut record: Vec, offset: usize, bytes: &[u8]) -> Vec { + record[offset..offset + bytes.len()].copy_from_slice(bytes); + record } fn deadline() -> Instant { Instant::now() + Duration::from_secs(5) } + /// Attaches through a scripted guest that supplied `records`, and returns the attachment + /// with the guest's transport. + fn scripted(records: &[Vec]) -> (Result, Scripted) { + let wire = Scripted::default(); + wire.supply(records); + let session = + ControlSession::attach(Box::new(wire.clone()), &[1; CAPABILITY_LEN], deadline()); + (session, wire) + } + + /// Attaches through a scripted guest whose broker attaches the host at its sequence 0, in + /// epoch 1. + fn ready() -> (ControlSession, Scripted) { + let (session, wire) = scripted(&[outer(OUTER_READY, 1, 0, &[])]); + (session.unwrap(), wire) + } + /// Kind and payload of each application request that a guest received. type Requests = Arc)>>>; @@ -608,33 +618,115 @@ mod tests { records } + /// Checks that `error` is a protocol violation with `message`. + #[track_caller] + fn assert_violation(error: &SessionError, message: &str, case: &str) { + let matched = matches!(error, SessionError::Protocol(text) if text == message); + assert!(matched, "{case}: {error:?}"); + } + #[test] - fn attach_waits_for_the_host_slot_then_authenticates() { - let (transport, output) = - scripted(&[outer(OUTER_WAIT, 0, 0, &[]), outer(OUTER_READY, 4, 9, &[])]); - let capability = [0x5a; CAPABILITY_LEN]; - let session = ControlSession::attach(transport, &capability, deadline()).unwrap(); - assert_eq!(session.epoch, 4); - assert_eq!(session.receive_sequence, 10); - let records = host_records(&output.lock().unwrap()); - assert_eq!(records, [(OUTER_HOST_ATTACH, 0, capability.to_vec())]); + fn attach_waits_for_the_guest_then_authenticates() { + // The broker asks the host to wait while the guest has not acknowledged the session. + let (session, wire) = + scripted(&[outer(OUTER_WAIT, 0, 0, &[]), outer(OUTER_READY, 4, 1, &[])]); + let session = session.unwrap(); + let identity = Identity { + instance_id: INSTANCE, + epoch: 4, + }; + assert_eq!(session.connection.identity(), identity); + let records = host_records(&wire.output()); + assert_eq!(records, [(OUTER_HOST_ATTACH, 0, vec![1; CAPABILITY_LEN])]); } #[test] fn attach_reports_authentication_failure() { - let (transport, _) = scripted(&[outer(OUTER_ERROR, 0, 0, &[])]); - let error = ControlSession::attach(transport, &[1; CAPABILITY_LEN], deadline()) - .err() - .unwrap(); - assert!(matches!(error, SessionError::AuthenticationFailed)); + let code = ErrorCode::Authentication.code().to_le_bytes(); + let (session, _) = scripted(&[outer(OUTER_ERROR, 0, 0, &code)]); + assert!(matches!(session, Err(SessionError::AuthenticationFailed))); + } + + #[test] + fn attach_responses_that_the_protocol_does_not_allow_fail_closed() { + let ready = outer(OUTER_READY, 1, 0, &[]); + let oversized = u32::try_from(OUTER_MAX_PAYLOAD + 1).unwrap().to_le_bytes(); + let payload = "control attach response carried a payload"; + let invalid = "control endpoint returned an invalid attach response"; + let outer_record = "control endpoint returned an invalid outer record"; + let cases = [ + ( + "a READY with a payload", + outer(OUTER_READY, 1, 0, b"x"), + payload, + ), + ( + "a WAIT with a payload", + outer(OUTER_WAIT, 0, 0, b"x"), + payload, + ), + ( + "a READY without an identity", + outer(OUTER_READY, 0, 0, &[]), + invalid, + ), + ( + "a READY out of sequence", + outer(OUTER_READY, 1, 3, &[]), + invalid, + ), + ("DATA before READY", outer(OUTER_DATA, 1, 0, b"x"), invalid), + ( + "an ERROR without a code", + outer(OUTER_ERROR, 0, 0, &[]), + invalid, + ), + ( + "an ERROR with an unknown code", + outer(OUTER_ERROR, 0, 0, &2u32.to_le_bytes()), + invalid, + ), + ("an unknown record type", outer(0x7f, 1, 0, &[]), invalid), + ( + "a wrong magic", + patched(ready.clone(), 0, b"NVXT"), + outer_record, + ), + ( + "another version", + patched(ready.clone(), 4, &2u16.to_le_bytes()), + outer_record, + ), + ( + "nonzero flags", + patched(ready.clone(), 7, &[1]), + outer_record, + ), + ( + "an oversized record", + patched(ready.clone(), 40, &oversized), + "control endpoint returned an oversized outer record", + ), + ]; + for (case, record, message) in cases { + let error = scripted(&[record]).0.err().unwrap(); + assert_violation(&error, message, case); + } + + // A stream that ends, at a record boundary or inside a record, has closed. + for records in [Vec::new(), vec![ready[..20].to_vec()]] { + let wire = Scripted::default(); + wire.supply(&records); + wire.close(); + let attached = ControlSession::attach(Box::new(wire), &[1; CAPABILITY_LEN], deadline()); + assert!(matches!(attached, Err(SessionError::Closed))); + } } #[test] fn exec_streams_events_in_order() { // The request ID is random, so the scripted guest answers the frames the host sent. - let (transport, output) = scripted(&[outer(OUTER_READY, 1, 0, &[])]); - let mut session = - ControlSession::attach(transport, &[1; CAPABILITY_LEN], deadline()).unwrap(); + let (mut session, wire) = ready(); let workload = Workload { argv: vec!["/bin/sh".to_owned(), "-c".to_owned(), "echo hi".to_owned()], timeout_ms: 1000, @@ -642,13 +734,13 @@ mod tests { environment: WorkloadEnvironment::Default, }; let request_id = session.start_exec(&workload, deadline()).unwrap(); - let records = host_records(&output.lock().unwrap()); + let records = host_records(&wire.output()); let (record_type, sequence, frame) = &records[1]; assert_eq!((*record_type, *sequence), (OUTER_DATA, 0)); let frame = decode_app(frame).unwrap(); assert_eq!((frame.kind, frame.request_id), (APP_EXEC, request_id)); - let replies = [ + wire.supply(&[ outer( OUTER_DATA, 1, @@ -667,8 +759,7 @@ mod tests { 3, &encode_app(APP_EXIT, request_id, 3, b"exit"), ), - ]; - session.transport = scripted(&replies).0; + ]); let events: Vec = (0..3) .map(|_| { session @@ -691,9 +782,7 @@ mod tests { #[test] fn exec_sends_the_working_directory_and_reports_its_refusal() { - let (transport, output) = scripted(&[outer(OUTER_READY, 1, 0, &[])]); - let mut session = - ControlSession::attach(transport, &[1; CAPABILITY_LEN], deadline()).unwrap(); + let (mut session, wire) = ready(); let workload = Workload { argv: vec!["/bin/pwd".to_owned()], timeout_ms: 0, @@ -701,7 +790,7 @@ mod tests { environment: WorkloadEnvironment::Default, }; let request_id = session.start_exec(&workload, deadline()).unwrap(); - let records = host_records(&output.lock().unwrap()); + let records = host_records(&wire.output()); let frame = decode_app(&records[1].2).unwrap(); assert_eq!((frame.kind, frame.request_id), (APP_EXEC, request_id)); assert_eq!( @@ -710,7 +799,7 @@ mod tests { ); let diagnostic = b"nvx-managed-agent: cannot enter working directory /work: No such file\n"; - session.transport = scripted(&[ + wire.supply(&[ outer( OUTER_DATA, 1, @@ -723,8 +812,7 @@ mod tests { 2, &encode_app(APP_ERROR, request_id, 2, protocol::CWD_FAILED.as_bytes()), ), - ]) - .0; + ]); let events: Vec = (0..2) .map(|_| { session @@ -745,38 +833,111 @@ mod tests { } #[test] - fn out_of_order_and_reset_records_fail_closed() { - let (transport, _) = scripted(&[outer(OUTER_READY, 1, 0, &[])]); - let mut session = - ControlSession::attach(transport, &[1; CAPABILITY_LEN], deadline()).unwrap(); - session.transport = - scripted(&[outer(OUTER_DATA, 1, 5, &encode_app(APP_STDOUT, 1, 0, b"x"))]).0; - assert!(matches!( - session.next_exec_event(1, Some(deadline())), - Err(SessionError::Protocol(_)) - )); - session.transport = scripted(&[outer(OUTER_RESET, 1, 1, &[])]).0; + fn records_that_the_protocol_does_not_allow_fail_closed() { + let frame = encode_app(APP_STDOUT, 1, 0, b"x"); + let data = outer(OUTER_DATA, 1, 1, &frame); + let oversized = u32::try_from(OUTER_MAX_PAYLOAD + 1).unwrap().to_le_bytes(); + let invalid = "control endpoint returned an invalid data record"; + let truncated = "control endpoint returned a truncated application frame"; + let cases = [ + ( + "DATA out of sequence", + outer(OUTER_DATA, 1, 5, &frame), + invalid, + ), + ( + "a repeated sequence number", + outer(OUTER_DATA, 1, 0, &frame), + invalid, + ), + ( + "DATA of another epoch", + outer(OUTER_DATA, 2, 1, &frame), + invalid, + ), + ( + "DATA of another instance", + encode_outer(OUTER_DATA, &[9; 16], 1, 1, &frame).unwrap(), + invalid, + ), + ("a WAIT", outer(OUTER_WAIT, 1, 1, &[]), invalid), + ( + "a RESET of the session's epoch", + outer(OUTER_RESET, 1, 1, &[]), + invalid, + ), + ("an unknown record type", outer(0x7f, 1, 1, &[]), invalid), + ( + "an empty DATA record", + outer(OUTER_DATA, 1, 1, &[]), + truncated, + ), + ( + "a truncated frame", + outer(OUTER_DATA, 1, 1, &frame[..APP_HEADER_LEN - 1]), + truncated, + ), + ( + "nonzero flags", + patched(data.clone(), 7, &[1]), + "control endpoint returned an invalid outer record", + ), + ( + "an oversized record", + patched(data.clone(), 40, &oversized), + "control endpoint returned an oversized outer record", + ), + ]; + for (case, record, message) in cases { + let (mut session, wire) = ready(); + wire.supply(&[record]); + let error = session.next_exec_event(1, Some(deadline())).unwrap_err(); + assert_violation(&error, message, case); + } + + // A violation ends the session. + let (mut session, wire) = ready(); + wire.supply(&[outer(OUTER_DATA, 1, 5, &frame)]); + assert!(session.next_exec_event(1, Some(deadline())).is_err()); + wire.supply(std::slice::from_ref(&data)); + let error = session.next_exec_event(1, Some(deadline())).unwrap_err(); + let ended = "the control session ended after an earlier failure"; + assert_violation(&error, ended, "a call after a violation"); + let error = session.cancel_exec(1, deadline()).unwrap_err(); + assert_violation(&error, ended, "a request after a violation"); + + // A reset to a later epoch ends the session too. + let (mut session, wire) = ready(); + wire.supply(&[outer(OUTER_RESET, 2, 1, &[])]); assert!(matches!( session.next_exec_event(1, Some(deadline())), Err(SessionError::Reset) )); + + // A stream that ends, at a record boundary or inside a record, has closed. + for records in [Vec::new(), vec![data[..OUTER_HEADER_LEN + 1].to_vec()]] { + let (mut session, wire) = ready(); + wire.supply(&records); + wire.close(); + assert!(matches!( + session.next_exec_event(1, Some(deadline())), + Err(SessionError::Closed) + )); + } } #[test] fn timed_out_polls_keep_partial_records() { - let (transport, _) = scripted(&[outer(OUTER_READY, 1, 0, &[])]); - let mut session = - ControlSession::attach(transport, &[1; CAPABILITY_LEN], deadline()).unwrap(); + let (mut session, wire) = ready(); let record = outer(OUTER_DATA, 1, 1, &encode_app(APP_EXIT, 9, 0, b"cancelled")); let (first, second) = record.split_at(30); - let (transport, output) = scripted(&[first.to_vec()]); - session.transport = transport; + wire.supply(&[first.to_vec()]); assert!(matches!( session.next_exec_event(9, Some(deadline())), Err(SessionError::TimedOut) )); session.cancel_exec(9, deadline()).unwrap(); - session.transport = scripted(&[second.to_vec()]).0; + wire.supply(&[second.to_vec()]); assert_eq!( session.next_exec_event(9, Some(deadline())).unwrap(), ExecEvent::Exit { @@ -784,11 +945,41 @@ mod tests { status: 0 } ); - let records = host_records(&output.lock().unwrap()); + let records = host_records(&wire.output()); let frame = decode_app(&records.last().unwrap().2).unwrap(); assert_eq!((frame.kind, frame.request_id), (APP_CANCEL, 9)); } + #[test] + fn frames_travel_in_one_record_each() { + let (mut session, wire) = ready(); + let longest = vec![0x5a; OUTER_MAX_PAYLOAD - APP_HEADER_LEN]; + session + .send_request(APP_EXEC, 1, &longest, deadline()) + .unwrap(); + let error = session + .send_request( + APP_EXEC, + 2, + &[longest.as_slice(), &[0]].concat(), + deadline(), + ) + .unwrap_err(); + let message = "control payload exceeds the outer protocol limit"; + assert_violation(&error, message, "a frame longer than a record"); + let records = host_records(&wire.output()); + assert_eq!(records.len(), 2); + let (record_type, sequence, frame) = &records[1]; + assert_eq!((*record_type, *sequence), (OUTER_DATA, 0)); + let frame = decode_app(frame).unwrap(); + assert_eq!((frame.kind, frame.request_id), (APP_EXEC, 1)); + assert_eq!(frame.payload, longest); + // The refused frame changed nothing. + session.cancel_exec(1, deadline()).unwrap(); + let records = host_records(&wire.output()); + assert_eq!((records[2].0, records[2].1), (OUTER_DATA, 1)); + } + /// The response kind that acknowledges a ping or a stop request. fn acknowledgement(kind: u8) -> u8 { if kind == APP_PING { @@ -955,8 +1146,8 @@ mod tests { #[test] fn expired_deadlines_time_out() { - let (transport, _) = scripted(&[]); - let error = ControlSession::attach(transport, &[1; CAPABILITY_LEN], Instant::now()) + let wire = Scripted::default(); + let error = ControlSession::attach(Box::new(wire), &[1; CAPABILITY_LEN], Instant::now()) .err() .unwrap(); assert!(matches!(error, SessionError::TimedOut)); diff --git a/nvx-rs/src/testing/fake_openvmm.rs b/nvx-rs/src/testing/fake_openvmm.rs index c8164acdb..1af037916 100644 --- a/nvx-rs/src/testing/fake_openvmm.rs +++ b/nvx-rs/src/testing/fake_openvmm.rs @@ -72,6 +72,7 @@ use std::process::ExitCode; use std::thread; use std::time::{Duration, Instant}; +use crate::protocol::nvxs::ErrorCode; use crate::protocol::{ APP_CANCEL, APP_ERROR, APP_EXEC, APP_EXIT, APP_FEATURES, APP_MAPS, APP_PING, APP_READY, APP_STDERR, APP_STDOUT, APP_STOP, APP_STOPPED, BUSY, CAPABILITY_LEN, CWD_FAILED, ExitCategory, @@ -967,9 +968,10 @@ impl Pending for fs::File { } } -/// Encodes an outer record without a payload, which the broker sends to answer an attach. -fn control_record(record_type: u8, instance: &[u8; 16], epoch: u64) -> Vec { - encode_outer(record_type, instance, epoch, 0, &[]).expect("an empty payload fits a record") +/// Encodes the broker's answer to an attachment, its first record to the host: READY, which +/// carries the session's identity, or ERROR, whose payload is the refusal's code. +fn control_record(record_type: u8, instance: &[u8; 16], epoch: u64, payload: &[u8]) -> Vec { + encode_outer(record_type, instance, epoch, 0, payload).expect("an answer fits a record") } /// Decoded outer header paired with its owned payload. @@ -1327,7 +1329,8 @@ fn serve_client( return Flow::Continue; }; if header.record_type != OUTER_HOST_ATTACH || payload != capability { - let _ = stream.write_all(&control_record(OUTER_ERROR, &[0; 16], 0)); + let code = ErrorCode::Authentication.code().to_le_bytes(); + let _ = stream.write_all(&control_record(OUTER_ERROR, &[0; 16], 0, &code)); return Flow::Continue; } let mut instance = [0u8; 16]; @@ -1337,7 +1340,7 @@ fn serve_client( } } if stream - .write_all(&control_record(OUTER_READY, &instance, epoch)) + .write_all(&control_record(OUTER_READY, &instance, epoch, &[])) .is_err() { return Flow::Continue;