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},