From 7b99f49f62de20124b7daaed6f08416dc7a40f8f Mon Sep 17 00:00:00 2001 From: KAAL1 Date: Sat, 5 Sep 2026 00:12:55 +0300 Subject: [PATCH 1/2] feat(nesinit): mount what the descriptor names, and relay the layer it cannot read MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Builds on the previous change, which had PID 1, the channel and the trait but mounted nothing. The shares are mounted now: a tag names an export, the descriptor names where it lands, and every share goes on nosuid and nodev whether or not it is writable — a share is data handed to the guest, and no descriptor has a way to ask for a setuid binary or a device node in one. Mounting needs privileges a test does not have, so the arguments and flags are derived by a function the tests can assert, which is where the read-only decision lives. Progress is reported in two messages rather than one. A share that did not mount and a command that did not run are not the same incident, and each carries the reason the operating system gave and the path it happened on: a permission error on a named directory can be acted on, where "the share did not mount" cannot. The second layer is relayed and never read. Bytes arrive on the channel in an envelope, cross a unix socket to the workload, and come back the same way. The body is a string rather than nested JSON on purpose: a document this component can index into is a document it can grow to depend on, and then the layer is no longer opaque and the boundary it exists to draw is gone. An envelope is never logged — not the body, not truncated, not at debug level — and the channel name with a byte count is the whole of what may be said about one. The type's Debug is written by hand for the same reason, because a derived one puts the body one careless format string away from a log line. A write to a channel nobody is reading now ends the session the same way a closed read does. A caller that has stopped listening has also stopped being able to say stop, which is one situation and was two outcomes. The guest listens on the relay socket and the workload dials in, which is the convention the other guest sockets already use and removes the startup ordering problem: a workload that is not running yet has simply not connected yet. --- apps/nesinit/README.md | 70 +++++- apps/nesinit/src/lib.rs | 1 + apps/nesinit/src/main.rs | 24 ++- apps/nesinit/src/payload.rs | 237 ++++++++++++++++++++ apps/nesinit/src/session.rs | 320 +++++++++++++++++++++++++--- apps/nesinit/src/workload.rs | 121 ++++++++++- crates/nesprotocol/src/lifecycle.rs | 126 +++++++++++ 7 files changed, 846 insertions(+), 53 deletions(-) create mode 100644 apps/nesinit/src/payload.rs diff --git a/apps/nesinit/README.md b/apps/nesinit/README.md index 0dcc43f3..ed63d7d6 100644 --- a/apps/nesinit/README.md +++ b/apps/nesinit/README.md @@ -25,6 +25,8 @@ The guest dials out on a fixed vsock port and speaks first: ``` guest → { "type": "ready", "protocol_version": 2 } guest ← { "type": "boot", "exec": {...}, "mounts": [...], "geometry": {...}, "on_exit": {...} } +guest → { "type": "mounted" } +guest → { "type": "started" } guest → { "type": "workload_exited", "exit_code": 0 } ``` @@ -42,6 +44,56 @@ The types are in [`nesprotocol::lifecycle`](../../crates/nesprotocol/src/lifecyc behind the `lifecycle` feature, so both ends of the channel read one definition and neither can drift from it silently. +`mounted` / `mount_failed` stay separate from `started` / `start_failed` +because the two want different things looked at: a share that did not appear +and a command that did not run are not the same incident. A failure carries the +reason in the words the operating system used, and the path it happened on — a +permission error on a named directory can be acted on, where "the share did not +mount" cannot. + +### Two layers, one channel + +The channel carries a lifecycle layer, above, and a payload layer that nesinit +relays and never reads: + +``` +{ "type": "payload", "channel": "", "body": "" } +``` + +Both directions. Inside the guest an envelope crosses a unix socket at +`/nestri/payload.sock`, which the guest listens on and the workload dials into. +That socket is a mechanism and expected to change; the envelope is the boundary +and is not. + +`body` is a string rather than nested JSON, deliberately. A document nesinit +can index into is a document nesinit can grow to depend on, and then the layer +is no longer opaque and the boundary it exists to draw is gone. + +**An envelope is never logged.** Not the body, not truncated, not at debug +level. The channel name and the byte count are the whole of what may be said +about one — what crosses here includes credentials meant for the workload and +nothing else. `Payload`'s `Debug` is written by hand for the same reason: a +derived one puts the body one careless `{:?}` away from a log line. + +### The shares + +Each `mounts` entry is a tag, a path to put it at, and whether it is read-only. +The tag names an export and is never a path on the other side of the channel, +so the guest learns nothing about the filesystem it is handed a piece of. +Choosing *where* a share lands is the descriptor's job, not the guest's: +deciding that means knowing what the workload expects to find there, which is +exactly the knowledge a workload-independent init does not have. + +Every share is mounted `nosuid` and `nodev`, whether or not it is writable. A +share is data handed to the guest, and no descriptor has a way to ask for a +setuid binary or a device node in one. + +`uid` and `gid` in `exec` are load-bearing rather than hygiene. Whoever writes +the descriptor also exported the writable share, so the two have to agree; when +they do not, the first write is refused and the failure surfaces here as a +permission error with a path, instead of as a workload that misbehaves much +later for no visible reason. + ### It reports; it does not supervise When the workload ends, the exit goes up the channel and the session is over. @@ -54,9 +106,9 @@ A signalled workload is reported as signalled, with no exit code. Reporting ### What is not here yet -Mounting shares. The descriptor's `mounts` are refused rather than ignored — a -workload started without the shares it was promised fails later, somewhere -else, for a reason nobody can see from the guest. +`geometry` is carried and parsed but nothing consumes it: nesinit does not +start the guest's own services yet. `ticket` exists as a message with no +producer wired to it. ### Testing @@ -64,8 +116,10 @@ else, for a reason nobody can see from the guest. cargo test -p nesinit ``` -No VM required, and that is the point of the two seams. Reaping is tested -against real forked children — `PR_SET_CHILD_SUBREAPER` makes a test process -inherit orphans the same way PID 1 does — and the channel is tested over an -in-memory pipe, because the transport contributes nothing to the protocol -beyond ordering and framing. +No VM required, and that is the point of the seams. Reaping is tested against +real forked children — `PR_SET_CHILD_SUBREAPER` makes a test process inherit +orphans the same way PID 1 does. The channel is tested over an in-memory pipe, +because the transport contributes nothing to the protocol beyond ordering and +framing. The relay is tested over a real unix socket. Mounting needs +privileges a test does not have, so what is asserted is the arguments and flags +the mount is given, which is where the read-only and `nosuid` decisions live. diff --git a/apps/nesinit/src/lib.rs b/apps/nesinit/src/lib.rs index 2ff5eecb..8ce6a6fc 100644 --- a/apps/nesinit/src/lib.rs +++ b/apps/nesinit/src/lib.rs @@ -8,6 +8,7 @@ // and what an exit means, and it carries that out; a field that only makes // sense for one kind of workload cannot reach it. ref(d-0033) +pub mod payload; pub mod reap; pub mod session; pub mod shutdown; diff --git a/apps/nesinit/src/main.rs b/apps/nesinit/src/main.rs index 9040150f..2952761f 100644 --- a/apps/nesinit/src/main.rs +++ b/apps/nesinit/src/main.rs @@ -4,8 +4,10 @@ // the workload the channel describes, and turn the end of either into an // ordered shutdown. +use std::path::Path; use std::time::Duration; +use nesinit::payload::{self, Ports}; use nesinit::reap::{self, Waiters}; use nesinit::session::{self, Outcome}; use nesinit::shutdown::{self, Machine}; @@ -17,6 +19,12 @@ use tokio_vsock::{VMADDR_CID_HOST, VsockAddr, VsockStream}; /// How long a process gets between being asked to stop and being made to. const GRACE: Duration = Duration::from_secs(10); +/// How many envelopes may be in flight in one direction. +/// +/// Small on purpose: what crosses this layer is re-sent when it changes, so a +/// deep queue holds stale copies of it rather than protecting anything. +const RELAY_DEPTH: usize = 8; + fn main() -> anyhow::Result<()> { tracing_subscriber::fmt() .with_env_filter( @@ -75,8 +83,22 @@ async fn guest(waiters: &Waiters, workload: &mut Process) -> anyhow::Result outcome?, + outcome = session::run(channel, workload, &mut ports) => outcome?, signal = asked_to_stop() => { signal?; tracing::info!("asked to stop"); diff --git a/apps/nesinit/src/payload.rs b/apps/nesinit/src/payload.rs new file mode 100644 index 00000000..1340f967 --- /dev/null +++ b/apps/nesinit/src/payload.rs @@ -0,0 +1,237 @@ +// The relay for the layer the guest does not read. +// +// Bytes arrive on the control channel inside an envelope, are handed to the +// workload over a unix socket, and come back the same way. Nothing here parses +// a body, and nothing here logs one. +// +// The socket is the soft part of this: it is a mechanism, where the envelope is +// a boundary. Expect the socket to change and do not let a change to it change +// anything above it. + +use std::io; +use std::path::Path; + +use nesprotocol::lifecycle::{Payload, from_line, to_line}; +use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader}; +use tokio::net::{UnixListener, UnixStream}; +use tokio::sync::mpsc::{Receiver, Sender}; + +/// Where the workload finds the relay. +pub const SOCKET: &str = "/nestri/payload.sock"; + +/// The two directions, as the session sees them. +pub struct Ports { + /// Envelopes bound for the workload. + pub to_workload: Sender, + /// Envelopes the workload sent. + pub from_workload: Receiver, +} + +/// Listen for the workload and relay in both directions for as long as it is +/// connected. +/// +/// The guest listens and the workload dials in, which removes the startup +/// ordering problem: a workload that is not running yet has simply not +/// connected yet, and one that reconnects gets the relay again. +pub async fn serve( + path: &Path, + mut outbound: Receiver, + inbound: Sender, +) -> io::Result<()> { + if let Some(directory) = path.parent() { + std::fs::create_dir_all(directory)?; + } + // A socket left behind by a previous boot would refuse the bind. Nothing + // durable lives in the guest, so there is nothing here to preserve. + let _ = std::fs::remove_file(path); + let listener = UnixListener::bind(path)?; + + loop { + let (stream, _) = listener.accept().await?; + tracing::info!("the workload is on the relay"); + match relay(stream, &mut outbound, &inbound).await { + Ok(()) => tracing::info!("the workload left the relay"), + Err(error) => tracing::warn!(%error, "the relay connection ended"), + } + } +} + +async fn relay( + stream: UnixStream, + outbound: &mut Receiver, + inbound: &Sender, +) -> io::Result<()> { + let (reader, mut writer) = stream.into_split(); + let mut lines = BufReader::new(reader).lines(); + + loop { + tokio::select! { + line = lines.next_line() => { + let Some(line) = line? else { return Ok(()) }; + // Deliberately not `{:?}` on the error either: a decoder is + // entitled to quote what it could not read, and what it could + // not read is a body. + let Ok(payload) = from_line::(&line) else { + tracing::warn!(bytes = line.len(), "ignoring an envelope that would not decode"); + continue; + }; + tracing::debug!(envelope = %payload.summary(), "relaying from the workload"); + if inbound.send(payload).await.is_err() { + return Ok(()); // the session is over + } + } + outgoing = outbound.recv() => { + let Some(payload) = outgoing else { return Ok(()) }; + tracing::debug!(envelope = %payload.summary(), "relaying to the workload"); + let line = to_line(&payload).map_err(io::Error::other)?; + writer.write_all(line.as_bytes()).await?; + writer.flush().await?; + } + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + use std::sync::{Arc, Mutex}; + use tokio::sync::mpsc; + + fn socket_path(name: &str) -> std::path::PathBuf { + std::env::temp_dir().join(format!("nesinit-{}-{name}.sock", std::process::id())) + } + + /// Somewhere to put log output so a test can read it back. + #[derive(Clone, Default)] + struct Capture(Arc>>); + + impl Capture { + fn text(&self) -> String { + String::from_utf8_lossy(&self.0.lock().unwrap()).into_owned() + } + } + + impl io::Write for Capture { + fn write(&mut self, buf: &[u8]) -> io::Result { + self.0.lock().unwrap().extend_from_slice(buf); + Ok(buf.len()) + } + fn flush(&mut self) -> io::Result<()> { + Ok(()) + } + } + + impl<'a> tracing_subscriber::fmt::MakeWriter<'a> for Capture { + type Writer = Capture; + fn make_writer(&'a self) -> Self::Writer { + self.clone() + } + } + + #[tokio::test] + async fn an_envelope_crosses_in_both_directions_untouched() { + let path = socket_path("both-ways"); + let (down_tx, down_rx) = mpsc::channel(4); + let (up_tx, mut up_rx) = mpsc::channel(4); + let server = tokio::spawn({ + let path = path.clone(); + async move { serve(&path, down_rx, up_tx).await } + }); + + let workload = connect(&path).await; + let (reader, mut writer) = workload.into_split(); + let mut lines = BufReader::new(reader).lines(); + + let body = r#"{"looks":"structured"} and is not"#; + down_tx.send(Payload::new("identity", body)).await.unwrap(); + let line = lines.next_line().await.unwrap().unwrap(); + let seen: Payload = from_line(&line).unwrap(); + assert_eq!(seen.body, body, "the body arrived changed"); + assert_eq!(seen.channel, "identity"); + + let back = to_line(&Payload::new("identity", "opaque back")).unwrap(); + writer.write_all(back.as_bytes()).await.unwrap(); + let up = up_rx.recv().await.unwrap(); + assert_eq!(up.body, "opaque back"); + + server.abort(); + let _ = std::fs::remove_file(&path); + } + + #[tokio::test] + async fn nothing_the_relay_logs_contains_a_body() { + let capture = Capture::default(); + let subscriber = tracing_subscriber::fmt() + .with_writer(capture.clone()) + .with_ansi(false) + .with_max_level(tracing::Level::TRACE) + .finish(); + let _log = tracing::subscriber::set_default(subscriber); + + let path = socket_path("logging"); + let (down_tx, down_rx) = mpsc::channel(4); + let (up_tx, mut up_rx) = mpsc::channel(4); + let server = tokio::spawn({ + let path = path.clone(); + async move { serve(&path, down_rx, up_tx).await } + }); + + let workload = connect(&path).await; + let (reader, mut writer) = workload.into_split(); + let mut lines = BufReader::new(reader).lines(); + + let secret = "a-credential-nobody-should-read"; + down_tx + .send(Payload::new("identity", secret)) + .await + .unwrap(); + lines.next_line().await.unwrap().unwrap(); + + writer + .write_all( + to_line(&Payload::new("identity", secret)) + .unwrap() + .as_bytes(), + ) + .await + .unwrap(); + up_rx.recv().await.unwrap(); + + // An envelope that will not decode is the other way a body reaches a + // log line — a decoder is entitled to quote what it could not read — + // so a broken one carrying the same secret goes through the same + // check. + let malformed = format!("{{\"channel\":\"identity\",\"body\":\"{secret}\"\n"); + writer.write_all(malformed.as_bytes()).await.unwrap(); + tokio::time::sleep(std::time::Duration::from_millis(50)).await; + + let logged = capture.text(); + assert!( + !logged.contains(secret), + "a body reached a log line:\n{logged}" + ); + // The relay's own line about a line it could not read says how long it + // was and nothing else. What may be said about an envelope it *could* + // read is asserted where that summary is written, because callsite + // interest is cached process-wide and a debug line another test + // reached first will not arrive here. + assert!( + logged.contains("bytes=62"), + "the byte count is loggable:\n{logged}" + ); + + server.abort(); + let _ = std::fs::remove_file(&path); + } + + /// The listener may not be bound the instant the task is spawned. + async fn connect(path: &Path) -> UnixStream { + for _ in 0..100 { + if let Ok(stream) = UnixStream::connect(path).await { + return stream; + } + tokio::time::sleep(std::time::Duration::from_millis(10)).await; + } + panic!("the relay never came up on {}", path.display()); + } +} diff --git a/apps/nesinit/src/session.rs b/apps/nesinit/src/session.rs index e0e21af1..0bb302cb 100644 --- a/apps/nesinit/src/session.rs +++ b/apps/nesinit/src/session.rs @@ -8,11 +8,12 @@ // restarting is repair or a loop. ref(d-0033) use nesprotocol::lifecycle::{ - BootDescriptor, CONTROL_VERSION, Exit, GuestToHost, HostToGuest, from_line, to_line, + CONTROL_VERSION, Exit, GuestToHost, HostToGuest, Payload, from_line, to_line, }; use tokio::io::{AsyncBufReadExt, AsyncRead, AsyncWrite, AsyncWriteExt, BufReader}; -use crate::workload::{Failure, Workload}; +use crate::payload::Ports; +use crate::workload::{Exited, Failure, Workload}; /// How a session ended. #[derive(Debug, Clone, PartialEq, Eq)] @@ -34,7 +35,42 @@ pub enum Outcome { /// Generic over the channel so the exchange can be driven from a test without /// a VM: the transport contributes nothing to the protocol beyond ordering and /// framing, which any byte stream has. -pub async fn run(channel: C, workload: &mut W) -> std::io::Result +pub async fn run( + channel: C, + workload: &mut W, + payload: &mut Ports, +) -> std::io::Result +where + C: AsyncRead + AsyncWrite, + W: Workload, +{ + match converse(channel, workload, payload).await { + Err(error) if channel_gone(&error) => { + // A caller that has stopped reading has also stopped being able to + // tell us to stop, which is the same situation as the channel + // closing under a read. One outcome, not two. + workload.signal_stop(); + Ok(Outcome::ChannelClosed) + } + other => other, + } +} + +/// Whether an error means the far end is gone rather than that something went +/// wrong here. +fn channel_gone(error: &std::io::Error) -> bool { + use std::io::ErrorKind::{BrokenPipe, ConnectionAborted, ConnectionReset, UnexpectedEof}; + matches!( + error.kind(), + BrokenPipe | ConnectionReset | ConnectionAborted | UnexpectedEof + ) +} + +async fn converse( + channel: C, + workload: &mut W, + payload: &mut Ports, +) -> std::io::Result where C: AsyncRead + AsyncWrite, W: Workload, @@ -53,19 +89,39 @@ where ) .await?; - let mut running: Option = None; + let mut running: Option = None; + let mut relay_open = true; loop { - let line = match running.as_mut() { + let event = match running.as_mut() { Some(exited) => tokio::select! { - ended = exited => { - let exit = ended?; - send(&mut writer, &GuestToHost::WorkloadExited { exit }).await?; - return Ok(Outcome::WorkloadExited(exit)); - } - line = lines.next_line() => line?, + ended = exited => Event::Ended(ended?), + line = lines.next_line() => Event::Line(line?), + up = payload.from_workload.recv(), if relay_open => Event::FromWorkload(up), }, - None => lines.next_line().await?, + None => tokio::select! { + line = lines.next_line() => Event::Line(line?), + up = payload.from_workload.recv(), if relay_open => Event::FromWorkload(up), + }, + }; + + let line = match event { + Event::Ended(exit) => { + send(&mut writer, &GuestToHost::WorkloadExited { exit }).await?; + return Ok(Outcome::WorkloadExited(exit)); + } + Event::FromWorkload(Some(payload)) => { + tracing::debug!(envelope = %payload.summary(), "sending an envelope on"); + send(&mut writer, &GuestToHost::Payload { payload }).await?; + continue; + } + Event::FromWorkload(None) => { + // The relay is gone. The session is not: the workload can + // still be stopped, and its exit still has to be reported. + relay_open = false; + continue; + } + Event::Line(line) => line, }; let Some(line) = line else { @@ -92,31 +148,65 @@ where tracing::warn!("ignoring a second descriptor: one is read per connection"); continue; } - match begin(&descriptor, workload) { - Ok(exited) => running = Some(exited), + + // The shares, then the command, and each reported separately. + // Which of the two failed decides what is worth looking at, + // so the two are never one message. + match workload.mount(&descriptor.mounts) { + Ok(()) => send(&mut writer, &GuestToHost::Mounted).await?, Err(failure) => { - tracing::error!(reason = %failure.reason, "the descriptor was refused"); + send( + &mut writer, + &GuestToHost::MountFailed { + reason: failure.reason.clone(), + }, + ) + .await?; + return Ok(Outcome::Refused(failure)); + } + } + + match workload.start(&descriptor.exec) { + Ok(exited) => { + send(&mut writer, &GuestToHost::Started).await?; + running = Some(exited); + } + Err(failure) => { + send( + &mut writer, + &GuestToHost::StartFailed { + reason: failure.reason.clone(), + }, + ) + .await?; return Ok(Outcome::Refused(failure)); } } } + HostToGuest::Payload { payload: envelope } => hand_over(payload, envelope).await, HostToGuest::Stop => workload.signal_stop(), HostToGuest::Shutdown => return Ok(Outcome::Shutdown), } } } -/// Carry out a descriptor: shares first, then the command. -/// -/// The two stay distinguishable on the way out because they want different -/// things looked at — a share that did not mount and a command that did not -/// start are not the same incident. -fn begin( - descriptor: &BootDescriptor, - workload: &mut W, -) -> Result { - workload.mount(&descriptor.mounts)?; - workload.start(&descriptor.exec) +/// What the session is waiting on, and there are only three things. +enum Event { + Line(Option), + Ended(Exit), + FromWorkload(Option), +} + +/// Hand an envelope to the relay, and treat a relay that is not there as the +/// caller's problem rather than a failure of this session. +async fn hand_over(ports: &mut Ports, envelope: Payload) { + tracing::debug!(envelope = %envelope.summary(), "handing an envelope over"); + if ports.to_workload.send(envelope).await.is_err() { + // The workload is not on the relay. Dropped rather than queued: what + // crosses here is re-sent when it changes, so a held copy is a stale + // copy. + tracing::warn!("dropped an envelope: nothing is on the relay"); + } } async fn send(writer: &mut W, message: &GuestToHost) -> std::io::Result<()> @@ -132,8 +222,9 @@ where mod tests { use super::*; use crate::workload::double::Double; - use nesprotocol::lifecycle::{Exec, Geometry, Mount, OnExit}; + use nesprotocol::lifecycle::{BootDescriptor, Exec, Geometry, Mount, OnExit}; use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader, DuplexStream}; + use tokio::sync::mpsc; fn descriptor() -> BootDescriptor { BootDescriptor { @@ -159,6 +250,21 @@ mod tests { } } + /// The relay's two ends, as the session sees them, plus the ends a + /// workload on the relay would hold. + fn ports() -> (Ports, mpsc::Receiver, mpsc::Sender) { + let (down_tx, down_rx) = mpsc::channel(4); + let (up_tx, up_rx) = mpsc::channel(4); + ( + Ports { + to_workload: down_tx, + from_workload: up_rx, + }, + down_rx, + up_tx, + ) + } + /// The other end of the channel, as a caller would drive it. struct Caller { lines: tokio::io::Lines>, @@ -181,6 +287,12 @@ mod tests { from_line(&line).unwrap() } + /// A descriptor being carried out: the shares, then the command. + async fn expect_started(&mut self) { + assert_eq!(self.expect().await, GuestToHost::Mounted); + assert_eq!(self.expect().await, GuestToHost::Started); + } + async fn say(&mut self, message: &HostToGuest) { let line = to_line(message).unwrap(); self.lines @@ -197,8 +309,9 @@ mod tests { let mut caller = Caller::new(host); let session = tokio::spawn(async move { + let (mut ports, _to_workload, _from_workload) = ports(); let mut workload = Double::exits_when_stopped(Exit::code(0)); - let outcome = run(guest, &mut workload).await.unwrap(); + let outcome = run(guest, &mut workload, &mut ports).await.unwrap(); (outcome, workload) }); @@ -221,8 +334,9 @@ mod tests { let mut caller = Caller::new(host); let session = tokio::spawn(async move { + let (mut ports, _to_workload, _from_workload) = ports(); let mut workload = Double::exits_when_stopped(Exit::code(0)); - let outcome = run(guest, &mut workload).await.unwrap(); + let outcome = run(guest, &mut workload, &mut ports).await.unwrap(); (outcome, workload) }); @@ -246,8 +360,9 @@ mod tests { let mut caller = Caller::new(host); let session = tokio::spawn(async move { + let (mut ports, _to_workload, _from_workload) = ports(); let mut workload = Double::exits_at_once(Exit::code(3)); - let outcome = run(guest, &mut workload).await.unwrap(); + let outcome = run(guest, &mut workload, &mut ports).await.unwrap(); (outcome, workload) }); @@ -257,6 +372,7 @@ mod tests { descriptor: Box::new(descriptor()), }) .await; + caller.expect_started().await; assert_eq!( caller.expect().await, @@ -280,8 +396,9 @@ mod tests { let mut caller = Caller::new(host); let session = tokio::spawn(async move { + let (mut ports, _to_workload, _from_workload) = ports(); let mut workload = Double::exits_at_once(Exit::signal(9)); - run(guest, &mut workload).await.unwrap() + run(guest, &mut workload, &mut ports).await.unwrap() }); assert!(matches!(caller.expect().await, GuestToHost::Ready { .. })); @@ -290,6 +407,7 @@ mod tests { descriptor: Box::new(descriptor()), }) .await; + caller.expect_started().await; assert_eq!( caller.expect().await, @@ -309,8 +427,9 @@ mod tests { let mut caller = Caller::new(host); let session = tokio::spawn(async move { + let (mut ports, _to_workload, _from_workload) = ports(); let mut workload = Double::exits_when_stopped(Exit::code(0)); - let outcome = run(guest, &mut workload).await.unwrap(); + let outcome = run(guest, &mut workload, &mut ports).await.unwrap(); (outcome, workload) }); @@ -333,9 +452,10 @@ mod tests { let mut caller = Caller::new(host); let session = tokio::spawn(async move { + let (mut ports, _to_workload, _from_workload) = ports(); let mut workload = Double::exits_at_once(Exit::code(0)); workload.mount_failure = Some(Failure::new("EACCES: /mnt/user")); - let outcome = run(guest, &mut workload).await.unwrap(); + let outcome = run(guest, &mut workload, &mut ports).await.unwrap(); (outcome, workload) }); @@ -346,6 +466,14 @@ mod tests { }) .await; + assert_eq!( + caller.expect().await, + GuestToHost::MountFailed { + reason: "EACCES: /mnt/user".into() + }, + "the reason is passed through as the operating system wrote it", + ); + let (outcome, workload) = session.await.unwrap(); assert_eq!(outcome, Outcome::Refused(Failure::new("EACCES: /mnt/user"))); assert!( @@ -360,8 +488,9 @@ mod tests { let mut caller = Caller::new(host); let session = tokio::spawn(async move { + let (mut ports, _to_workload, _from_workload) = ports(); let mut workload = Double::exits_when_stopped(Exit::code(0)); - run(guest, &mut workload).await.unwrap() + run(guest, &mut workload, &mut ports).await.unwrap() }); assert!(matches!(caller.expect().await, GuestToHost::Ready { .. })); @@ -382,8 +511,9 @@ mod tests { let mut caller = Caller::new(host); let session = tokio::spawn(async move { + let (mut ports, _to_workload, _from_workload) = ports(); let mut workload = Double::exits_when_stopped(Exit::code(0)); - let outcome = run(guest, &mut workload).await.unwrap(); + let outcome = run(guest, &mut workload, &mut ports).await.unwrap(); (outcome, workload) }); @@ -405,4 +535,124 @@ mod tests { "the workload was left running with nobody listening" ); } + + #[tokio::test] + async fn a_command_that_will_not_run_is_reported_apart_from_a_share_that_will_not_mount() { + let (guest, host) = tokio::io::duplex(4096); + let mut caller = Caller::new(host); + + let session = tokio::spawn(async move { + let (mut ports, _to_workload, _from_workload) = ports(); + let mut workload = Double::exits_at_once(Exit::code(0)); + workload.start_failure = Some(Failure::new("ENOENT: /usr/bin/workload")); + let outcome = run(guest, &mut workload, &mut ports).await.unwrap(); + (outcome, workload) + }); + + assert!(matches!(caller.expect().await, GuestToHost::Ready { .. })); + caller + .say(&HostToGuest::Boot { + descriptor: Box::new(descriptor()), + }) + .await; + + // The shares are reported as fine, and the failure is a different + // message: which of the two went wrong decides what to look at. + assert_eq!(caller.expect().await, GuestToHost::Mounted); + assert_eq!( + caller.expect().await, + GuestToHost::StartFailed { + reason: "ENOENT: /usr/bin/workload".into() + }, + ); + + let (outcome, workload) = session.await.unwrap(); + assert_eq!( + outcome, + Outcome::Refused(Failure::new("ENOENT: /usr/bin/workload")) + ); + assert_eq!(workload.mounted.len(), 1); + } + + #[tokio::test] + async fn an_envelope_crosses_the_session_in_both_directions_unread() { + let (guest, host) = tokio::io::duplex(4096); + let mut caller = Caller::new(host); + let (down_tx, down_rx) = mpsc::channel(4); + let (up_tx, up_rx) = mpsc::channel(4); + + let session = tokio::spawn(async move { + let mut ports = Ports { + to_workload: down_tx, + from_workload: up_rx, + }; + let mut workload = Double::exits_when_stopped(Exit::code(0)); + run(guest, &mut workload, &mut ports).await.unwrap() + }); + let mut to_relay = down_rx; + + assert!(matches!(caller.expect().await, GuestToHost::Ready { .. })); + + // Down: an envelope arrives before any descriptor does, and still + // reaches the relay — what crosses this layer is not the boot + // sequence's business. + let body = r#"{"looks":"structured"} and is not"#; + caller + .say(&HostToGuest::Payload { + payload: Payload::new("identity", body), + }) + .await; + let handed_over = to_relay.recv().await.unwrap(); + assert_eq!(handed_over.body, body, "the body arrived changed"); + assert_eq!(handed_over.channel, "identity"); + + // Up: the same, in reverse. + up_tx + .send(Payload::new("identity", "opaque back")) + .await + .unwrap(); + assert_eq!( + caller.expect().await, + GuestToHost::Payload { + payload: Payload::new("identity", "opaque back") + }, + ); + + caller.say(&HostToGuest::Shutdown).await; + assert_eq!(session.await.unwrap(), Outcome::Shutdown); + } + + #[tokio::test] + async fn a_relay_nothing_is_on_does_not_end_a_session() { + let (guest, host) = tokio::io::duplex(4096); + let mut caller = Caller::new(host); + let (down_tx, down_rx) = mpsc::channel(1); + let (_up_tx, up_rx) = mpsc::channel::(1); + drop(down_rx); // nothing is on the relay + + let session = tokio::spawn(async move { + let mut ports = Ports { + to_workload: down_tx, + from_workload: up_rx, + }; + let mut workload = Double::exits_when_stopped(Exit::code(0)); + run(guest, &mut workload, &mut ports).await.unwrap() + }); + + assert!(matches!(caller.expect().await, GuestToHost::Ready { .. })); + caller + .say(&HostToGuest::Payload { + payload: Payload::new("identity", "dropped"), + }) + .await; + caller + .say(&HostToGuest::Boot { + descriptor: Box::new(descriptor()), + }) + .await; + caller.expect_started().await; + caller.say(&HostToGuest::Shutdown).await; + + assert_eq!(session.await.unwrap(), Outcome::Shutdown); + } } diff --git a/apps/nesinit/src/workload.rs b/apps/nesinit/src/workload.rs index 0ef5433f..ef5d00f6 100644 --- a/apps/nesinit/src/workload.rs +++ b/apps/nesinit/src/workload.rs @@ -6,6 +6,7 @@ // reported rather than acted on — is behaviour of the caller, which a double // can test without a VM, a share or a workload. +use std::ffi::CString; use std::future::Future; use std::io; use std::pin::Pin; @@ -119,16 +120,13 @@ impl Process { impl Workload for Process { fn mount(&mut self, mounts: &[Mount]) -> Result<(), Failure> { - if mounts.is_empty() { - return Ok(()); + for share in mounts { + // Stops at the first failure rather than mounting what it can: a + // workload given some of its shares fails later, somewhere else, + // for a reason nobody can see from here. + mount_share(share)?; } - // Refused rather than ignored: a workload started without the shares - // it was promised fails later, somewhere else, for a reason nobody can - // see from here. - Err(Failure::new(format!( - "this build mounts nothing; {} share(s) were requested", - mounts.len() - ))) + Ok(()) } fn start(&mut self, exec: &Exec) -> Result { @@ -188,6 +186,111 @@ impl Workload for Process { } } +/// Mount one share where the descriptor says to put it. +/// +/// The tag names an export; nothing here is a path on the other side of the +/// channel, so the guest still learns nothing about the filesystem it is being +/// handed a piece of. +fn mount_share(share: &Mount) -> Result<(), Failure> { + // The mount point may not exist yet: a share can land anywhere the + // descriptor names, including a directory no image created. + std::fs::create_dir_all(&share.at).map_err(|error| failed(share, error))?; + + let (source, target, flags) = options(share); + // SAFETY: mount takes two paths, a filesystem name and a flag word, all + // of which outlive the call, and no options string. + let mounted = unsafe { + libc::mount( + source.as_ptr(), + target.as_ptr(), + FSTYPE.as_ptr(), + flags, + std::ptr::null(), + ) + }; + if mounted != 0 { + return Err(failed(share, io::Error::last_os_error())); + } + Ok(()) +} + +/// The shares arrive over a virtio transport, which is the only kind of +/// filesystem this mounts. A descriptor cannot name another. +const FSTYPE: &std::ffi::CStr = c"virtiofs"; + +/// What the mount call is given, split out because this is the part worth +/// asserting: mounting itself needs privileges a test does not have. +fn options(share: &Mount) -> (CString, CString, libc::c_ulong) { + // nosuid and nodev on every share, whether or not it is writable. A share + // is data handed to the guest; a setuid binary or a device node appearing + // in one is not something a workload should be able to use, and no + // descriptor has a way to ask for it. + let mut flags = libc::MS_NOSUID | libc::MS_NODEV; + if share.ro { + flags |= libc::MS_RDONLY; + } + // Interior nul bytes are the caller's mistake; a path with one cannot be + // mounted under any flags. + let source = CString::new(share.tag.as_str()).unwrap_or_default(); + let target = CString::new(share.at.as_str()).unwrap_or_default(); + (source, target, flags) +} + +/// A failure names the path, which is what makes it actionable: a permission +/// error and the directory it happened on can be acted on, where "the share +/// did not mount" cannot. +fn failed(share: &Mount, error: io::Error) -> Failure { + Failure::new(format!("{}: {error}", share.at)) +} + +#[cfg(test)] +mod tests { + use super::*; + + fn share(ro: bool) -> Mount { + Mount { + tag: "user".into(), + at: "/mnt/user".into(), + ro, + } + } + + #[test] + fn a_writable_share_is_still_mounted_without_devices_or_setuid() { + let (source, target, flags) = options(&share(false)); + assert_eq!( + source.to_str().unwrap(), + "user", + "the tag is the source, never a path" + ); + assert_eq!(target.to_str().unwrap(), "/mnt/user"); + assert_eq!(flags & libc::MS_NOSUID, libc::MS_NOSUID); + assert_eq!(flags & libc::MS_NODEV, libc::MS_NODEV); + assert_eq!(flags & libc::MS_RDONLY, 0); + } + + #[test] + fn a_read_only_share_is_mounted_read_only() { + let (_, _, flags) = options(&share(true)); + assert_eq!(flags & libc::MS_RDONLY, libc::MS_RDONLY); + } + + #[test] + fn a_failure_names_the_path_it_happened_on() { + let failure = failed(&share(false), io::Error::from_raw_os_error(libc::EACCES)); + assert!( + failure.reason.starts_with("/mnt/user: "), + "{}", + failure.reason + ); + assert!( + failure.reason.contains("ermission denied"), + "{}", + failure.reason + ); + } +} + #[cfg(test)] pub mod double { use super::*; diff --git a/crates/nesprotocol/src/lifecycle.rs b/crates/nesprotocol/src/lifecycle.rs index f08448ea..8589eb8b 100644 --- a/crates/nesprotocol/src/lifecycle.rs +++ b/crates/nesprotocol/src/lifecycle.rs @@ -146,12 +146,31 @@ impl Exit { pub enum GuestToHost { /// First line on the connection, before anything else is read or written. Ready { protocol_version: u32 }, + /// Every share the descriptor named is where it said to put it. + Mounted, + /// A share could not be mounted, in the words the operating system used. + /// + /// Kept separate from `StartFailed` because the two want different things + /// looked at: a share that did not appear and a command that did not run + /// are not the same incident. + MountFailed { reason: String }, + /// The command the descriptor named is running. + Started, + /// The command could not be run, in the words the operating system used. + StartFailed { reason: String }, /// The workload the descriptor named has ended. Terminal or not is the /// descriptor's answer, not this message's. WorkloadExited { #[serde(flatten)] exit: Exit, }, + /// How a client reaches this box's media, once it is known. + Ticket { ticket: String }, + /// Bytes from the workload, relayed. See [`Payload`]. + Payload { + #[serde(flatten)] + payload: Payload, + }, } /// What the guest is told. @@ -167,6 +186,55 @@ pub enum HostToGuest { Stop, /// Shut the guest down. Shutdown, + /// Bytes for the workload, relayed. See [`Payload`]. + Payload { + #[serde(flatten)] + payload: Payload, + }, +} + +/// The second layer of the channel: bytes the guest carries and never reads. +/// +/// `body` is a string rather than nested JSON, and that is the structural part +/// of it. A document the guest can index into is a document the guest can grow +/// to depend on, and then this layer is no longer opaque and the boundary it +/// exists to draw is gone. +/// +/// **An envelope is never logged.** Not the body, not truncated, not at debug +/// level. The channel name and the byte count are the whole of what may be +/// said about one, because what crosses here includes credentials meant for +/// the workload and nothing else. ref(d-0033) +#[derive(Clone, PartialEq, Eq, Serialize, Deserialize)] +pub struct Payload { + /// Which conversation this belongs to. Loggable. + pub channel: String, + /// Opaque bytes. Never logged, never parsed, never inspected. + pub body: String, +} + +impl Payload { + pub fn new(channel: impl Into, body: impl Into) -> Self { + Self { + channel: channel.into(), + body: body.into(), + } + } + + /// What may be said about an envelope, and all of it. + pub fn summary(&self) -> String { + format!("{} ({} bytes)", self.channel, self.body.len()) + } +} + +/// Written by hand, and it is load-bearing: a derived `Debug` puts the body +/// one careless `{:?}` away from a log line. +impl std::fmt::Debug for Payload { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + f.debug_struct("Payload") + .field("channel", &self.channel) + .field("body", &format_args!("<{} bytes>", self.body.len())) + .finish() + } } /// Encode one message as a line, framing included. @@ -253,6 +321,64 @@ mod tests { ); } + #[test] + fn an_envelope_does_not_print_its_body() { + let payload = Payload::new("identity", "a-credential-nobody-should-read"); + let printed = format!("{payload:?}"); + assert!( + !printed.contains("a-credential"), + "the body reached a log line: {printed}" + ); + assert!( + printed.contains("identity"), + "the channel name is loggable: {printed}" + ); + assert_eq!(payload.summary(), "identity (31 bytes)"); + } + + #[test] + fn an_envelope_body_stays_a_string_in_both_directions() { + // Nested JSON in the body has to survive as text: the moment it + // arrives as structure, this layer is one field access from being + // parsed. + let body = r#"{"looks":"structured"}"#; + let line = to_line(&GuestToHost::Payload { + payload: Payload::new("identity", body), + }) + .unwrap(); + let back: GuestToHost = from_line(&line).unwrap(); + let GuestToHost::Payload { payload } = back else { + panic!("not an envelope: {line}") + }; + assert_eq!(payload.body, body); + + let line = to_line(&HostToGuest::Payload { + payload: Payload::new("identity", body), + }) + .unwrap(); + let back: HostToGuest = from_line(&line).unwrap(); + let HostToGuest::Payload { payload } = back else { + panic!("not an envelope: {line}") + }; + assert_eq!(payload.body, body); + } + + #[test] + fn a_mount_failure_keeps_its_reason_verbatim() { + let reason = "EACCES: /mnt/user"; + let line = to_line(&GuestToHost::MountFailed { + reason: reason.into(), + }) + .unwrap(); + let back: GuestToHost = from_line(&line).unwrap(); + assert_eq!( + back, + GuestToHost::MountFailed { + reason: reason.into() + } + ); + } + #[test] fn defaults_cover_what_a_caller_may_leave_out() { let json = r#"{"exec":{"argv":["/bin/sh"],"uid":1000,"gid":1000}, From a94b323edb1479aaab9a7045fbd6143ab95d7a94 Mon Sep 17 00:00:00 2001 From: KAAL1 Date: Sat, 5 Sep 2026 09:48:34 +0300 Subject: [PATCH 2/2] fix(nesinit): the relay may not stall the session, and may not buffer without end MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Three problems in the relay, all of them found in review. Handing an envelope over waited for room. That loop also carries stop, shutdown and the workload's exit, so a workload slow to read its own mail — or one that never connected — could hold the lifecycle layer still behind it. It never waits now: an envelope that will not fit is dropped, which costs nothing, because what crosses this layer is re-sent when it changes. Envelopes were queued for a workload that was not there. The queue filled with copies that would be stale by the time anyone connected, and filling it was what stalled the session. Nothing is held while the socket has nobody on it. A frame had no maximum length. The workload can write for as long as it likes without ever sending a newline, and the process assembling that is the one the kernel has been told not to kill, so the memory it takes comes out of everything else in the guest. Past 64 KiB the connection is dropped and the relay waits for the next one; the failure says how long the frame got and nothing about what was in it. Also, a tag or a mount point with a nul byte in it was quietly turned into an empty string, so an unmountable descriptor arrived later as a mount failure about something else, after the mount point had already been created. It is refused by name now, before anything is created. The relay's tests grew a harness that waits for the connection to be carried before sending anything down it, because dropping what arrives with nobody connected made "connected" something a test has to establish rather than assume. --- apps/nesinit/README.md | 13 ++ apps/nesinit/src/payload.rs | 303 +++++++++++++++++++++++++++++------ apps/nesinit/src/session.rs | 78 ++++++++- apps/nesinit/src/workload.rs | 53 ++++-- 4 files changed, 381 insertions(+), 66 deletions(-) diff --git a/apps/nesinit/README.md b/apps/nesinit/README.md index ed63d7d6..0aec304e 100644 --- a/apps/nesinit/README.md +++ b/apps/nesinit/README.md @@ -65,6 +65,19 @@ Both directions. Inside the guest an envelope crosses a unix socket at That socket is a mechanism and expected to change; the envelope is the boundary and is not. +Nothing is held for a workload that is not on the relay, and nothing waits on +one that is slow to read. An envelope that arrives with nobody connected is +dropped, and so is one that arrives faster than the workload reads: what +crosses this layer is re-sent when it changes, so a queued copy is a stale copy +— and the queue that would hold it is on the same loop that carries stop, +shutdown and the workload's exit, none of which may wait behind it. + +A frame is capped at 64 KiB. The workload is on the other end of that socket +and can write for as long as it likes without ever sending a newline; the +process assembling it is the one the kernel has been told not to kill, so an +unbounded buffer there comes out of everything else in the guest. Past the cap +the connection is dropped and the relay waits for the next one. + `body` is a string rather than nested JSON, deliberately. A document nesinit can index into is a document nesinit can grow to depend on, and then the layer is no longer opaque and the boundary it exists to draw is gone. diff --git a/apps/nesinit/src/payload.rs b/apps/nesinit/src/payload.rs index 1340f967..96b3a4ad 100644 --- a/apps/nesinit/src/payload.rs +++ b/apps/nesinit/src/payload.rs @@ -12,13 +12,22 @@ use std::io; use std::path::Path; use nesprotocol::lifecycle::{Payload, from_line, to_line}; -use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader}; +use tokio::io::{AsyncBufRead, AsyncBufReadExt, AsyncWriteExt, BufReader}; use tokio::net::{UnixListener, UnixStream}; use tokio::sync::mpsc::{Receiver, Sender}; /// Where the workload finds the relay. pub const SOCKET: &str = "/nestri/payload.sock"; +/// The longest envelope this will assemble before giving up on the connection. +/// +/// A cap rather than a preference. The workload is on the other end of this +/// socket and can write for as long as it likes without ever sending a +/// newline; without a limit, the process holding the buffer is the one process +/// in the guest the kernel has been told not to kill, so the memory it takes +/// comes out of everything else. +const LONGEST_ENVELOPE: usize = 64 * 1024; + /// The two directions, as the session sees them. pub struct Ports { /// Envelopes bound for the workload. @@ -47,7 +56,25 @@ pub async fn serve( let listener = UnixListener::bind(path)?; loop { - let (stream, _) = listener.accept().await?; + let stream = tokio::select! { + accepted = listener.accept() => accepted?.0, + waiting = outbound.recv() => { + // Dropped rather than queued for whoever connects next. What + // crosses this layer is re-sent when it changes, so what a + // queue would hold is a stale copy, and holding it is also + // what would eventually block the session that fills it. + match waiting { + Some(envelope) => { + tracing::warn!( + envelope = %envelope.summary(), + "dropped an envelope: nothing is on the relay", + ); + continue; + } + None => return Ok(()), // the session is over + } + } + }; tracing::info!("the workload is on the relay"); match relay(stream, &mut outbound, &inbound).await { Ok(()) => tracing::info!("the workload left the relay"), @@ -62,19 +89,27 @@ async fn relay( inbound: &Sender, ) -> io::Result<()> { let (reader, mut writer) = stream.into_split(); - let mut lines = BufReader::new(reader).lines(); + let mut reader = BufReader::new(reader); + let mut frame = Vec::new(); loop { tokio::select! { - line = lines.next_line() => { - let Some(line) = line? else { return Ok(()) }; + read = read_capped(&mut reader, &mut frame) => { + if !read? { + return Ok(()); // the workload closed the socket + } // Deliberately not `{:?}` on the error either: a decoder is // entitled to quote what it could not read, and what it could // not read is a body. - let Ok(payload) = from_line::(&line) else { - tracing::warn!(bytes = line.len(), "ignoring an envelope that would not decode"); + let decoded = std::str::from_utf8(&frame) + .ok() + .and_then(|line| from_line::(line).ok()); + let Some(payload) = decoded else { + tracing::warn!(bytes = frame.len(), "ignoring an envelope that would not decode"); + frame.clear(); continue; }; + frame.clear(); tracing::debug!(envelope = %payload.summary(), "relaying from the workload"); if inbound.send(payload).await.is_err() { return Ok(()); // the session is over @@ -91,10 +126,62 @@ async fn relay( } } +/// Read one newline-terminated frame into `frame`, refusing to grow it past +/// [`LONGEST_ENVELOPE`]. `false` at end of stream. +/// +/// Cancel-safe, which it has to be to sit in a `select!`: bytes are copied out +/// of the reader and consumed together, so a cancelled read leaves the partial +/// frame in `frame` and the rest of it in the socket. +async fn read_capped(reader: &mut R, frame: &mut Vec) -> io::Result +where + R: AsyncBufRead + Unpin, +{ + loop { + let consumed; + let complete; + { + let available = reader.fill_buf().await?; + if available.is_empty() { + return Ok(false); + } + match available.iter().position(|byte| *byte == b'\n') { + Some(end) => { + within_cap(frame.len() + end)?; + frame.extend_from_slice(&available[..end]); + consumed = end + 1; + complete = true; + } + None => { + within_cap(frame.len() + available.len())?; + frame.extend_from_slice(available); + consumed = available.len(); + complete = false; + } + } + } + reader.consume(consumed); + if complete { + return Ok(true); + } + } +} + +/// The error says how long, and nothing about what: what did not fit is a +/// body, and a body is not logged even when it is malformed. +fn within_cap(length: usize) -> io::Result<()> { + if length > LONGEST_ENVELOPE { + return Err(io::Error::other(format!( + "an envelope grew past {LONGEST_ENVELOPE} bytes without ending" + ))); + } + Ok(()) +} + #[cfg(test)] mod tests { use super::*; use std::sync::{Arc, Mutex}; + use tokio::net::unix::{OwnedReadHalf, OwnedWriteHalf}; use tokio::sync::mpsc; fn socket_path(name: &str) -> std::path::PathBuf { @@ -128,22 +215,79 @@ mod tests { } } + /// A relay, and the two ends of it a test drives. + struct Relay { + path: std::path::PathBuf, + down: mpsc::Sender, + up: mpsc::Receiver, + server: tokio::task::JoinHandle>, + } + + impl Relay { + fn start(name: &str) -> Self { + let path = socket_path(name); + let (down, outbound) = mpsc::channel(8); + let (inbound, up) = mpsc::channel(8); + let server = tokio::spawn({ + let path = path.clone(); + async move { serve(&path, outbound, inbound).await } + }); + Self { + path, + down, + up, + server, + } + } + + /// Dial the relay and wait until it is demonstrably carrying the + /// connection. + /// + /// `serve` drops what arrives while nothing is connected, so a test + /// that sends downward before the accept has completed is racing it. A + /// line upward is the barrier: it can only arrive once the relay is + /// carrying this connection. + async fn workload(&mut self) -> (BufReader, OwnedWriteHalf) { + let (reader, mut writer) = self.dial().await.into_split(); + writer + .write_all(to_line(&Payload::new("handshake", "")).unwrap().as_bytes()) + .await + .unwrap(); + assert_eq!(self.up.recv().await.unwrap().channel, "handshake"); + (BufReader::new(reader), writer) + } + + /// The listener may not be bound the instant the task is spawned. + async fn dial(&self) -> UnixStream { + for _ in 0..100 { + if let Ok(stream) = UnixStream::connect(&self.path).await { + return stream; + } + tokio::time::sleep(std::time::Duration::from_millis(10)).await; + } + panic!("the relay never came up on {}", self.path.display()); + } + } + + impl Drop for Relay { + fn drop(&mut self) { + self.server.abort(); + let _ = std::fs::remove_file(&self.path); + } + } + #[tokio::test] async fn an_envelope_crosses_in_both_directions_untouched() { - let path = socket_path("both-ways"); - let (down_tx, down_rx) = mpsc::channel(4); - let (up_tx, mut up_rx) = mpsc::channel(4); - let server = tokio::spawn({ - let path = path.clone(); - async move { serve(&path, down_rx, up_tx).await } - }); - - let workload = connect(&path).await; - let (reader, mut writer) = workload.into_split(); - let mut lines = BufReader::new(reader).lines(); + let mut relay = Relay::start("both-ways"); + let (reader, mut writer) = relay.workload().await; + let mut lines = reader.lines(); let body = r#"{"looks":"structured"} and is not"#; - down_tx.send(Payload::new("identity", body)).await.unwrap(); + relay + .down + .send(Payload::new("identity", body)) + .await + .unwrap(); let line = lines.next_line().await.unwrap().unwrap(); let seen: Payload = from_line(&line).unwrap(); assert_eq!(seen.body, body, "the body arrived changed"); @@ -151,11 +295,7 @@ mod tests { let back = to_line(&Payload::new("identity", "opaque back")).unwrap(); writer.write_all(back.as_bytes()).await.unwrap(); - let up = up_rx.recv().await.unwrap(); - assert_eq!(up.body, "opaque back"); - - server.abort(); - let _ = std::fs::remove_file(&path); + assert_eq!(relay.up.recv().await.unwrap().body, "opaque back"); } #[tokio::test] @@ -168,20 +308,13 @@ mod tests { .finish(); let _log = tracing::subscriber::set_default(subscriber); - let path = socket_path("logging"); - let (down_tx, down_rx) = mpsc::channel(4); - let (up_tx, mut up_rx) = mpsc::channel(4); - let server = tokio::spawn({ - let path = path.clone(); - async move { serve(&path, down_rx, up_tx).await } - }); - - let workload = connect(&path).await; - let (reader, mut writer) = workload.into_split(); - let mut lines = BufReader::new(reader).lines(); + let mut relay = Relay::start("logging"); + let (reader, mut writer) = relay.workload().await; + let mut lines = reader.lines(); let secret = "a-credential-nobody-should-read"; - down_tx + relay + .down .send(Payload::new("identity", secret)) .await .unwrap(); @@ -195,7 +328,7 @@ mod tests { ) .await .unwrap(); - up_rx.recv().await.unwrap(); + relay.up.recv().await.unwrap(); // An envelope that will not decode is the other way a body reaches a // log line — a decoder is entitled to quote what it could not read — @@ -219,19 +352,91 @@ mod tests { logged.contains("bytes=62"), "the byte count is loggable:\n{logged}" ); - - server.abort(); - let _ = std::fs::remove_file(&path); } - /// The listener may not be bound the instant the task is spawned. - async fn connect(path: &Path) -> UnixStream { - for _ in 0..100 { - if let Ok(stream) = UnixStream::connect(path).await { - return stream; - } - tokio::time::sleep(std::time::Duration::from_millis(10)).await; - } - panic!("the relay never came up on {}", path.display()); + #[tokio::test] + async fn an_envelope_sent_while_nobody_is_connected_is_dropped_rather_than_queued() { + let mut relay = Relay::start("nobody-home"); + + // Sent into a relay nothing is on. It is not held for whoever connects + // next: what crosses this layer is re-sent when it changes, so a held + // copy is a stale copy — and a queue that fills is what would + // eventually stall the session filling it. + relay + .down + .send(Payload::new("identity", "stale")) + .await + .unwrap(); + + let (reader, _writer) = relay.workload().await; + let mut lines = reader.lines(); + relay + .down + .send(Payload::new("identity", "current")) + .await + .unwrap(); + + let line = lines.next_line().await.unwrap().unwrap(); + let seen: Payload = from_line(&line).unwrap(); + assert_eq!( + seen.body, "current", + "a stale envelope was delivered on connect" + ); + } + + #[tokio::test] + async fn a_frame_that_never_ends_costs_the_connection_and_not_the_guest() { + use tokio::io::AsyncReadExt; + + let mut relay = Relay::start("unbounded"); + let (mut reader, mut writer) = relay.dial().await.into_split(); + + // Well past the cap, and not a newline in it. Without a limit the + // buffer holding this grows in the one process the kernel has been + // told not to kill, so what it takes comes out of everything else. + let flood = vec![b'a'; LONGEST_ENVELOPE + 4096]; + // The write fails once the relay drops the connection, which is the + // outcome under test rather than a problem with it. + let _ = writer.write_all(&flood).await; + + // The relay closes on us rather than keeping the buffer. Bounded, + // because the failure being guarded against is a relay that reads for + // as long as the workload writes. + let mut unread = Vec::new(); + let closed = tokio::time::timeout( + std::time::Duration::from_secs(5), + reader.read_to_end(&mut unread), + ) + .await + .expect("the relay is still assembling a frame that never ends"); + assert_eq!( + closed.unwrap(), + 0, + "the relay answered a frame it should have refused" + ); + + // And it is there for whoever connects next. + let (reader, mut writer) = + tokio::time::timeout(std::time::Duration::from_secs(5), relay.workload()) + .await + .expect("the relay never came back for the next workload"); + let mut lines = reader.lines(); + writer + .write_all( + to_line(&Payload::new("identity", "after the flood")) + .unwrap() + .as_bytes(), + ) + .await + .unwrap(); + assert_eq!(relay.up.recv().await.unwrap().body, "after the flood"); + + relay + .down + .send(Payload::new("identity", "down")) + .await + .unwrap(); + let line = lines.next_line().await.unwrap().unwrap(); + assert_eq!(from_line::(&line).unwrap().body, "down"); } } diff --git a/apps/nesinit/src/session.rs b/apps/nesinit/src/session.rs index 0bb302cb..78e0e28d 100644 --- a/apps/nesinit/src/session.rs +++ b/apps/nesinit/src/session.rs @@ -183,7 +183,7 @@ where } } } - HostToGuest::Payload { payload: envelope } => hand_over(payload, envelope).await, + HostToGuest::Payload { payload: envelope } => hand_over(payload, envelope), HostToGuest::Stop => workload.signal_stop(), HostToGuest::Shutdown => return Ok(Outcome::Shutdown), } @@ -199,13 +199,24 @@ enum Event { /// Hand an envelope to the relay, and treat a relay that is not there as the /// caller's problem rather than a failure of this session. -async fn hand_over(ports: &mut Ports, envelope: Payload) { - tracing::debug!(envelope = %envelope.summary(), "handing an envelope over"); - if ports.to_workload.send(envelope).await.is_err() { - // The workload is not on the relay. Dropped rather than queued: what - // crosses here is re-sent when it changes, so a held copy is a stale - // copy. - tracing::warn!("dropped an envelope: nothing is on the relay"); +/// +/// It never waits. This loop also carries stop, shutdown and the workload's +/// exit, and none of those may be held up by a workload that is slow to read +/// its own mail — or by one that never connected at all. What crosses this +/// layer is re-sent when it changes, so a dropped copy costs less than a +/// stalled session. +fn hand_over(ports: &mut Ports, envelope: Payload) { + use tokio::sync::mpsc::error::TrySendError; + + let summary = envelope.summary(); + match ports.to_workload.try_send(envelope) { + Ok(()) => tracing::debug!(envelope = %summary, "handed an envelope over"), + Err(TrySendError::Full(_)) => { + tracing::warn!(envelope = %summary, "dropped an envelope: the relay is behind") + } + Err(TrySendError::Closed(_)) => { + tracing::warn!(envelope = %summary, "dropped an envelope: the relay is gone") + } } } @@ -655,4 +666,55 @@ mod tests { assert_eq!(session.await.unwrap(), Outcome::Shutdown); } + + #[tokio::test] + async fn a_relay_that_is_not_draining_does_not_stall_the_session() { + // Bounded, because the failure is a session that stops rather than one + // that answers wrongly. + tokio::time::timeout(std::time::Duration::from_secs(5), async { + a_backed_up_relay().await + }) + .await + .expect("the session stalled on the relay"); + } + + async fn a_backed_up_relay() { + let (guest, host) = tokio::io::duplex(4096); + let mut caller = Caller::new(host); + let (down_tx, down_rx) = mpsc::channel(1); + let (_up_tx, up_rx) = mpsc::channel::(1); + // Held and never read: a workload that is slow to read its own mail, + // or one that connected and stopped. + let _backed_up = down_rx; + + let session = tokio::spawn(async move { + let mut ports = Ports { + to_workload: down_tx, + from_workload: up_rx, + }; + let mut workload = Double::exits_when_stopped(Exit::code(0)); + run(guest, &mut workload, &mut ports).await.unwrap() + }); + + assert!(matches!(caller.expect().await, GuestToHost::Ready { .. })); + for _ in 0..8 { + caller + .say(&HostToGuest::Payload { + payload: Payload::new("identity", "backlog"), + }) + .await; + } + + // The lifecycle layer still moves: stop, shutdown and an exit are on + // this loop too, and none of them may wait on the relay. + caller + .say(&HostToGuest::Boot { + descriptor: Box::new(descriptor()), + }) + .await; + caller.expect_started().await; + caller.say(&HostToGuest::Shutdown).await; + + assert_eq!(session.await.unwrap(), Outcome::Shutdown); + } } diff --git a/apps/nesinit/src/workload.rs b/apps/nesinit/src/workload.rs index ef5d00f6..d9f7c3b9 100644 --- a/apps/nesinit/src/workload.rs +++ b/apps/nesinit/src/workload.rs @@ -192,11 +192,15 @@ impl Workload for Process { /// channel, so the guest still learns nothing about the filesystem it is being /// handed a piece of. fn mount_share(share: &Mount) -> Result<(), Failure> { + // Checked before anything is created: a descriptor this component cannot + // act on should leave no directory behind to confuse whoever reads the + // failure. + let (source, target, flags) = options(share)?; + // The mount point may not exist yet: a share can land anywhere the // descriptor names, including a directory no image created. std::fs::create_dir_all(&share.at).map_err(|error| failed(share, error))?; - let (source, target, flags) = options(share); // SAFETY: mount takes two paths, a filesystem name and a flag word, all // of which outlive the call, and no options string. let mounted = unsafe { @@ -220,7 +224,7 @@ const FSTYPE: &std::ffi::CStr = c"virtiofs"; /// What the mount call is given, split out because this is the part worth /// asserting: mounting itself needs privileges a test does not have. -fn options(share: &Mount) -> (CString, CString, libc::c_ulong) { +fn options(share: &Mount) -> Result<(CString, CString, libc::c_ulong), Failure> { // nosuid and nodev on every share, whether or not it is writable. A share // is data handed to the guest; a setuid binary or a device node appearing // in one is not something a workload should be able to use, and no @@ -229,11 +233,23 @@ fn options(share: &Mount) -> (CString, CString, libc::c_ulong) { if share.ro { flags |= libc::MS_RDONLY; } - // Interior nul bytes are the caller's mistake; a path with one cannot be - // mounted under any flags. - let source = CString::new(share.tag.as_str()).unwrap_or_default(); - let target = CString::new(share.at.as_str()).unwrap_or_default(); - (source, target, flags) + // A nul byte inside a tag or a path is a descriptor that cannot be carried + // out under any flags. Refused by name rather than silently emptied: an + // empty source turns up later as a mount failure about something else + // entirely, which is the wrong thing to go and look at. + let source = CString::new(share.tag.as_str()).map_err(|_| { + Failure::new(format!( + "the share tag contains a nul byte: {:?}", + share.tag + )) + })?; + let target = CString::new(share.at.as_str()).map_err(|_| { + Failure::new(format!( + "the mount point contains a nul byte: {:?}", + share.at + )) + })?; + Ok((source, target, flags)) } /// A failure names the path, which is what makes it actionable: a permission @@ -257,7 +273,7 @@ mod tests { #[test] fn a_writable_share_is_still_mounted_without_devices_or_setuid() { - let (source, target, flags) = options(&share(false)); + let (source, target, flags) = options(&share(false)).unwrap(); assert_eq!( source.to_str().unwrap(), "user", @@ -271,10 +287,29 @@ mod tests { #[test] fn a_read_only_share_is_mounted_read_only() { - let (_, _, flags) = options(&share(true)); + let (_, _, flags) = options(&share(true)).unwrap(); assert_eq!(flags & libc::MS_RDONLY, libc::MS_RDONLY); } + #[test] + fn a_descriptor_with_a_nul_byte_in_it_is_refused_by_name() { + let tagged = Mount { + tag: "us\0er".into(), + at: "/mnt/user".into(), + ro: false, + }; + let failure = options(&tagged).expect_err("an empty source would have been mounted"); + assert!(failure.reason.contains("tag"), "{}", failure.reason); + + let placed = Mount { + tag: "user".into(), + at: "/mnt/us\0er".into(), + ro: false, + }; + let failure = options(&placed).expect_err("an empty target would have been mounted"); + assert!(failure.reason.contains("mount point"), "{}", failure.reason); + } + #[test] fn a_failure_names_the_path_it_happened_on() { let failure = failed(&share(false), io::Error::from_raw_os_error(libc::EACCES));