diff --git a/.github/actions/check-nvx-rs/action.yml b/.github/actions/check-nvx-rs/action.yml index 0cbae299f..cad79c383 100644 --- a/.github/actions/check-nvx-rs/action.yml +++ b/.github/actions/check-nvx-rs/action.yml @@ -143,8 +143,8 @@ runs: python3 ../scripts/nvx.py test-guest-binaries --binaries "${directory}" # The runtime adapter builds on nvx-rs as an independent adapter does: with a lockfile of its - # own, default features off, and the public API only. Its tests check consumer-contract - # scenario C0 in each configuration, which also builds with nvx-rs's minimum Rust version. + # own, default features off, and the public API only. Its tests check that each configuration + # builds that way, and each configuration also builds with nvx-rs's minimum Rust version. - name: Check the runtime adapter's protocol-only configuration shell: bash working-directory: tests/runtime-adapter @@ -168,7 +168,7 @@ runs: cargo "+${{ steps.toolchains.outputs.toolchain }}" test --locked --features host cargo "+${{ steps.toolchains.outputs.msrv }}" check --locked --all-targets --features host - # The guest half runs in Linux guests, as a static executable that a test checks. + # The guest agent runs in Linux guests, as a static executable that a test checks. - name: Check the runtime adapter's guest-only configuration if: runner.os == 'Linux' shell: bash diff --git a/.github/workflows/run-nvx-microvm-tests.yml b/.github/workflows/run-nvx-microvm-tests.yml index 3deb79666..7bc311d93 100644 --- a/.github/workflows/run-nvx-microvm-tests.yml +++ b/.github/workflows/run-nvx-microvm-tests.yml @@ -1180,10 +1180,10 @@ jobs: --backend "${{ inputs.backend }}" --output-dir "build/test-results/aci-edge-sandboxes-${{ inputs.backend }}" - # The hook of the runtime adapter's scenarios on this backend: the ignored - # tests of its host-only configuration, which later steps add and which - # read the backend through `nvx::Hypervisor::from_env_or_default`. There - # are none yet, so these steps only build the adapter. + # The runtime adapter's hypervisor scenarios on this backend: any ignored + # tests of its host-only configuration, which read the backend through + # `nvx::Hypervisor::from_env_or_default`. The step builds the adapter + # whether or not it declares any. - name: Run runtime adapter scenarios on Linux if: runner.os != 'Windows' && !inputs.debug-kernel shell: bash diff --git a/doc/ci.md b/doc/ci.md index d887d3447..3d641e7f1 100644 --- a/doc/ci.md +++ b/doc/ci.md @@ -657,10 +657,14 @@ lockfile of its own, default features off, and the public API only. For each of its configurations, protocol-only, host-only, and, on Linux, guest-only for `x86_64-unknown-linux-musl`, the action runs Clippy with warnings denied, the build, the tests, and a check with the minimum Rust version of `nvx`, after -rustfmt. The tests check consumer-contract scenario C0: each configuration -enables only its side of `nvx`, the guest-only build pulls in neither Tokio -nor a dependency that only the host side declares, and the guest binary is a -static executable. +rustfmt. The tests check that each configuration builds as an independent +adapter's does: it enables only its side of `nvx`, the guest-only build pulls +in neither Tokio nor a dependency that only the host side declares, and the +guest binary is a static executable. In the guest-only configuration, they +also check that the adapter's guest agent serves the control session between +the host and the guest with the adapter's own framing, against a test broker +that sends it messages that span records, resets, disconnects, and malformed +records. The development release job depends on `aci-edge-sandboxes` and `nvx-rs` and requires their combined Linux/Windows results to be successful; failed, @@ -680,12 +684,12 @@ it keeps the OpenVMM log under `build/test-results/aci-edge-sandboxes-` uploaded with the other microVM logs. Changes under `aci_edge_sandboxes/` or `nvx-rs/` therefore trigger the NVX microVM test matrix. -The same jobs, except the debug-kernel ones, run the runtime adapter's -scenarios on their backend: the ignored tests of its host-only configuration, -with `NVX_HYPERVISOR` naming the backend. Later steps of -[microsoft/nvx#535](https://github.com/microsoft/nvx/issues/535) add them; until -then, the step only builds the adapter. Changes under `tests/runtime-adapter/` -therefore trigger the NVX microVM test matrix too. +The same jobs, except the debug-kernel ones, build the runtime adapter's +host-only configuration on their backend and run any ignored tests that it +declares, its hypervisor scenarios, with `NVX_HYPERVISOR` naming the backend. +The adapter's other tests, including those of its guest agent, run in the +`nvx-rs` job. Changes under `tests/runtime-adapter/` therefore trigger the NVX +microVM test matrix too. ## Adversarial campaigns diff --git a/doc/contribute.md b/doc/contribute.md index 08b71fefe..0d62c2f49 100644 --- a/doc/contribute.md +++ b/doc/contribute.md @@ -116,8 +116,8 @@ runs `tests/guest_tools.rs` against the exact guest-profile binaries. Tests of t mechanics, such as the golden fixtures of the control protocol, belong in `nvx`, and tests of the ACI Edge Sandboxes contract in `aci_edge_sandboxes`. Tests of what an independent adapter observes through -the public API of `nvx` alone, such as the consumer-contract scenarios of -[microsoft/nvx#535](https://github.com/microsoft/nvx/issues/535), belong in the +the public API of `nvx` alone, such as its own framing over the control +session between the host and the guest, belong in the [runtime adapter](../tests/runtime-adapter/README.md), whose ignored tests the microVM jobs run on each backend. The `scripts/nvx.py test-aci-edge-sandboxes --backend BACKEND` command runs all ignored `openvmm_e2e` tests on a real hypervisor, diff --git a/doc/project-structure.md b/doc/project-structure.md index ac5a64d47..70d178b55 100644 --- a/doc/project-structure.md +++ b/doc/project-structure.md @@ -161,8 +161,8 @@ nvx/ | `-- test_*.py Python tooling unit tests |-- tests/ Tests that build on the repository's crates from outside them | `-- runtime-adapter/ Rust crate `runtime-adapter`, an independent adapter of nvx -| |-- src/ Its library, and its host and guest skeleton binaries -| `-- tests/ Consumer-contract scenario C0, the independent build +| |-- src/ Its framing, its host program, and its guest agent +| `-- tests/ Its independent build, and its guest agent against a test broker |-- SECURITY.md Microsoft security vulnerability reporting policy |-- .dockerignore Docker build-context exclusions |-- .gitattributes Git path attributes @@ -335,8 +335,9 @@ Tests that build on the repository's crates from outside them. crate with its own `Cargo.lock` that depends on `nvx` with default features off and uses its public API only. Its protocol-only, host-only, and guest-only configurations build in CI, and its tests check that each one builds as an -independent adapter's does. See its -[README](../tests/runtime-adapter/README.md). +independent adapter's does. Its guest agent serves the control session between +the host and the guest with the adapter's own framing, which its tests check +against a test broker. See its [README](../tests/runtime-adapter/README.md). ## Root files diff --git a/tests/runtime-adapter/Cargo.toml b/tests/runtime-adapter/Cargo.toml index 3506cfc58..076a8c1d2 100644 --- a/tests/runtime-adapter/Cargo.toml +++ b/tests/runtime-adapter/Cargo.toml @@ -14,9 +14,9 @@ publish = false # No default features: each configuration selects one side of `nvx`, as an independent adapter # does. Without features, the adapter builds against the protocol-only `nvx`. default = [] -# The host half: the host API of `nvx` and the `runtime-adapter-host` binary. +# The host program: the host API of `nvx` and the `runtime-adapter-host` binary. host = ["nvx/host"] -# The guest half: the guest API of `nvx` and the `runtime-adapter-guest` binary, which builds as a +# The guest agent: the guest API of `nvx` and the `runtime-adapter-guest` binary, which builds as a # static `x86_64-unknown-linux-musl` executable. guest = ["nvx/guest"] diff --git a/tests/runtime-adapter/README.md b/tests/runtime-adapter/README.md index 578658886..afab47e02 100644 --- a/tests/runtime-adapter/README.md +++ b/tests/runtime-adapter/README.md @@ -1,38 +1,78 @@ # Runtime adapter -The runtime adapter is an independent adapter of the NVX runtime. It lives in -this repository so that NVX's own CI builds and runs it, and it depends on -[`nvx`](../../nvx-rs/README.md) the way an adapter outside the repository -does: +The runtime adapter is an independent adapter of the NVX runtime, with a host +program and a guest agent. It lives in this repository so that NVX's own CI +builds and runs it, and it depends on [`nvx`](../../nvx-rs/README.md) the way +an adapter outside the repository does: - as a path dependency with default features off; - with a `Cargo.lock` of its own, outside any other workspace; - through the public API alone, with unsafe code forbidden; - with the minimum supported Rust version of `nvx`. -[microsoft/nvx#535](https://github.com/microsoft/nvx/issues/535) sets the -contract that it proves: an adapter brings its own application framing over -the NVXS session, its own init, and its own status collector. Today its -binaries are skeletons that report the contract that they were built against. -Later steps fill its guest half -([microsoft/nvx#619](https://github.com/microsoft/nvx/issues/619)) and its host -half ([microsoft/nvx#620](https://github.com/microsoft/nvx/issues/620)), and add -the scenarios that run them on a hypervisor. +The adapter brings its own application framing to the control session between +the host and the guest, in place of NVX's `NVXC` frames. Its guest agent serves +requests in that framing. Its host program reports the contract that it was +built against. ## Configurations Each configuration enables one side of `nvx`: -| Configuration | Cargo options | `nvx` features | Binary | -| --- | --- | --- | --- | -| Protocol-only | none | none | none | -| Host-only | `--features host` | `host` | `runtime-adapter-host` | -| Guest-only | `--features guest --target x86_64-unknown-linux-musl` | `guest` | `runtime-adapter-guest`, a static executable | +| Configuration | Cargo options | `nvx` features | Modules | Binary | +| --- | --- | --- | --- | --- | +| Protocol-only | none | none | `frame` | none | +| Host-only | `--features host` | `host` | `frame` | `runtime-adapter-host` | +| Guest-only | `--features guest --target x86_64-unknown-linux-musl` | `guest` | `frame`, `guest` | `runtime-adapter-guest`, a static executable | + +## Framing + +The `frame` module is the adapter's application framing, which its host program +and guest agent exchange over the control session in place of NVX's `NVXC` +frames. A message is a 5-byte header, then a body. The header holds the body's +length, a little-endian `u32`, then the message's kind: + +| Kind | Message | Body | +| --- | --- | --- | +| 1 | `Ping`: asks the guest to answer with `Pong` | none | +| 2 | `Pong`: the guest's answer to `Ping` | none | +| 3 | `Echo`: bytes that the guest sends back in an `Echo` of its own | at most 256 KiB, what four of the session's largest records carry | +| 4 | `Stop`: asks the guest to stop serving | none | +| 5 | `Stopped`: the guest's confirmation of `Stop`, its last message | none | + +The session keeps no message boundaries, so a message may span several of its +records, and one record may end a message and start the next. `Decoder` +reassembles the messages. It checks each header before it takes any byte of +the body, and holds at most one partial message. A malformed message ends the +stream, because the framing cannot find the next message after it. + +## Guest agent + +`guest::serve` serves the host program on a stream to the guest's end of the +control session, through `nvx::guest::nvxs`. It attaches, then: + +- acknowledges each reset of the session with the smallest receive window that + the session allows, once it dropped the earlier epoch's partial message, + because a new epoch starts with a new message; +- credits each payload back once it holds the payload, so that a message + longer than the window still arrives whole; +- answers each `Ping` with `Pong`, sends each `Echo` back, and answers `Stop` + with `Stopped`, then returns. + +A failure of the session, a malformed message, and a message that only the +guest sends end serving with an error. + +`runtime-adapter-guest serve` runs the guest agent on its standard input and +output, which connect to the guest's end of the session, such as a control +console in raw mode. It exits with 0 once the host stopped it or the broker +closed the stream, and with 1, after a diagnostic on standard error, once +serving failed. Without arguments, it reports the contract that it was built +against. ## Tests -`tests/independent_build.rs` checks consumer-contract scenario C0 in every -configuration: +`tests/independent_build.rs` checks that every configuration builds as an +independent adapter's does: - The adapter is a workspace of its own with its own lockfile, and declares the minimum supported Rust version of `nvx`. @@ -46,6 +86,22 @@ configuration: guest binary is a static executable: it has no program interpreter, needs no shared library, and keeps its stack non-executable. +The unit tests of `frame` run in every configuration. They cover each kind's +encoding, messages in pieces of any size and split at every byte, and the +headers that the framing refuses before their body arrives. + +`tests/guest_session.rs` checks, in the guest-only configuration, that the +guest agent serves the control session with the adapter's own framing. A test +broker plays the broker's end of the guest's session over a Unix socket pair +with the public record API of `nvx`, for `guest::serve` and for +`runtime-adapter-guest serve`. It covers messages that span records both ways +and fill the receive window, a record whose bytes arrive one at a time, resets +in the middle of a message, with the next epoch of the instance and with a new +instance, streams that end at and inside a record, and the records and +messages that the guest agent refuses: sequence, epoch, and instance errors, +stale and identity-less resets, record lengths, bootstrap identity, ERROR +payloads, and malformed headers. + Run the checks of each configuration in this directory. The guest-only configuration needs Linux and the musl target, which `rustup target add x86_64-unknown-linux-musl` installs: @@ -69,7 +125,7 @@ supported Rust version, on Linux and Windows, where it skips the guest-only configuration. A change to the adapter is a test input of CI's [change classification](../../doc/ci.md#change-classification), so it runs the NVX microVM tests. Their -[workflow](../../.github/workflows/run-nvx-microvm-tests.yml) runs the -adapter's hypervisor scenarios on each backend: its ignored tests in the -host-only configuration, with `NVX_HYPERVISOR` naming the backend, which -`nvx::Hypervisor::from_env_or_default` reads. There are none yet. +[workflow](../../.github/workflows/run-nvx-microvm-tests.yml) builds the +host-only configuration on each backend and runs any ignored tests that it +declares, the adapter's hypervisor scenarios, with `NVX_HYPERVISOR` naming the +backend, which `nvx::Hypervisor::from_env_or_default` reads. diff --git a/tests/runtime-adapter/src/bin/runtime-adapter-guest.rs b/tests/runtime-adapter/src/bin/runtime-adapter-guest.rs index e157e5bd5..7a14a138f 100644 --- a/tests/runtime-adapter/src/bin/runtime-adapter-guest.rs +++ b/tests/runtime-adapter/src/bin/runtime-adapter-guest.rs @@ -1,20 +1,76 @@ -//! The guest half of the runtime adapter, a static `x86_64-unknown-linux-musl` executable. This -//! skeleton reports the contract that it was built against; later steps make it serve the guest -//! role of the NVXS session with the adapter's own framing. +//! The runtime adapter's guest agent, a static `x86_64-unknown-linux-musl` executable. +//! +//! - `runtime-adapter-guest` reports the contract that it was built against. +//! - `runtime-adapter-guest serve` serves the host program with the adapter's own framing on its +//! standard input and output, which connect to the guest's end of the control session between +//! the host and the guest, such as a control console in raw mode. It exits with 0 once the host +//! stopped it or the broker closed the stream, and with 1, after a diagnostic on standard error, +//! once serving failed. +//! +//! Anything else is a usage error, with status 2. use std::process::ExitCode; #[cfg(target_os = "linux")] fn main() -> ExitCode { - if std::env::args_os().len() > 1 { - eprintln!("usage: runtime-adapter-guest"); - return ExitCode::from(2); + let arguments = std::env::args_os().skip(1).collect::>(); + match arguments.as_slice() { + [] => { + println!("runtime-adapter-guest: {}", runtime_adapter::contract()); + ExitCode::SUCCESS + } + [command] if command == "serve" => serve(), + _ => { + eprintln!("usage: runtime-adapter-guest [serve]"); + ExitCode::from(2) + } } - println!("runtime-adapter-guest: {}", runtime_adapter::contract()); - ExitCode::SUCCESS } -// The guest half runs only in Linux guests. Elsewhere this stub fails, so that every target still +/// Serves the session on standard input and output. +#[cfg(target_os = "linux")] +fn serve() -> ExitCode { + let stdio = Stdio { + input: std::io::stdin().lock(), + output: std::io::stdout().lock(), + }; + match runtime_adapter::guest::serve(stdio) { + Ok(_) => ExitCode::SUCCESS, + Err(error) => { + eprintln!("runtime-adapter-guest: {error}"); + ExitCode::FAILURE + } + } +} + +/// Standard input and output as one stream, which carries nothing but the session's records. +#[cfg(target_os = "linux")] +struct Stdio { + /// Standard input, from the broker. + input: std::io::StdinLock<'static>, + /// Standard output, to the broker. + output: std::io::StdoutLock<'static>, +} + +#[cfg(target_os = "linux")] +impl std::io::Read for Stdio { + fn read(&mut self, buffer: &mut [u8]) -> std::io::Result { + self.input.read(buffer) + } +} + +#[cfg(target_os = "linux")] +impl std::io::Write for Stdio { + fn write(&mut self, bytes: &[u8]) -> std::io::Result { + self.output.write(bytes) + } + + fn flush(&mut self) -> std::io::Result<()> { + self.output.flush() + } +} + +// The guest agent runs only in Linux guests. Elsewhere this stub fails, so that every target still // builds with every feature. #[cfg(not(target_os = "linux"))] fn main() -> ExitCode { diff --git a/tests/runtime-adapter/src/bin/runtime-adapter-host.rs b/tests/runtime-adapter/src/bin/runtime-adapter-host.rs index ffaa3c85c..34471149b 100644 --- a/tests/runtime-adapter/src/bin/runtime-adapter-host.rs +++ b/tests/runtime-adapter/src/bin/runtime-adapter-host.rs @@ -1,6 +1,5 @@ -//! The host half of the runtime adapter. This skeleton reports the contract that it was built -//! against and whether `nvx` supports this host; later steps make it start, reattach to, and -//! stop sandboxes through the host API. +//! The runtime adapter's host program. It reports the contract that it was built against and +//! whether `nvx` supports this host. use std::process::ExitCode; diff --git a/tests/runtime-adapter/src/frame.rs b/tests/runtime-adapter/src/frame.rs new file mode 100644 index 000000000..3a4656d4a --- /dev/null +++ b/tests/runtime-adapter/src/frame.rs @@ -0,0 +1,409 @@ +//! The adapter's own application framing, which its host program and guest agent exchange over the +//! control session between the host and the guest, in place of NVX's `NVXC` frames. +//! +//! A message is a header of [`HEADER_LEN`] bytes, then a body. The header holds the body's length, +//! a little-endian `u32`, then the message's kind. The session carries bytes and keeps no message +//! boundaries, so a message may span several of its records, and one record may end a message and +//! start the next. [`Decoder`] reassembles the messages: it checks each header before it takes any +//! byte of the body, and holds at most one partial message, whose body has at most [`MAX_BODY`] +//! bytes. +//! +//! The framing cannot find the next message after a malformed one, so a [`FrameError`] ends the +//! stream. + +use std::fmt; + +use nvx::protocol::OUTER_MAX_PAYLOAD; + +/// Bytes of a message's header: the body's length, then the message's kind. +pub const HEADER_LEN: usize = 5; + +/// Most bytes of a message's body: as many as four of the session's largest records carry, so that +/// the longest message, with its header, spans five records. +pub const MAX_BODY: usize = 4 * OUTER_MAX_PAYLOAD; + +/// A message of the adapter's framing. +#[derive(Clone, Debug, PartialEq, Eq)] +pub enum Message { + /// Asks the guest to answer with [`Message::Pong`]. + Ping, + /// The guest's answer to [`Message::Ping`]. + Pong, + /// Bytes that the guest sends back in an `Echo` of its own. + Echo(Vec), + /// Asks the guest to stop serving, which it confirms with [`Message::Stopped`]. + Stop, + /// The guest's confirmation of [`Message::Stop`], its last message. + Stopped, +} + +impl Message { + /// Returns the message's kind, the last byte of its header. + pub fn kind(&self) -> u8 { + self.parts().0.code() + } + + /// Returns the message's encoding: its header, then its body. An `Echo` body has at most + /// [`MAX_BODY`] bytes. + pub fn encode(&self) -> Result, FrameError> { + let (kind, body) = self.parts(); + if body.len() > kind.max_body() { + return Err(FrameError::Length { + kind: kind.code(), + length: body.len(), + }); + } + // The body has at most `MAX_BODY` bytes, so its length fits the header's field. + let length = body.len() as u32; + let mut encoded = Vec::with_capacity(HEADER_LEN + body.len()); + encoded.extend_from_slice(&length.to_le_bytes()); + encoded.push(kind.code()); + encoded.extend_from_slice(body); + Ok(encoded) + } + + /// Returns the message's kind and body. + fn parts(&self) -> (Kind, &[u8]) { + match self { + Self::Ping => (Kind::Ping, &[]), + Self::Pong => (Kind::Pong, &[]), + Self::Echo(body) => (Kind::Echo, body), + Self::Stop => (Kind::Stop, &[]), + Self::Stopped => (Kind::Stopped, &[]), + } + } +} + +/// Kind of a [`Message`]. +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +enum Kind { + /// [`Message::Ping`]. + Ping, + /// [`Message::Pong`]. + Pong, + /// [`Message::Echo`]. + Echo, + /// [`Message::Stop`]. + Stop, + /// [`Message::Stopped`]. + Stopped, +} + +impl Kind { + /// Returns the kind with `code`, if the framing defines one. + fn from_code(code: u8) -> Option { + match code { + 1 => Some(Self::Ping), + 2 => Some(Self::Pong), + 3 => Some(Self::Echo), + 4 => Some(Self::Stop), + 5 => Some(Self::Stopped), + _ => None, + } + } + + /// Returns the kind's code, the last byte of a header. + fn code(self) -> u8 { + match self { + Self::Ping => 1, + Self::Pong => 2, + Self::Echo => 3, + Self::Stop => 4, + Self::Stopped => 5, + } + } + + /// Returns the most bytes of a body of this kind. + fn max_body(self) -> usize { + match self { + Self::Echo => MAX_BODY, + Self::Ping | Self::Pong | Self::Stop | Self::Stopped => 0, + } + } + + /// Returns the message of this kind with `body`, which has at most + /// [`max_body`](Self::max_body) bytes. + fn message(self, body: Vec) -> Message { + match self { + Self::Ping => Message::Ping, + Self::Pong => Message::Pong, + Self::Echo => Message::Echo(body), + Self::Stop => Message::Stop, + Self::Stopped => Message::Stopped, + } + } +} + +/// A malformed message. +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +#[non_exhaustive] +pub enum FrameError { + /// A kind that the framing does not define. + Kind(u8), + /// A body longer than [`MAX_BODY`] bytes, or a body for a kind that has none. + Length { + /// Kind of the message. + kind: u8, + /// Bytes of the body. + length: usize, + }, +} + +impl fmt::Display for FrameError { + fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result { + match self { + Self::Kind(kind) => write!(formatter, "unknown message kind {kind}"), + Self::Length { kind, length } => write!( + formatter, + "message of kind {kind} with a body of {length} bytes" + ), + } + } +} + +impl std::error::Error for FrameError {} + +/// Reassembles messages from bytes that arrive in pieces of any size, such as the payloads of the +/// control session's records. +#[derive(Clone, Debug, Default)] +pub struct Decoder { + /// Header of the message in progress. + header: [u8; HEADER_LEN], + /// Bytes of the header that arrived. + header_len: usize, + /// Kind of the message in progress, once its header arrived and checked out. + kind: Option, + /// Bytes of the body that the header announced. + body_len: usize, + /// Bytes of the body that arrived. + body: Vec, +} + +impl Decoder { + /// Creates a decoder that expects the start of a message. + pub fn new() -> Self { + Self::default() + } + + /// Consumes bytes until a message completes or `input` runs out. + /// + /// Returns the number of bytes consumed and the message that the last of them completed, if + /// any. Bytes after that message are not consumed, so the caller passes them again. A + /// malformed header fails before any byte of its body is taken, and leaves the decoder + /// empty. + pub fn push(&mut self, input: &[u8]) -> Result<(usize, Option), FrameError> { + let mut consumed = 0; + let kind = match self.kind { + Some(kind) => kind, + None => { + consumed = (HEADER_LEN - self.header_len).min(input.len()); + self.header[self.header_len..][..consumed].copy_from_slice(&input[..consumed]); + self.header_len += consumed; + if self.header_len < HEADER_LEN { + return Ok((consumed, None)); + } + let [b0, b1, b2, b3, code] = self.header; + // A length that `usize` cannot hold exceeds every limit. + let length = + usize::try_from(u32::from_le_bytes([b0, b1, b2, b3])).unwrap_or(usize::MAX); + let kind = Kind::from_code(code).ok_or(FrameError::Kind(code)); + let checked = kind.and_then(|kind| { + if length > kind.max_body() { + Err(FrameError::Length { kind: code, length }) + } else { + Ok(kind) + } + }); + let kind = checked.inspect_err(|_| self.clear())?; + self.kind = Some(kind); + self.body_len = length; + self.body.reserve_exact(length); + kind + } + }; + let taken = (self.body_len - self.body.len()).min(input.len() - consumed); + self.body + .extend_from_slice(&input[consumed..consumed + taken]); + consumed += taken; + if self.body.len() < self.body_len { + return Ok((consumed, None)); + } + let body = std::mem::take(&mut self.body); + self.clear(); + Ok((consumed, Some(kind.message(body)))) + } + + /// Drops the message in progress, if any, such as when the session resets and its new epoch + /// starts with a new message. + pub fn clear(&mut self) { + *self = Self::default(); + } + + /// Returns whether part of a message arrived. + pub fn is_partial(&self) -> bool { + self.header_len != 0 + } +} + +#[cfg(test)] +mod tests { + use super::{Decoder, FrameError, HEADER_LEN, MAX_BODY, Message}; + + /// Returns one message of each kind, with bodies of no, few, and the most bytes. + fn messages() -> Vec { + vec![ + Message::Ping, + Message::Echo(Vec::new()), + Message::Echo(b"hello".to_vec()), + Message::Pong, + Message::Echo((0..=u8::MAX).cycle().take(MAX_BODY).collect()), + Message::Stop, + Message::Stopped, + ] + } + + /// Returns the encodings of `messages`, one after the other. + fn stream(messages: &[Message]) -> Vec { + messages + .iter() + .flat_map(|message| message.encode().unwrap()) + .collect() + } + + /// Decodes `pieces` in order, and returns the messages that they complete. + fn decode<'a>( + decoder: &mut Decoder, + pieces: impl IntoIterator, + ) -> Vec { + let mut decoded = Vec::new(); + for piece in pieces { + let mut rest = piece; + while !rest.is_empty() { + let (consumed, message) = decoder.push(rest).unwrap(); + assert_ne!(consumed, 0, "a push consumes bytes"); + rest = &rest[consumed..]; + decoded.extend(message); + } + } + decoded + } + + #[test] + fn the_header_holds_the_body_length_then_the_kind() { + assert_eq!(Message::Ping.encode().unwrap(), [0, 0, 0, 0, 1]); + assert_eq!(Message::Pong.encode().unwrap(), [0, 0, 0, 0, 2]); + assert_eq!( + Message::Echo(b"abc".to_vec()).encode().unwrap(), + [3, 0, 0, 0, 3, b'a', b'b', b'c'] + ); + assert_eq!(Message::Stop.encode().unwrap(), [0, 0, 0, 0, 4]); + assert_eq!(Message::Stopped.encode().unwrap(), [0, 0, 0, 0, 5]); + for message in messages() { + assert_eq!(message.encode().unwrap()[HEADER_LEN - 1], message.kind()); + } + } + + #[test] + fn messages_arrive_whole_in_pieces_of_any_size() { + let messages = messages(); + let stream = stream(&messages); + for size in [1, 2, 3, HEADER_LEN, 7, 4096, 64 * 1024, stream.len()] { + let mut decoder = Decoder::new(); + assert_eq!( + decode(&mut decoder, stream.chunks(size)), + messages, + "{size}" + ); + assert!(!decoder.is_partial()); + } + } + + #[test] + fn messages_arrive_whole_across_a_split_at_every_byte() { + let messages = [ + Message::Echo(b"split".to_vec()), + Message::Ping, + Message::Stop, + ]; + let stream = stream(&messages); + for split in 0..=stream.len() { + let mut decoder = Decoder::new(); + let pieces = [&stream[..split], &stream[split..]]; + assert_eq!(decode(&mut decoder, pieces), messages, "{split}"); + assert!(!decoder.is_partial()); + } + } + + #[test] + fn a_push_stops_at_the_end_of_a_message() { + let stream = stream(&[Message::Ping, Message::Echo(b"next".to_vec())]); + let mut decoder = Decoder::new(); + assert_eq!(decoder.push(&stream), Ok((HEADER_LEN, Some(Message::Ping)))); + assert!(!decoder.is_partial()); + assert_eq!( + decoder.push(&stream[HEADER_LEN..]), + Ok((HEADER_LEN + 4, Some(Message::Echo(b"next".to_vec())))) + ); + } + + #[test] + fn malformed_headers_fail_before_their_body_is_taken() { + let too_long = u32::try_from(MAX_BODY + 1).unwrap().to_le_bytes(); + let cases = [ + ([0, 0, 0, 0, 0], FrameError::Kind(0)), + ([0, 0, 0, 0, 6], FrameError::Kind(6)), + ([9, 0, 0, 0, 0xff], FrameError::Kind(0xff)), + ([1, 0, 0, 0, 1], FrameError::Length { kind: 1, length: 1 }), + ([1, 0, 0, 0, 4], FrameError::Length { kind: 4, length: 1 }), + ( + [too_long[0], too_long[1], too_long[2], too_long[3], 3], + FrameError::Length { + kind: 3, + length: MAX_BODY + 1, + }, + ), + ( + [0xff, 0xff, 0xff, 0xff, 3], + FrameError::Length { + kind: 3, + length: u32::MAX as usize, + }, + ), + ]; + for (header, error) in cases { + let mut decoder = Decoder::new(); + assert_eq!(decoder.push(&header[..2]), Ok((2, None))); + assert!(decoder.is_partial()); + let mut rest = header[2..].to_vec(); + rest.extend_from_slice(b"body"); + assert_eq!(decoder.push(&rest), Err(error), "{header:?}"); + assert!(!decoder.is_partial(), "{header:?}"); + assert_eq!(decoder.body.capacity(), 0, "{header:?}"); + } + } + + #[test] + fn clearing_drops_a_partial_message() { + let echo = Message::Echo(vec![1; 100]).encode().unwrap(); + let mut decoder = Decoder::new(); + assert_eq!(decoder.push(&echo[..50]), Ok((50, None))); + assert!(decoder.is_partial()); + decoder.clear(); + assert!(!decoder.is_partial()); + let ping = Message::Ping.encode().unwrap(); + assert_eq!(decoder.push(&ping), Ok((HEADER_LEN, Some(Message::Ping)))); + } + + #[test] + fn encoding_refuses_a_body_beyond_the_limit() { + let longest = Message::Echo(vec![0; MAX_BODY]).encode().unwrap(); + assert_eq!(longest.len(), HEADER_LEN + MAX_BODY); + assert_eq!( + Message::Echo(vec![0; MAX_BODY + 1]).encode(), + Err(FrameError::Length { + kind: 3, + length: MAX_BODY + 1, + }) + ); + } +} diff --git a/tests/runtime-adapter/src/guest.rs b/tests/runtime-adapter/src/guest.rs new file mode 100644 index 000000000..c81470d21 --- /dev/null +++ b/tests/runtime-adapter/src/guest.rs @@ -0,0 +1,124 @@ +//! The adapter's guest agent: it serves the host program over the guest's end of the control +//! session between the host and the guest, with the adapter's own [framing](crate::frame), through +//! `nvx::guest::nvxs` alone. + +use std::fmt; +use std::io::{Read, Write}; + +use nvx::guest::nvxs::{GuestConnection, GuestConnectionError}; +use nvx::protocol::OUTER_MIN_RECEIVE_WINDOW; +use nvx::protocol::nvxs::GuestEvent; + +use crate::frame::{Decoder, FrameError, Message}; + +/// The receive window that the guest acknowledges each epoch with: the smallest that the session +/// allows, so that the rest of a longer message arrives only as the guest credits what it took. +pub const RECEIVE_WINDOW: u32 = OUTER_MIN_RECEIVE_WINDOW; + +/// How serving ended, when nothing failed. +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub enum Served { + /// The host asked the guest to stop, and the guest confirmed. + Stopped, + /// The broker closed the stream at a record boundary. + Closed, +} + +/// A failure that ended serving. +#[derive(Debug)] +#[non_exhaustive] +pub enum ServeError { + /// The control session failed. + Session(GuestConnectionError), + /// The host sent a malformed message. + Frame(FrameError), + /// The host sent a message of a kind that only the guest sends. + Unexpected(u8), +} + +impl fmt::Display for ServeError { + fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result { + match self { + Self::Session(error) => error.fmt(formatter), + Self::Frame(error) => error.fmt(formatter), + Self::Unexpected(kind) => { + write!( + formatter, + "the host sent a message of kind {kind}, which only the guest sends" + ) + } + } + } +} + +impl std::error::Error for ServeError { + fn source(&self) -> Option<&(dyn std::error::Error + 'static)> { + match self { + Self::Session(error) => Some(error), + Self::Frame(error) => Some(error), + Self::Unexpected(_) => None, + } + } +} + +impl From for ServeError { + fn from(error: GuestConnectionError) -> Self { + Self::Session(error) + } +} + +impl From for ServeError { + fn from(error: FrameError) -> Self { + Self::Frame(error) + } +} + +/// Serves the host program on `stream`, the guest's end of the control session, until the host +/// stops the guest or the broker closes the stream. +/// +/// The guest attaches, then: +/// +/// - It acknowledges each reset of the session with [`RECEIVE_WINDOW`], once it dropped the +/// earlier epoch's partial message: a new epoch starts with a new message. +/// - It credits each payload back once it holds the payload, so that a message longer than the +/// window arrives whole. +/// - It answers [`Message::Ping`] with [`Message::Pong`], sends each [`Message::Echo`] back, and +/// answers [`Message::Stop`] with [`Message::Stopped`], then returns without reading further. +/// +/// A failure of the session, a malformed message, and a message that only the guest sends end +/// serving with an error. +pub fn serve(stream: S) -> Result { + let mut connection = GuestConnection::new(stream); + connection.attach()?; + let mut decoder = Decoder::new(); + loop { + let payload = match connection.next_event() { + Ok(GuestEvent::Reset(_)) => { + decoder.clear(); + connection.acknowledge(RECEIVE_WINDOW)?; + continue; + } + Ok(GuestEvent::Data(payload)) => payload, + Err(GuestConnectionError::Closed) => return Ok(Served::Closed), + Err(error) => return Err(error.into()), + }; + // A DATA payload has at most `OUTER_MAX_PAYLOAD` bytes, so its length fits a `u32`. + connection.credit(payload.len() as u32)?; + let mut rest = payload.as_slice(); + while !rest.is_empty() { + let (consumed, message) = decoder.push(rest)?; + rest = &rest[consumed..]; + let reply = match message { + None => continue, + Some(Message::Ping) => Message::Pong, + Some(Message::Echo(body)) => Message::Echo(body), + Some(Message::Stop) => { + connection.send(&Message::Stopped.encode()?)?; + return Ok(Served::Stopped); + } + Some(message) => return Err(ServeError::Unexpected(message.kind())), + }; + connection.send(&reply.encode()?)?; + } + } +} diff --git a/tests/runtime-adapter/src/lib.rs b/tests/runtime-adapter/src/lib.rs index 9bfffe743..602b28848 100644 --- a/tests/runtime-adapter/src/lib.rs +++ b/tests/runtime-adapter/src/lib.rs @@ -1,19 +1,25 @@ -//! An independent adapter of the NVX runtime. NVX's CI builds it against `nvx` the way an -//! adapter outside the repository would: as a path dependency with default features off, with a -//! lockfile of its own, and through the public API alone. +//! An independent adapter of the NVX runtime, with a host program and a guest agent. NVX's CI +//! builds it against `nvx` the way an adapter outside the repository would: as a path dependency +//! with default features off, with a lockfile of its own, and through the public API alone. //! //! Each configuration enables one side of `nvx`: //! //! - protocol-only, without features: the control protocol, the contract versions, and the other -//! items that `nvx` has without features; -//! - host-only, with the `host` feature: the host API too, and the `runtime-adapter-host` -//! binary; -//! - guest-only, with the `guest` feature: the guest API too, and the `runtime-adapter-guest` -//! binary, a static `x86_64-unknown-linux-musl` executable. +//! items that `nvx` has without features, with the adapter's own [`frame`]s; +//! - host-only, with the `host` feature: the host API too, and the host program, +//! `runtime-adapter-host`; +//! - guest-only, with the `guest` feature: the guest API too, and the guest agent, in the `guest` +//! module and the `runtime-adapter-guest` binary, a static `x86_64-unknown-linux-musl` +//! executable. //! -//! The binaries are skeletons that report the contract they were built against. Later steps give -//! the adapter its own framing over the NVXS session, fill its host and guest halves, and add the -//! scenarios that run them on a hypervisor. +//! The adapter brings its own framing to the control session between the host and the guest, in +//! place of NVX's `NVXC` frames. The guest agent serves requests in that framing, and +//! `runtime-adapter-guest serve` runs it on its standard input and output. The host program +//! reports the contract that it was built against. + +pub mod frame; +#[cfg(all(feature = "guest", target_os = "linux"))] +pub mod guest; /// Describes the runtime contract that this build speaks: the control contract revision, and the /// versions of the microVM ABI, the NVXS records, and the NVXC frames. diff --git a/tests/runtime-adapter/tests/guest_session.rs b/tests/runtime-adapter/tests/guest_session.rs new file mode 100644 index 000000000..283bba86b --- /dev/null +++ b/tests/runtime-adapter/tests/guest_session.rs @@ -0,0 +1,703 @@ +//! The adapter's guest agent serves the control session between the host and the guest with the +//! adapter's own framing, through public APIs only. +//! +//! A test broker plays the broker's end of the guest's session over a Unix socket pair, with the +//! public record API of `nvx`: for the guest agent's `serve` in a thread, and for the +//! `runtime-adapter-guest` binary, whose standard input and output are the socket. The tests cover +//! messages that span records, partial records, resets in the middle of a message, disconnects, +//! record lengths, bootstrap identity, ERROR payloads, sequence and identity errors, and the +//! bounded buffer of the adapter's decoder. +#![cfg(all(feature = "guest", target_os = "linux"))] + +use std::io::{self, Read, Write}; +use std::net::Shutdown; +use std::os::fd::OwnedFd; +use std::os::unix::net::UnixStream; +use std::process::{Child, Command, Output, Stdio}; +use std::sync::mpsc::{self, Receiver, Sender}; +use std::thread::{self, JoinHandle}; +use std::time::{Duration, Instant}; + +use nvx::guest::nvxs::GuestConnectionError; +use nvx::protocol::OUTER_MAX_PAYLOAD; +use nvx::protocol::nvxs::{ + ErrorCode, Identity, Record, RecordError, RecordParser, RecordType, SessionError, encode_record, +}; +use runtime_adapter::frame::{Decoder, FrameError, HEADER_LEN, MAX_BODY, Message}; +use runtime_adapter::guest::{self, RECEIVE_WINDOW, ServeError, Served}; + +/// Identity of the broker's first epoch. +const FIRST: Identity = Identity { + instance_id: [0x5c; 16], + epoch: 1, +}; + +/// Longest wait for a record from the guest, or for the guest to end. +const TIMEOUT: Duration = Duration::from_secs(30); + +/// Offset of the protocol version in a record's header. +const VERSION_OFFSET: usize = 4; +/// Offset of the record type in a record's header. +const TYPE_OFFSET: usize = 6; +/// Offset of the flags in a record's header. +const FLAGS_OFFSET: usize = 7; +/// Offset of the instance ID in a record's header. +const INSTANCE_OFFSET: usize = 8; +/// Offset of the payload length in a record's header. +const LENGTH_OFFSET: usize = 40; + +/// The broker's end of the guest's session. It resets the guest once the guest attaches, numbers +/// its records to the guest from 0 in each epoch, and sends DATA only while the guest's receive +/// window has room. It checks the identity and sequence number of each of the guest's records, +/// and panics on any record that the protocol does not allow. +struct Broker { + /// Stream to the guest. + stream: UnixStream, + /// The guest's records, which a thread reads as they arrive, so that the guest never waits to + /// write while the broker writes; then `None`, once the guest closed its stream at a record + /// boundary. + records: Receiver>, + /// Identity of the current epoch. + identity: Identity, + /// Sequence number of the broker's next record. + send_sequence: u64, + /// Sequence number of the guest's next record. + receive_sequence: u64, + /// Bytes for which the guest's receive window has room. + room: usize, + /// Decoder of the guest's messages. + decoder: Decoder, +} + +impl Broker { + /// Creates the broker on its end of a socket pair. + fn new(stream: UnixStream) -> Self { + let reader = stream.try_clone().unwrap(); + let (sender, records) = mpsc::channel(); + thread::spawn(move || read_records(reader, &sender)); + Self { + stream, + records, + identity: FIRST, + send_sequence: 0, + receive_sequence: 0, + room: 0, + decoder: Decoder::new(), + } + } + + /// Takes the guest's attachment, then starts the epoch of `identity`. + fn attach(&mut self, identity: Identity) { + let record = self.read().expect("the guest attaches"); + assert_eq!(record.record_type(), RecordType::GuestAttach); + let header = (record.instance_id(), record.epoch(), record.sequence()); + assert_eq!(header, ([0; 16], 0, 0)); + self.reset(identity); + } + + /// Starts the epoch of `identity` with a RESET, and takes the guest's acknowledgement. + fn reset(&mut self, identity: Identity) { + self.identity = identity; + self.send_sequence = 0; + self.receive_sequence = 0; + self.room = 0; + self.send(RecordType::Reset, &[]); + let ack = self.expect(); + assert_eq!(ack.record_type(), RecordType::Ack); + assert_eq!(ack.window(), Some(RECEIVE_WINDOW)); + self.room = RECEIVE_WINDOW as usize; + } + + /// Sends a record of the current epoch at the broker's next sequence number. + fn send(&mut self, record_type: RecordType, payload: &[u8]) { + let record = record(record_type, self.identity, self.send_sequence, payload); + self.send_sequence += 1; + self.write(&record); + } + + /// Writes bytes to the guest. A guest that stops reading fails the test once `TIMEOUT` passes, + /// instead of hanging it: the socket's buffer can fill before the guest's receive window does. + fn write(&mut self, mut bytes: &[u8]) { + let deadline = Instant::now() + TIMEOUT; + while !bytes.is_empty() { + let remaining = deadline.saturating_duration_since(Instant::now()); + assert!( + !remaining.is_zero(), + "the guest reads the broker's bytes in time" + ); + self.stream.set_write_timeout(Some(remaining)).unwrap(); + match self.stream.write(bytes) { + Ok(count) => bytes = &bytes[count..], + Err(error) + if matches!( + error.kind(), + io::ErrorKind::WouldBlock + | io::ErrorKind::TimedOut + | io::ErrorKind::Interrupted + ) => {} + Err(error) => panic!("cannot write to the guest: {error}"), + } + } + } + + /// Ends the broker's stream to the guest. + fn close(&self) { + self.stream.shutdown(Shutdown::Write).unwrap(); + } + + /// Returns the guest's next record, or `None` once the guest closed its stream at a record + /// boundary. + fn read(&mut self) -> Option { + self.records + .recv_timeout(TIMEOUT) + .expect("the guest sends a record or ends its stream in time") + } + + /// Reads the guest's next record, and checks that it carries the epoch's identity and the + /// guest's next sequence number. + fn expect(&mut self) -> Record { + let record = self.read().expect("the guest sends a record"); + let header = (record.instance_id(), record.epoch(), record.sequence()); + let expected = ( + self.identity.instance_id, + self.identity.epoch, + self.receive_sequence, + ); + assert_eq!(header, expected, "{record:?}"); + self.receive_sequence += 1; + record + } + + /// Takes the guest's next record: a CREDIT, which makes room in its window, or DATA, whose + /// complete messages it returns. + fn take(&mut self) -> Vec { + let record = self.expect(); + match record.record_type() { + RecordType::Credit => { + self.room += record.credit().unwrap() as usize; + assert!( + self.room <= RECEIVE_WINDOW as usize, + "the guest credited more than it received" + ); + Vec::new() + } + RecordType::Data => decode(&mut self.decoder, record.payload()), + record_type => panic!("the guest sent {record_type} in an active epoch"), + } + } + + /// Sends the host's `bytes` in DATA records of at most `size` bytes, each once the guest's + /// window has room for it, and returns the messages that the guest sent meanwhile. + fn send_data(&mut self, bytes: &[u8], size: usize) -> Vec { + let mut messages = Vec::new(); + for chunk in bytes.chunks(size) { + while self.room < chunk.len() { + messages.extend(self.take()); + } + self.room -= chunk.len(); + self.send(RecordType::Data, chunk); + } + messages + } + + /// Sends `messages` in DATA records of at most `size` bytes, and returns the guest's next + /// `replies` messages. + fn exchange(&mut self, messages: &[Message], size: usize, replies: usize) -> Vec { + let mut received = self.send_data(&encode(messages), size); + while received.len() < replies { + received.extend(self.take()); + } + assert_eq!(received.len(), replies, "{received:?}"); + received + } + + /// Takes the guest's credit for everything that the broker sent, so that no record of the + /// epoch is in flight. + fn settle(&mut self) { + while self.room < RECEIVE_WINDOW as usize { + assert_eq!(self.take(), [], "the guest sent no message"); + } + } +} + +/// Reads the guest's records from `stream` and sends them, then `None` once the guest closed its +/// stream at a record boundary. +fn read_records(mut stream: UnixStream, records: &Sender>) { + let mut parser = RecordParser::new(); + let mut buffer = [0; 4096]; + loop { + let wanted = parser.needed().min(buffer.len()); + let count = match stream.read(&mut buffer[..wanted]) { + Ok(count) => count, + // The guest closed its end before it read everything that the broker wrote, as after + // a record that it refused. Linux returns what the guest wrote first. + Err(error) if error.kind() == io::ErrorKind::ConnectionReset => 0, + Err(error) => panic!("cannot read the guest's stream: {error}"), + }; + if count == 0 { + assert_eq!( + parser.buffered(), + 0, + "the guest ended its stream inside a record" + ); + let _ = records.send(None); + return; + } + let (consumed, record) = parser.push(&buffer[..count]).unwrap(); + assert_eq!(consumed, count); + // A test that ended takes no more records. + if record.is_some_and(|record| records.send(Some(record)).is_err()) { + return; + } + } +} + +/// Encodes a valid record. +fn record(record_type: RecordType, identity: Identity, sequence: u64, payload: &[u8]) -> Vec { + let mut record = Vec::new(); + let (instance, epoch) = (identity.instance_id, identity.epoch); + encode_record( + record_type, + &instance, + epoch, + sequence, + payload, + &mut record, + ) + .unwrap(); + record +} + +/// Returns the encodings of `messages`, one after the other. +fn encode(messages: &[Message]) -> Vec { + messages + .iter() + .flat_map(|message| message.encode().unwrap()) + .collect() +} + +/// Returns the messages that `bytes` complete. +fn decode(decoder: &mut Decoder, mut bytes: &[u8]) -> Vec { + let mut messages = Vec::new(); + while !bytes.is_empty() { + let (consumed, message) = decoder.push(bytes).unwrap(); + bytes = &bytes[consumed..]; + messages.extend(message); + } + messages +} + +/// Returns an echo of the longest body: with its header, the message spans five records. +fn longest() -> Message { + Message::Echo((0..=u8::MAX).cycle().take(MAX_BODY).collect()) +} + +/// Starts the guest agent's `serve` in a thread, on one end of a socket pair, and returns the +/// broker on the other end with the thread. +fn start() -> (Broker, JoinHandle>) { + let (broker, guest) = UnixStream::pair().unwrap(); + let serving = thread::spawn(move || guest::serve(guest)); + (Broker::new(broker), serving) +} + +/// Returns how the guest agent's `serve` ended, and fails the test if it does not end in time, so +/// that a guest that keeps serving fails the test instead of hanging it. +fn joined(serving: JoinHandle>) -> Result { + let deadline = Instant::now() + TIMEOUT; + while !serving.is_finished() { + assert!(Instant::now() < deadline, "the guest agent ends in time"); + thread::sleep(Duration::from_millis(10)); + } + serving.join().unwrap() +} + +/// Starts the guest agent, attaches it in the first epoch, sends it `bytes`, and returns the +/// error that ended serving. +fn refused(bytes: &[u8]) -> ServeError { + let (mut broker, serving) = start(); + broker.attach(FIRST); + broker.write(bytes); + let error = joined(serving).unwrap_err(); + // The guest ended its stream. + while let Some(record) = broker.read() { + assert_eq!(record.record_type(), RecordType::Credit, "{error:?}"); + } + error +} + +#[test] +fn the_guest_agent_answers_messages_that_span_records() { + let (mut broker, serving) = start(); + broker.attach(FIRST); + let replies = broker.exchange(&[Message::Ping], OUTER_MAX_PAYLOAD, 1); + assert_eq!(replies, [Message::Pong]); + + // Messages in records of several sizes. The longest spans five records each way, and the + // guest's window holds a quarter of it, so the rest arrives as the guest credits it. + let messages = [ + Message::Ping, + Message::Echo(b"two".to_vec()), + longest(), + Message::Echo(Vec::new()), + Message::Ping, + ]; + let expected = [ + Message::Pong, + Message::Echo(b"two".to_vec()), + longest(), + Message::Echo(Vec::new()), + Message::Pong, + ]; + for size in [7, 4096, OUTER_MAX_PAYLOAD] { + let replies = broker.exchange(&messages, size, expected.len()); + assert_eq!(replies, expected, "records of {size} bytes"); + } + + // A record whose bytes arrive one at a time. + broker.settle(); + let ping = Message::Ping.encode().unwrap(); + let bytes = record(RecordType::Data, FIRST, broker.send_sequence, &ping); + broker.send_sequence += 1; + broker.room -= ping.len(); + for byte in bytes { + broker.write(&[byte]); + thread::sleep(Duration::from_millis(1)); + } + let mut replies = Vec::new(); + while replies.is_empty() { + replies = broker.take(); + } + assert_eq!(replies, [Message::Pong]); + + let replies = broker.exchange(&[Message::Stop], OUTER_MAX_PAYLOAD, 1); + assert_eq!(replies, [Message::Stopped]); + assert_eq!(joined(serving).unwrap(), Served::Stopped); + assert!(broker.read().is_none()); +} + +#[test] +fn a_reset_starts_the_new_epoch_with_a_new_message() { + let (mut broker, serving) = start(); + broker.attach(FIRST); + let echo = Message::Echo(vec![1; 1000]).encode().unwrap(); + + // Half of a message, then the host's connection drops, which starts the instance's next + // epoch. The guest drops the partial message. + assert_eq!(broker.send_data(&echo[..500], OUTER_MAX_PAYLOAD), []); + broker.settle(); + broker.reset(Identity { epoch: 2, ..FIRST }); + let after = Message::Echo(b"after".to_vec()); + let replies = broker.exchange(std::slice::from_ref(&after), OUTER_MAX_PAYLOAD, 1); + assert_eq!(replies, [after]); + + // A broker restored from a snapshot has a new instance, which starts at epoch 1, here in + // the middle of a header. + assert_eq!( + broker.send_data(&echo[..HEADER_LEN - 1], OUTER_MAX_PAYLOAD), + [] + ); + broker.settle(); + broker.reset(Identity { + instance_id: [0x7e; 16], + epoch: 1, + }); + let replies = broker.exchange(&[Message::Ping, Message::Stop], OUTER_MAX_PAYLOAD, 2); + assert_eq!(replies, [Message::Pong, Message::Stopped]); + assert_eq!(joined(serving).unwrap(), Served::Stopped); +} + +#[test] +fn the_guest_agent_ends_with_its_stream() { + // A stream that ends at a record boundary closes the session. + let (mut broker, serving) = start(); + broker.attach(FIRST); + let replies = broker.exchange(&[Message::Ping], OUTER_MAX_PAYLOAD, 1); + assert_eq!(replies, [Message::Pong]); + broker.close(); + assert_eq!(joined(serving).unwrap(), Served::Closed); + assert!(broker.read().is_none()); + + // One that ends inside a record fails it. + let (mut broker, serving) = start(); + broker.attach(FIRST); + let ping = record(RecordType::Data, FIRST, 1, &Message::Ping.encode().unwrap()); + broker.write(&ping[..ping.len() - 1]); + broker.close(); + let error = joined(serving).unwrap_err(); + let truncated = matches!( + &error, + ServeError::Session(GuestConnectionError::Protocol(SessionError::Truncated { buffered })) + if *buffered == ping.len() - 1 + ); + assert!(truncated, "{error:?}"); +} + +#[test] +fn the_guest_agent_fails_on_records_that_the_protocol_does_not_allow() { + let ping = Message::Ping.encode().unwrap(); + let data = |identity, sequence| record(RecordType::Data, identity, sequence, &ping); + let next_epoch = Identity { epoch: 2, ..FIRST }; + let other_instance = Identity { + instance_id: [0x7e; 16], + ..FIRST + }; + let patched = |mut bytes: Vec, offset: usize, patch: &[u8]| { + bytes[offset..offset + patch.len()].copy_from_slice(patch); + bytes + }; + // A RESET with a payload, an empty DATA record, and a GUEST_ATTACH with an identity, which + // the record API cannot encode. + let mut reset_with_payload = patched( + record(RecordType::Reset, next_epoch, 1, &[]), + LENGTH_OFFSET, + &4u32.to_le_bytes(), + ); + reset_with_payload.extend_from_slice(&[0; 4]); + let mut empty_data = patched(data(FIRST, 1), LENGTH_OFFSET, &0u32.to_le_bytes()); + empty_data.truncate(empty_data.len() - ping.len()); + let attach_with_identity = patched( + record( + RecordType::GuestAttach, + Identity { + instance_id: [0; 16], + epoch: 0, + }, + 0, + &[], + ), + INSTANCE_OFFSET, + &[1], + ); + let error_record = |code: u32| { + let payload = ErrorCode::Authentication.code().to_le_bytes(); + patched( + record(RecordType::Error, FIRST, 1, &payload), + LENGTH_OFFSET + 4, + &code.to_le_bytes(), + ) + }; + let wrong_magic = patched(data(FIRST, 1), 0, b"NVXT"); + let cases = [ + ( + "a skipped sequence number", + data(FIRST, 2), + SessionError::Sequence { + expected: 1, + actual: 2, + }, + ), + ( + "a repeated sequence number", + data(FIRST, 0), + SessionError::Sequence { + expected: 1, + actual: 0, + }, + ), + ( + "DATA of another epoch", + data(next_epoch, 1), + SessionError::Epoch { + expected: 1, + actual: 2, + }, + ), + ( + "DATA of another instance", + data(other_instance, 1), + SessionError::Instance { + expected: FIRST.instance_id, + actual: other_instance.instance_id, + }, + ), + ( + "a RESET that repeats the epoch", + record(RecordType::Reset, FIRST, 1, &[]), + SessionError::StaleEpoch { + current: 1, + actual: 1, + }, + ), + ( + "a RESET without an identity", + record( + RecordType::Reset, + Identity { + instance_id: [0; 16], + epoch: 0, + }, + 1, + &[], + ), + SessionError::MissingIdentity(RecordType::Reset), + ), + ( + "a RESET with a payload", + reset_with_payload, + SessionError::Record(RecordError::Length { + record_type: RecordType::Reset, + length: 4, + }), + ), + ( + "an empty DATA record", + empty_data, + SessionError::Record(RecordError::Length { + record_type: RecordType::Data, + length: 0, + }), + ), + ( + "a bootstrap record with an identity", + attach_with_identity, + SessionError::Record(RecordError::BootstrapIdentity(RecordType::GuestAttach)), + ), + ( + "an ERROR record with an unknown code", + error_record(2), + SessionError::Record(RecordError::ErrorCode(2)), + ), + ( + "an ERROR record", + error_record(ErrorCode::Authentication.code()), + SessionError::Unexpected(RecordType::Error), + ), + ( + "an ACK record", + record(RecordType::Ack, FIRST, 1, &RECEIVE_WINDOW.to_le_bytes()), + SessionError::Unexpected(RecordType::Ack), + ), + ( + "an unknown record type", + patched(data(FIRST, 1), TYPE_OFFSET, &[0x7f]), + SessionError::Record(RecordError::Type(0x7f)), + ), + ( + "nonzero flags", + patched(data(FIRST, 1), FLAGS_OFFSET, &[1]), + SessionError::Record(RecordError::Flags(1)), + ), + ( + "another protocol version", + patched(data(FIRST, 1), VERSION_OFFSET, &2u16.to_le_bytes()), + SessionError::Record(RecordError::Version(2)), + ), + ( + "a wrong magic", + wrong_magic, + SessionError::Record(RecordError::Magic(*b"NVXT")), + ), + ]; + for (name, bytes, expected) in cases { + let error = refused(&bytes); + let failed = matches!( + &error, + ServeError::Session(GuestConnectionError::Protocol(error)) if *error == expected + ); + assert!(failed, "{name}: {error:?}"); + } +} + +#[test] +fn the_guest_agent_refuses_messages_that_its_framing_does_not_allow() { + // A message is refused at its header, before its body arrives: the decoder never buffers a + // body beyond the limit. + let too_long = u32::try_from(MAX_BODY + 1).unwrap().to_le_bytes(); + let header = |length: [u8; 4], kind: u8| { + let mut header = length.to_vec(); + header.push(kind); + header + }; + let cases = [ + (header([0; 4], 0), FrameError::Kind(0)), + (header([0; 4], 9), FrameError::Kind(9)), + ( + header(too_long, 3), + FrameError::Length { + kind: 3, + length: MAX_BODY + 1, + }, + ), + ( + header(1u32.to_le_bytes(), 1), + FrameError::Length { kind: 1, length: 1 }, + ), + ]; + for (message, expected) in cases { + let error = refused(&record(RecordType::Data, FIRST, 1, &message)); + let failed = matches!(&error, ServeError::Frame(error) if *error == expected); + assert!(failed, "{message:?}: {error:?}"); + } + + // The guest agent answers requests, and refuses the messages that only it sends. + for message in [Message::Pong, Message::Stopped] { + let bytes = record(RecordType::Data, FIRST, 1, &message.encode().unwrap()); + let error = refused(&bytes); + let failed = matches!(&error, ServeError::Unexpected(kind) if *kind == message.kind()); + assert!(failed, "{message:?}: {error:?}"); + } +} + +/// Starts the guest binary's `serve` with one end of a socket pair as its standard input and +/// output, and returns the broker on the other end with the process. +fn start_binary() -> (Broker, Child) { + let (broker, guest) = UnixStream::pair().unwrap(); + let output = guest.try_clone().unwrap(); + let child = Command::new(env!("CARGO_BIN_EXE_runtime-adapter-guest")) + .arg("serve") + .stdin(Stdio::from(OwnedFd::from(guest))) + .stdout(Stdio::from(OwnedFd::from(output))) + .stderr(Stdio::piped()) + .spawn() + .unwrap(); + (Broker::new(broker), child) +} + +/// Returns the guest binary's output once it exits. If it does not exit in time, kills and reaps +/// it, and fails the test instead of hanging it. +fn finished(mut child: Child) -> Output { + let deadline = Instant::now() + TIMEOUT; + while child.try_wait().unwrap().is_none() { + if Instant::now() >= deadline { + let _ = child.kill(); + let output = child.wait_with_output().unwrap(); + panic!("the guest binary did not exit in time: {output:?}"); + } + thread::sleep(Duration::from_millis(10)); + } + child.wait_with_output().unwrap() +} + +#[test] +fn the_guest_binary_serves_the_session_on_its_standard_streams() { + // The host stops the guest. + let (mut broker, child) = start_binary(); + broker.attach(FIRST); + let echo = Message::Echo(vec![0x42; OUTER_MAX_PAYLOAD]); + let messages = [Message::Ping, echo.clone(), Message::Stop]; + let replies = broker.exchange(&messages, OUTER_MAX_PAYLOAD, 3); + assert_eq!(replies, [Message::Pong, echo, Message::Stopped]); + let output = finished(child); + assert!(output.status.success(), "{output:?}"); + assert!(output.stderr.is_empty(), "{output:?}"); + + // The broker closes the stream at a record boundary. + let (mut broker, child) = start_binary(); + broker.attach(FIRST); + broker.close(); + let output = finished(child); + assert!(output.status.success(), "{output:?}"); + assert!(output.stderr.is_empty(), "{output:?}"); + + // The host sends a malformed message. + let (mut broker, child) = start_binary(); + broker.attach(FIRST); + broker.send_data(&[0, 0, 0, 0, 9], OUTER_MAX_PAYLOAD); + let output = finished(child); + assert_eq!(output.status.code(), Some(1), "{output:?}"); + let diagnostic = String::from_utf8_lossy(&output.stderr); + assert_eq!( + diagnostic, + "runtime-adapter-guest: unknown message kind 9\n" + ); +} diff --git a/tests/runtime-adapter/tests/independent_build.rs b/tests/runtime-adapter/tests/independent_build.rs index f2d433f7c..b0b51a307 100644 --- a/tests/runtime-adapter/tests/independent_build.rs +++ b/tests/runtime-adapter/tests/independent_build.rs @@ -1,4 +1,4 @@ -//! Consumer-contract scenario C0: the adapter builds on `nvx` independently. +//! The adapter builds on `nvx` independently, as an adapter outside the repository does. //! //! - It is a workspace of its own, with a lockfile of its own, and declares the minimum supported //! Rust version of `nvx`.