From 3d24a8e130171f737add41a9122feda4f6a8b2f4 Mon Sep 17 00:00:00 2001 From: Wanjohi Date: Sun, 6 Sep 2026 13:55:35 +0300 Subject: [PATCH 1/4] feat(nesinit): carry the session's address out of the guest Whatever serves media in a box knows how it can be reached, and the person who needs to know is not in the box. Standard output here is a log file inside a VM, so the control channel is the delivery path rather than a convenience -- which makes this init's job and not a detail of whichever component happens to bind the port. The socket it reads was already documented as being read this way; nothing read it. Polled rather than read once, because an address is not a value but the best answer so far. An endpoint discovers more ways to reach it after it binds, so the first answer is the one that works on a local network and fails from anywhere else. Only a changed answer is forwarded. It dials rather than listens, which is the opposite of the relay next door and deliberate: there the guest listens because the workload starts later, and here the server is the long-lived one. Dialling also makes a server that has not bound yet something to retry rather than something to wait for without knowing whether it is coming. The address itself is never logged. It is a capability to reach the session, and a log inside the guest is the one place it has no reason to be. A carrier that stops does not end a session: whatever was already reported is still correct, and the workload's exit still has to be. --- apps/nesinit/src/lib.rs | 1 + apps/nesinit/src/main.rs | 18 ++- apps/nesinit/src/session.rs | 163 +++++++++++++++++++++++--- apps/nesinit/src/ticket.rs | 224 ++++++++++++++++++++++++++++++++++++ 4 files changed, 391 insertions(+), 15 deletions(-) create mode 100644 apps/nesinit/src/ticket.rs diff --git a/apps/nesinit/src/lib.rs b/apps/nesinit/src/lib.rs index 8ce6a6fc..1f2b3588 100644 --- a/apps/nesinit/src/lib.rs +++ b/apps/nesinit/src/lib.rs @@ -12,4 +12,5 @@ pub mod payload; pub mod reap; pub mod session; pub mod shutdown; +pub mod ticket; pub mod workload; diff --git a/apps/nesinit/src/main.rs b/apps/nesinit/src/main.rs index 2952761f..aa157c6a 100644 --- a/apps/nesinit/src/main.rs +++ b/apps/nesinit/src/main.rs @@ -4,13 +4,14 @@ // the workload the channel describes, and turn the end of either into an // ordered shutdown. -use std::path::Path; +use std::path::{Path, PathBuf}; use std::time::Duration; use nesinit::payload::{self, Ports}; use nesinit::reap::{self, Waiters}; use nesinit::session::{self, Outcome}; use nesinit::shutdown::{self, Machine}; +use nesinit::ticket; use nesinit::workload::{Process, Workload}; use nesprotocol::lifecycle::CONTROL_PORT; use tokio::signal::unix::{SignalKind, signal}; @@ -25,6 +26,13 @@ const GRACE: Duration = Duration::from_secs(10); /// deep queue holds stale copies of it rather than protecting anything. const RELAY_DEPTH: usize = 8; +/// How many addresses may be waiting to be forwarded. +/// +/// Two, because only the newest one matters: an address is superseded by the +/// next one rather than added to, so a deeper queue holds stale copies of it +/// and delays the one that is current. +const ADDRESS_DEPTH: usize = 2; + fn main() -> anyhow::Result<()> { tracing_subscriber::fmt() .with_env_filter( @@ -97,8 +105,14 @@ async fn guest(waiters: &Waiters, workload: &mut Process) -> anyhow::Result outcome?, + outcome = session::run(channel, workload, &mut ports, &mut found_rx) => outcome?, signal = asked_to_stop() => { signal?; tracing::info!("asked to stop"); diff --git a/apps/nesinit/src/session.rs b/apps/nesinit/src/session.rs index 78e0e28d..8490f499 100644 --- a/apps/nesinit/src/session.rs +++ b/apps/nesinit/src/session.rs @@ -14,6 +14,7 @@ use tokio::io::{AsyncBufReadExt, AsyncRead, AsyncWrite, AsyncWriteExt, BufReader use crate::payload::Ports; use crate::workload::{Exited, Failure, Workload}; +use tokio::sync::mpsc::Receiver; /// How a session ended. #[derive(Debug, Clone, PartialEq, Eq)] @@ -39,12 +40,13 @@ pub async fn run( channel: C, workload: &mut W, payload: &mut Ports, + addresses: &mut Receiver, ) -> std::io::Result where C: AsyncRead + AsyncWrite, W: Workload, { - match converse(channel, workload, payload).await { + match converse(channel, workload, payload, addresses).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 @@ -70,6 +72,7 @@ async fn converse( channel: C, workload: &mut W, payload: &mut Ports, + addresses: &mut Receiver, ) -> std::io::Result where C: AsyncRead + AsyncWrite, @@ -91,6 +94,7 @@ where let mut running: Option = None; let mut relay_open = true; + let mut carrier_open = true; loop { let event = match running.as_mut() { @@ -98,10 +102,12 @@ where ended = exited => Event::Ended(ended?), line = lines.next_line() => Event::Line(line?), up = payload.from_workload.recv(), if relay_open => Event::FromWorkload(up), + found = addresses.recv(), if carrier_open => Event::Address(found), }, None => tokio::select! { line = lines.next_line() => Event::Line(line?), up = payload.from_workload.recv(), if relay_open => Event::FromWorkload(up), + found = addresses.recv(), if carrier_open => Event::Address(found), }, }; @@ -115,6 +121,20 @@ where send(&mut writer, &GuestToHost::Payload { payload }).await?; continue; } + Event::Address(Some(ticket)) => { + // Sent whenever a better one is found, not only the first time: + // a caller that keeps the first address it is given works on a + // local network and fails from anywhere else. + send(&mut writer, &GuestToHost::Ticket { ticket }).await?; + continue; + } + Event::Address(None) => { + // Nothing will look for an address again. Not an ending: the + // session has whatever it was already told, and the workload + // still has to be stopped and its exit reported. + carrier_open = false; + 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. @@ -195,6 +215,7 @@ enum Event { Line(Option), Ended(Exit), FromWorkload(Option), + Address(Option), } /// Hand an envelope to the relay, and treat a relay that is not there as the @@ -276,6 +297,17 @@ mod tests { ) } + /// A carrier that never finds an address, for the tests that are not + /// about one. Held open rather than closed: a closed channel is itself a + /// case, and it is tested on purpose below. + fn nowhere() -> mpsc::Receiver { + let (tx, rx) = mpsc::channel(1); + // Kept alive for the process, so `recv` pends rather than resolving + // `None` and taking a branch these tests are not exercising. + Box::leak(Box::new(tx)); + rx + } + /// The other end of the channel, as a caller would drive it. struct Caller { lines: tokio::io::Lines>, @@ -322,7 +354,9 @@ mod tests { 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, &mut ports).await.unwrap(); + let outcome = run(guest, &mut workload, &mut ports, &mut nowhere()) + .await + .unwrap(); (outcome, workload) }); @@ -339,6 +373,87 @@ mod tests { assert_eq!(outcome, Outcome::Shutdown); } + /// The address goes up the channel as it is found. This is the only way + /// out: standard output here is a log file inside a VM and the person who + /// needs the address is outside it. + #[tokio::test] + async fn an_address_that_is_found_is_reported_to_the_caller() { + let (guest, host) = tokio::io::duplex(4096); + let mut caller = Caller::new(host); + let (found_tx, mut found_rx) = mpsc::channel(4); + + 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, &mut ports, &mut found_rx) + .await + .unwrap() + }); + + assert_eq!( + caller.expect().await, + GuestToHost::Ready { + protocol_version: 2 + } + ); + + found_tx + .send("nestri:local-only".to_string()) + .await + .unwrap(); + assert_eq!( + caller.expect().await, + GuestToHost::Ticket { + ticket: "nestri:local-only".into() + } + ); + + // And a better one replaces it rather than being the caller's problem + // to have missed. + found_tx + .send("nestri:with-relays".to_string()) + .await + .unwrap(); + assert_eq!( + caller.expect().await, + GuestToHost::Ticket { + ticket: "nestri:with-relays".into() + } + ); + + caller.say(&HostToGuest::Shutdown).await; + assert_eq!(session.await.unwrap(), Outcome::Shutdown); + } + + /// Nothing looking for an address any more is not an ending. The session + /// keeps whatever it was already told and still has to report an exit. + #[tokio::test] + async fn a_carrier_that_stops_does_not_end_the_session() { + let (guest, host) = tokio::io::duplex(4096); + let mut caller = Caller::new(host); + let (found_tx, mut found_rx) = mpsc::channel(4); + + 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, &mut ports, &mut found_rx) + .await + .unwrap() + }); + + assert_eq!( + caller.expect().await, + GuestToHost::Ready { + protocol_version: 2 + } + ); + drop(found_tx); + + // Still answering, which is the whole assertion. + caller.say(&HostToGuest::Shutdown).await; + assert_eq!(session.await.unwrap(), Outcome::Shutdown); + } + #[tokio::test] async fn the_descriptor_mounts_and_starts_what_it_names() { let (guest, host) = tokio::io::duplex(4096); @@ -347,7 +462,9 @@ mod tests { 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, &mut ports).await.unwrap(); + let outcome = run(guest, &mut workload, &mut ports, &mut nowhere()) + .await + .unwrap(); (outcome, workload) }); @@ -373,7 +490,9 @@ mod tests { 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, &mut ports).await.unwrap(); + let outcome = run(guest, &mut workload, &mut ports, &mut nowhere()) + .await + .unwrap(); (outcome, workload) }); @@ -409,7 +528,9 @@ mod tests { 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, &mut ports).await.unwrap() + run(guest, &mut workload, &mut ports, &mut nowhere()) + .await + .unwrap() }); assert!(matches!(caller.expect().await, GuestToHost::Ready { .. })); @@ -440,7 +561,9 @@ mod tests { 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, &mut ports).await.unwrap(); + let outcome = run(guest, &mut workload, &mut ports, &mut nowhere()) + .await + .unwrap(); (outcome, workload) }); @@ -466,7 +589,9 @@ mod tests { 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, &mut ports).await.unwrap(); + let outcome = run(guest, &mut workload, &mut ports, &mut nowhere()) + .await + .unwrap(); (outcome, workload) }); @@ -501,7 +626,9 @@ mod tests { 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, &mut ports).await.unwrap() + run(guest, &mut workload, &mut ports, &mut nowhere()) + .await + .unwrap() }); assert!(matches!(caller.expect().await, GuestToHost::Ready { .. })); @@ -524,7 +651,9 @@ mod tests { 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, &mut ports).await.unwrap(); + let outcome = run(guest, &mut workload, &mut ports, &mut nowhere()) + .await + .unwrap(); (outcome, workload) }); @@ -556,7 +685,9 @@ mod tests { 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(); + let outcome = run(guest, &mut workload, &mut ports, &mut nowhere()) + .await + .unwrap(); (outcome, workload) }); @@ -598,7 +729,9 @@ mod tests { from_workload: up_rx, }; let mut workload = Double::exits_when_stopped(Exit::code(0)); - run(guest, &mut workload, &mut ports).await.unwrap() + run(guest, &mut workload, &mut ports, &mut nowhere()) + .await + .unwrap() }); let mut to_relay = down_rx; @@ -647,7 +780,9 @@ mod tests { from_workload: up_rx, }; let mut workload = Double::exits_when_stopped(Exit::code(0)); - run(guest, &mut workload, &mut ports).await.unwrap() + run(guest, &mut workload, &mut ports, &mut nowhere()) + .await + .unwrap() }); assert!(matches!(caller.expect().await, GuestToHost::Ready { .. })); @@ -693,7 +828,9 @@ mod tests { from_workload: up_rx, }; let mut workload = Double::exits_when_stopped(Exit::code(0)); - run(guest, &mut workload, &mut ports).await.unwrap() + run(guest, &mut workload, &mut ports, &mut nowhere()) + .await + .unwrap() }); assert!(matches!(caller.expect().await, GuestToHost::Ready { .. })); diff --git a/apps/nesinit/src/ticket.rs b/apps/nesinit/src/ticket.rs new file mode 100644 index 00000000..2d98fba5 --- /dev/null +++ b/apps/nesinit/src/ticket.rs @@ -0,0 +1,224 @@ +// Carrying the address a client needs from inside the guest to outside it. +// +// Whatever serves media in the guest knows how it can be reached, and the +// person who needs to know is not in the guest. Standard output here is a log +// file inside a VM, so the control channel is the delivery path rather than a +// convenience — which is why this is init's job and not a detail of whichever +// component happens to bind the port. +// +// # Why it is polled rather than read once +// +// An address is not a value, it is the best answer so far. An endpoint +// discovers more ways to reach it after it binds — a local one immediately, a +// relayed or hole-punched one some seconds later — so the first answer is the +// one that works on a local network and fails from anywhere else. Re-reading +// and forwarding only what changed keeps that from being decided by whoever +// asked first. +// +// # Why it dials rather than listens +// +// The opposite of the payload relay next door, and deliberately: there the +// guest listens because the workload starts later, and here the server is the +// long-lived one. Dialling also means a server that has not bound yet is an +// error this retries, rather than a connection that has to be waited for +// without knowing whether it is coming. + +use std::io; +use std::path::{Path, PathBuf}; +use std::time::Duration; + +use tokio::io::{AsyncBufReadExt, AsyncReadExt, BufReader}; +use tokio::net::UnixStream; +use tokio::sync::mpsc::Sender; + +/// Where the address is read from. +pub const SOCKET: &str = "/tmp/nestri-ticket.sock"; + +/// How often to look for a better address. +/// +/// Short, because the window this closes is the first few seconds of a session +/// and a player waiting to connect is waiting on exactly this. It costs one +/// connection to a unix socket per interval. +const EVERY: Duration = Duration::from_secs(2); + +/// The longest address this will read. +/// +/// One line of text. A cap rather than a preference: the process on the other +/// end can write without ever sending a newline, and the process holding the +/// buffer is the one the kernel has been told not to kill. +const LONGEST: u64 = 8 * 1024; + +/// Forward every new address for as long as the session lasts. +/// +/// Never returns on its own. A server that is not there yet, or has gone away, +/// is retried at the next interval — there is nothing here worth ending a +/// running session over, and an address that stops being re-offered does not +/// stop being correct. +pub async fn carry(path: PathBuf, out: Sender) { + let mut sent: Option = None; + loop { + match read(&path).await { + Ok(current) if Some(¤t) != sent.as_ref() => { + // The address itself is not logged. It is a capability to reach + // this session, and a log inside the guest is the one place it + // has no reason to be. + tracing::info!( + first = sent.is_none(), + "forwarding an address for this session" + ); + if out.send(current.clone()).await.is_err() { + // The session is over. Nothing else reads this. + return; + } + sent = Some(current); + } + Ok(_) => {} + Err(error) => { + // Expected until whatever serves the address has bound, so it + // is not a warning the first several times. It stays at this + // level afterwards too: a session that already has an address + // is not harmed by failing to look for a better one. + tracing::debug!(%error, "no address available yet"); + } + } + tokio::time::sleep(EVERY).await; + } +} + +/// One line from the socket, which is the whole protocol. +async fn read(path: &Path) -> io::Result { + let stream = UnixStream::connect(path).await?; + let mut line = String::new(); + BufReader::new(stream.take(LONGEST)) + .read_line(&mut line) + .await?; + let line = line.trim().to_string(); + if line.is_empty() { + return Err(io::Error::new( + io::ErrorKind::UnexpectedEof, + "the socket answered with nothing", + )); + } + Ok(line) +} + +#[cfg(test)] +mod tests { + use super::*; + use tokio::io::AsyncWriteExt; + use tokio::net::UnixListener; + use tokio::sync::mpsc; + + /// Serve a sequence of addresses, one per connection, the way a real one + /// does: it answers every dial with what it currently knows. + fn serve(path: PathBuf, answers: Vec>) { + let listener = UnixListener::bind(&path).unwrap(); + tokio::spawn(async move { + let mut answers = answers.into_iter().cycle(); + loop { + let Ok((mut stream, _)) = listener.accept().await else { + return; + }; + if let Some(answer) = answers.next().flatten() { + let _ = stream.write_all(format!("{answer}\n").as_bytes()).await; + } + } + }); + } + + fn scratch(name: &str) -> PathBuf { + let dir = + std::env::temp_dir().join(format!("nesinit-ticket-{name}-{}", std::process::id())); + let _ = std::fs::remove_dir_all(&dir); + std::fs::create_dir_all(&dir).unwrap(); + dir.join("ticket.sock") + } + + #[tokio::test] + async fn an_address_reaches_the_channel() { + let path = scratch("one"); + serve(path.clone(), vec![Some("nestri:abc".into())]); + + let (tx, mut rx) = mpsc::channel(4); + tokio::spawn(carry(path.clone(), tx)); + + let first = tokio::time::timeout(Duration::from_secs(10), rx.recv()) + .await + .expect("no address arrived") + .expect("the channel closed"); + assert_eq!(first, "nestri:abc"); + + // And it is not sent again. An address that has not changed is the same + // address, and re-sending it is a write per interval for nothing. + assert!( + tokio::time::timeout(EVERY * 3, rx.recv()).await.is_err(), + "an unchanged address was forwarded again" + ); + let _ = std::fs::remove_file(&path); + } + + /// The case that decides whether this works from anywhere but a local + /// network: a better address arrives after the first one has been sent. + #[tokio::test] + async fn a_better_address_replaces_the_one_before_it() { + let path = scratch("better"); + serve( + path.clone(), + vec![ + Some("nestri:local-only".into()), + Some("nestri:with-relays".into()), + ], + ); + + let (tx, mut rx) = mpsc::channel(4); + tokio::spawn(carry(path.clone(), tx)); + + let mut seen = Vec::new(); + while seen.len() < 2 { + let next = tokio::time::timeout(Duration::from_secs(10), rx.recv()) + .await + .expect("the second address never arrived") + .expect("the channel closed"); + seen.push(next); + } + assert_eq!(seen, ["nestri:local-only", "nestri:with-relays"]); + let _ = std::fs::remove_file(&path); + } + + /// Nothing serving the socket yet is the ordinary case at boot, not a + /// failure: this starts before whatever binds it. + #[tokio::test] + async fn a_socket_that_is_not_there_yet_is_waited_out_rather_than_failed() { + let path = scratch("late"); + let (tx, mut rx) = mpsc::channel(4); + tokio::spawn(carry(path.clone(), tx)); + + tokio::time::sleep(EVERY * 2).await; + serve(path.clone(), vec![Some("nestri:late".into())]); + + let first = tokio::time::timeout(Duration::from_secs(10), rx.recv()) + .await + .expect("an address that arrived late was never picked up") + .expect("the channel closed"); + assert_eq!(first, "nestri:late"); + let _ = std::fs::remove_file(&path); + } + + /// An answer with nothing in it is not an address. Forwarding one would + /// publish an empty string as somewhere to connect. + #[tokio::test] + async fn an_empty_answer_is_not_an_address() { + let path = scratch("empty"); + serve(path.clone(), vec![None, Some("nestri:real".into())]); + + let (tx, mut rx) = mpsc::channel(4); + tokio::spawn(carry(path.clone(), tx)); + + let first = tokio::time::timeout(Duration::from_secs(10), rx.recv()) + .await + .expect("no address arrived") + .expect("the channel closed"); + assert_eq!(first, "nestri:real"); + let _ = std::fs::remove_file(&path); + } +} From 18864b97f0c2bd0acbef297734ab447c4f502f83 Mon Sep 17 00:00:00 2001 From: KAAL1 Date: Sun, 6 Sep 2026 14:23:56 +0300 Subject: [PATCH 2/4] fix(nesinit): mount what the guest needs before anything asks for it The root arrives read-only and this process is PID 1, so until it mounts them there is no /proc and nowhere in the filesystem to put a socket. Nothing else in the guest is an init system, so nothing else was going to. The symptom was three failures that look unrelated and share one cause. On a real box the payload relay could not bind, with EROFS; whatever serves the session's address could not bind either, the same way; and this process could not make itself ineligible for the OOM killer, because /proc was not there to write to. What the caller saw was a workload that ran and published nothing, which is true and says nothing about why. /proc is mounted first and unconditionally: finding out what an image already mounted requires it, and it is therefore the one entry that cannot be checked that way itself. Everything after it is skipped when it is already present, so an image that does this properly is not mounted over. Failures warn rather than abort. Refusing to boot would replace a session that fails with a reason by a guest that never dialled out at all, and the second is harder to diagnose from the outside. The relay's directory is named by the module that owns the socket rather than spelled again here, with a test tying the two together: a rename that reached one and not the other would put the relay back exactly as it was. --- apps/nesinit/src/filesystems.rs | 215 ++++++++++++++++++++++++++++++++ apps/nesinit/src/lib.rs | 1 + apps/nesinit/src/main.rs | 5 + apps/nesinit/src/payload.rs | 8 ++ 4 files changed, 229 insertions(+) create mode 100644 apps/nesinit/src/filesystems.rs diff --git a/apps/nesinit/src/filesystems.rs b/apps/nesinit/src/filesystems.rs new file mode 100644 index 00000000..8c2768a8 --- /dev/null +++ b/apps/nesinit/src/filesystems.rs @@ -0,0 +1,215 @@ +// The filesystems PID 1 has to establish before anything asks for them. +// +// The root arrives read-only — the host attaches it that way and nothing in +// the guest may write to it — and there is no init system behind this process +// to make up the difference. So a guest gets `/proc`, and it gets somewhere to +// put a socket, only if this mounts them. +// +// Without that the failure is not an error anyone sees. Everything that needs +// a writable path fails one layer down, separately, as `EROFS` on a socket: +// the payload relay never binds, whatever serves the session's address never +// binds either, and the session is reported as a workload that ran and +// published nothing. Three unrelated-looking symptoms, one missing mount. + +use std::ffi::CString; +use std::path::Path; + +/// A filesystem this process mounts, and why it has to exist. +struct Early { + /// What appears in `/proc/mounts` as the source. Conventionally the type. + source: &'static str, + target: &'static str, + fstype: &'static str, + flags: libc::c_ulong, + /// Mount options, or empty for none. + data: &'static str, + /// Said when it could not be mounted, in terms of what stops working. + cost: &'static str, +} + +/// `nosuid` and `nodev` on everything: none of these carry an image's files, +/// so a device node or a setuid bit appearing in one did not come from us. +const NOSUID_NODEV: libc::c_ulong = libc::MS_NOSUID | libc::MS_NODEV; + +const EARLY: &[Early] = &[ + Early { + source: "proc", + target: "/proc", + fstype: "proc", + flags: NOSUID_NODEV | libc::MS_NOEXEC, + data: "", + cost: "this process cannot make itself ineligible for the OOM killer, \ + and nothing in the guest can read its own state", + }, + Early { + source: "sysfs", + target: "/sys", + fstype: "sysfs", + flags: NOSUID_NODEV | libc::MS_NOEXEC, + data: "", + cost: "a workload that looks up a device finds nothing", + }, + // Writable, and the reason any of this is here. Both sockets in this + // component live on a tmpfs because the root they would otherwise sit on + // is read-only. + Early { + source: "tmpfs", + target: "/tmp", + fstype: "tmpfs", + flags: NOSUID_NODEV, + // The sticky bit, because the workload does not run as this process + // does and what it binds here is its own. + data: "mode=1777", + cost: "whatever serves this session's address cannot bind its socket, \ + so the session never gets one", + }, + Early { + source: "tmpfs", + target: crate::payload::DIRECTORY, + fstype: "tmpfs", + flags: NOSUID_NODEV | libc::MS_NOEXEC, + // Only this process and the workload it starts, and they are the only + // two that ever have business here. + data: "mode=0770", + cost: "the payload relay cannot bind, so nothing reaches the workload \ + over the channel", + }, +]; + +/// Mount what the rest of this component assumes is already there. +/// +/// Best effort, one line per failure. Refusing to boot over any of these would +/// replace a session that fails with a reason with a guest that never dialled +/// out at all, and the second is strictly harder to diagnose from the host. +pub fn establish() { + // `/proc` first and unconditionally: it is the only way to find out what is + // already mounted, so everything after it can be skipped when an image has + // done it already, and it cannot itself be checked that way. + mount(&EARLY[0]); + + let existing = std::fs::read_to_string("/proc/self/mountinfo").unwrap_or_default(); + for early in &EARLY[1..] { + if mounted_at(&existing, early.target) { + tracing::debug!(target = early.target, "already mounted by the image"); + continue; + } + mount(early); + } +} + +/// Whether `mountinfo` already has a mount at this exact path. +/// +/// The mount point is the fifth field and it is the one that has to match: +/// a prefix test would read `/tmpfoo` as `/tmp`, and a substring test would +/// find the path in the options of something else entirely. +fn mounted_at(mountinfo: &str, target: &str) -> bool { + mountinfo + .lines() + .filter_map(|line| line.split_whitespace().nth(4)) + .any(|point| point == target) +} + +fn mount(early: &Early) { + // A mount point that is not in the image cannot be created on a read-only + // root, so this is allowed to fail and the mount below reports it. + if !Path::new(early.target).exists() { + let _ = std::fs::create_dir_all(early.target); + } + + let (Ok(source), Ok(target), Ok(fstype), Ok(data)) = ( + CString::new(early.source), + CString::new(early.target), + CString::new(early.fstype), + CString::new(early.data), + ) else { + // Every one of these is a literal in this file, so this is + // unreachable rather than a case to handle. + tracing::error!(target = early.target, "a mount table entry has a nul byte"); + return; + }; + let data = if early.data.is_empty() { + std::ptr::null() + } else { + data.as_ptr().cast() + }; + + // SAFETY: four pointers that outlive the call, and a flag word. + let mounted = unsafe { + libc::mount( + source.as_ptr(), + target.as_ptr(), + fstype.as_ptr(), + early.flags, + data, + ) + }; + if mounted != 0 { + tracing::warn!( + target = early.target, + error = %std::io::Error::last_os_error(), + cost = early.cost, + "could not mount" + ); + } +} + +#[cfg(test)] +mod tests { + use super::*; + + /// The whole point of reading `mountinfo` is to not mount twice over + /// something an image already did. + #[test] + fn a_mount_point_that_is_present_is_recognised() { + let info = "\ +23 1 0:5 / /proc rw,nosuid,nodev,noexec - proc proc rw +24 1 0:6 / /tmp rw,nosuid,nodev - tmpfs tmpfs rw,mode=1777"; + assert!(mounted_at(info, "/proc")); + assert!(mounted_at(info, "/tmp")); + } + + /// A prefix is not a mount point, and neither is a path that only appears + /// in another line's options. + #[test] + fn something_else_is_not_mistaken_for_a_mount_point() { + let info = "\ +23 1 0:5 / /tmpfoo rw - tmpfs tmpfs rw +24 1 0:6 / /var rw - ext4 /dev/vda rw,journal_path=/nestri"; + assert!(!mounted_at(info, "/tmp")); + assert!(!mounted_at(info, "/nestri")); + } + + /// Every entry has to be nul-free, because `establish` treats a nul as + /// unreachable rather than handling it. + #[test] + fn the_mount_table_can_be_carried_out() { + for early in EARLY { + assert!(CString::new(early.source).is_ok(), "{}", early.target); + assert!(CString::new(early.target).is_ok(), "{}", early.target); + assert!(CString::new(early.fstype).is_ok(), "{}", early.target); + assert!(CString::new(early.data).is_ok(), "{}", early.target); + assert!(!early.cost.is_empty(), "{} has no cost", early.target); + } + } + + /// `/proc` is mounted before `mountinfo` is read, so it has to be first. + #[test] + fn proc_is_the_first_entry() { + assert_eq!(EARLY[0].target, "/proc"); + } + + /// The relay's directory is the one this cannot hardcode: it belongs to + /// `payload`, and a rename there that missed this file would take the + /// relay down again in exactly the way this exists to prevent. + #[test] + fn the_relay_directory_is_the_one_the_relay_uses() { + let entry = EARLY + .iter() + .find(|e| e.target == crate::payload::DIRECTORY) + .expect("the relay's directory is mounted"); + assert!( + crate::payload::SOCKET.starts_with(entry.target), + "the relay's socket is not under the directory that is mounted for it" + ); + } +} diff --git a/apps/nesinit/src/lib.rs b/apps/nesinit/src/lib.rs index 1f2b3588..32df7edd 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 filesystems; pub mod payload; pub mod reap; pub mod session; diff --git a/apps/nesinit/src/main.rs b/apps/nesinit/src/main.rs index aa157c6a..239be9b7 100644 --- a/apps/nesinit/src/main.rs +++ b/apps/nesinit/src/main.rs @@ -40,6 +40,11 @@ fn main() -> anyhow::Result<()> { ) .init(); + // Before everything, including the two below: the root is read-only and + // nothing else in this guest is an init system, so until this runs there + // is no `/proc` to score this process in and nowhere to put a socket. + nesinit::filesystems::establish(); + // Both before anything is started, so nothing can be orphaned or scored // in the window where neither is true yet. if let Err(error) = reap::become_subreaper() { diff --git a/apps/nesinit/src/payload.rs b/apps/nesinit/src/payload.rs index 96b3a4ad..944bcbca 100644 --- a/apps/nesinit/src/payload.rs +++ b/apps/nesinit/src/payload.rs @@ -16,6 +16,14 @@ use tokio::io::{AsyncBufRead, AsyncBufReadExt, AsyncWriteExt, BufReader}; use tokio::net::{UnixListener, UnixStream}; use tokio::sync::mpsc::{Receiver, Sender}; +/// The directory the relay's socket lives in. +/// +/// Named separately because it is mounted before it is used: the guest's root +/// is read-only, so this is a tmpfs that `filesystems` puts there, and a +/// rename here that did not reach the mount table would take the relay down +/// with an `EROFS` that looks like nothing to do with a path. +pub const DIRECTORY: &str = "/nestri"; + /// Where the workload finds the relay. pub const SOCKET: &str = "/nestri/payload.sock"; From 00a2bdab30f7b8a6062c5c35a2db0935f62d2330 Mon Sep 17 00:00:00 2001 From: KAAL1 Date: Sun, 6 Sep 2026 14:24:04 +0300 Subject: [PATCH 3/4] fix(nesinit): give one look at the address socket a deadline There was a cap on how much this would read and none on how long it would wait. The far end can accept a connection and then write nothing, and a read with no deadline turns that into a poll loop that never runs again: the address already forwarded stays correct, and the better one that arrives afterwards is never seen. That is the failure this polls to avoid, reached by a different route. Five seconds, against a two second interval, so an answer that is merely slow still lands and one that is never coming is abandoned. The test hangs against the code as it was, which is the whole point of it. --- apps/nesinit/src/ticket.rs | 61 +++++++++++++++++++++++++++++++++++++- 1 file changed, 60 insertions(+), 1 deletion(-) diff --git a/apps/nesinit/src/ticket.rs b/apps/nesinit/src/ticket.rs index 2d98fba5..991bd8f6 100644 --- a/apps/nesinit/src/ticket.rs +++ b/apps/nesinit/src/ticket.rs @@ -41,6 +41,15 @@ pub const SOCKET: &str = "/tmp/nestri-ticket.sock"; /// connection to a unix socket per interval. const EVERY: Duration = Duration::from_secs(2); +/// How long one look is given before it is abandoned. +/// +/// A cap on time, next to the cap on size below and for the same reason: the +/// far end can accept a connection and then write nothing at all, and a read +/// with no deadline turns that into a poll loop that never runs again. The +/// address it already forwarded stays correct; the better one that arrives +/// later never would. +const PATIENCE: Duration = Duration::from_secs(5); + /// The longest address this will read. /// /// One line of text. A cap rather than a preference: the process on the other @@ -57,7 +66,7 @@ const LONGEST: u64 = 8 * 1024; pub async fn carry(path: PathBuf, out: Sender) { let mut sent: Option = None; loop { - match read(&path).await { + match look(&path).await { Ok(current) if Some(¤t) != sent.as_ref() => { // The address itself is not logged. It is a capability to reach // this session, and a log inside the guest is the one place it @@ -85,6 +94,17 @@ pub async fn carry(path: PathBuf, out: Sender) { } } +/// One look at the socket, abandoned if it takes longer than [`PATIENCE`]. +async fn look(path: &Path) -> io::Result { + match tokio::time::timeout(PATIENCE, read(path)).await { + Ok(result) => result, + Err(_) => Err(io::Error::new( + io::ErrorKind::TimedOut, + "the socket accepted a connection and did not answer", + )), + } +} + /// One line from the socket, which is the whole protocol. async fn read(path: &Path) -> io::Result { let stream = UnixStream::connect(path).await?; @@ -204,6 +224,45 @@ mod tests { let _ = std::fs::remove_file(&path); } + /// A server that accepts and then says nothing must not take the poll loop + /// with it. Without a deadline on the read this hangs forever, and the + /// address that arrives afterwards is never seen. + #[tokio::test] + async fn a_server_that_answers_nothing_does_not_stop_the_search() { + let path = scratch("mute"); + let listener = UnixListener::bind(&path).unwrap(); + let held = std::sync::Arc::new(tokio::sync::Mutex::new(Vec::new())); + { + let held = held.clone(); + tokio::spawn(async move { + // Accepted and kept open, deliberately unanswered, which is + // what a wedged producer looks like from here. + let mut answered = false; + loop { + let Ok((mut stream, _)) = listener.accept().await else { + return; + }; + if answered { + let _ = stream.write_all(b"nestri:eventually\n").await; + } else { + answered = true; + held.lock().await.push(stream); + } + } + }); + } + + let (tx, mut rx) = mpsc::channel(4); + tokio::spawn(carry(path.clone(), tx)); + + let first = tokio::time::timeout(PATIENCE + EVERY * 4, rx.recv()) + .await + .expect("the carrier never got past a server that would not answer") + .expect("the channel closed"); + assert_eq!(first, "nestri:eventually"); + let _ = std::fs::remove_file(&path); + } + /// An answer with nothing in it is not an address. Forwarding one would /// publish an empty string as somewhere to connect. #[tokio::test] From f74de9beb8f72402c74ad42fb792864b6d0fe009 Mon Sep 17 00:00:00 2001 From: Wanjohi Date: Sun, 6 Sep 2026 18:20:30 +0300 Subject: [PATCH 4/4] fix(nesinit): do not mount over the share tree, and check who serves an address Four findings from review, all of them real. The relay's directory was mounted on the tree a session's shares live in. A fresh tmpfs there hides every directory the image prepared underneath it: the install, the user state, the work directory, and the mount point the log share is attached to from fstab. A box would have come up with a socket and without any of the places its workload looks for its files, and the exact-path check could not notice, because what fstab mounts is a directory inside that tree rather than the tree itself. It moves to /run, which is where a runtime socket belongs, is a tmpfs already, and has nothing else mounted inside it. It was also owned by this process and closed to everyone else, which stopped the workload traversing it to reach the relay at all. The directory is now readable and searchable, and still writable by nothing but this process, which is what makes the socket in it unreplaceable; the socket itself is what the workload is allowed to connect to. The permission belongs on the socket rather than on the path. The address served to a reader was built once at startup and served forever, so a reader that polls for a better one could only ever get the first. An endpoint does not know all of its own addresses when it binds: the first is the one that works on the same network and fails from anywhere else. It is now rebuilt per read, which is what makes polling for it worth doing. And the address was taken from whoever held a path in a directory the workload can write. Workload code could unlink the socket a service was listening on, bind its own, and every read afterwards would hand the client an address of its choosing -- a session given to somebody else rather than a session that fails. The peer's credentials are now checked before a byte is read, from the kernel rather than from anything the peer says about itself, and an address served by the workload's own user is refused and said loudly. That check is only worth something while the workload has a user of its own, so the image grows one. Two users, and they must stay two: one runs the services that ship in the image, the other is who a workload runs as. Sharing one does not weaken the check, it makes every session fail it. A workload running as root is every user at once and cannot be told apart from anything; the check stands down there and says so at boot instead, because refusing root would refuse whatever legitimately serves the address as well. Also bumps tinyvec by a patch release. It does not build on this toolchain -- `vec` resolves to the module and not the macro -- which made every crate that depends on an endpoint, including this one, unbuildable. Pre-existing and nothing to do with this change; the lockfile said the same version before it. --- Cargo.lock | 4 +- apps/neshub/src/ipc_listener.rs | 22 +++- apps/neshub/src/main.rs | 12 +- apps/nesinit/src/filesystems.rs | 67 ++++++++++- apps/nesinit/src/main.rs | 12 +- apps/nesinit/src/payload.rs | 24 +++- apps/nesinit/src/session.rs | 177 ++++++++++++++++++++++------- apps/nesinit/src/ticket.rs | 192 ++++++++++++++++++++++++++++++-- build/Dockerfile | 16 ++- 9 files changed, 460 insertions(+), 66 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index c7bbf979..67f8d545 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -4214,9 +4214,9 @@ dependencies = [ [[package]] name = "tinyvec" -version = "1.13.0" +version = "1.13.2" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "0ba2077be5cd93c7408c849e45a4ab261b59f923f4bcdee2a724b8ed5a175a77" +checksum = "4cf0ded5c4e56918d8f8a339e1bb67d038d3bc6d144ac407904015ba2e4cde9b" dependencies = [ "tinyvec_macros", ] diff --git a/apps/neshub/src/ipc_listener.rs b/apps/neshub/src/ipc_listener.rs index e61d3fd9..c3f977cb 100644 --- a/apps/neshub/src/ipc_listener.rs +++ b/apps/neshub/src/ipc_listener.rs @@ -292,7 +292,24 @@ pub async fn run_stats_ipc_listener( tracing::info!("stats IPC listener exited"); } -pub async fn run_ticket_ipc_listener(socket_path: PathBuf, ticket: crate::NestriTicket) { +/// Serve the address a client needs, rebuilt on every read. +/// +/// **The ticket is built per connection and not once at startup.** An endpoint +/// does not know all of its own addresses when it binds: a direct one is there +/// immediately, and a relayed or hole-punched one becomes known seconds later. +/// A ticket captured once therefore carries only the address that was available +/// first, which is the one that works on the same network and fails from +/// anywhere else — and whoever reads this polls precisely so that a better +/// answer can replace it. Serving a snapshot made that polling pointless: every +/// read returned the same local-only address forever. +/// +/// The stream name is generated once and kept, because it identifies this +/// session rather than describing how to reach it. Only the addresses change. +pub async fn run_ticket_ipc_listener( + socket_path: PathBuf, + endpoint: iroh::Endpoint, + stream_name: String, +) { if socket_path.exists() { let _ = std::fs::remove_file(&socket_path); } @@ -319,6 +336,9 @@ pub async fn run_ticket_ipc_listener(socket_path: PathBuf, ticket: crate::Nestri loop { match listener.accept().await { Ok((mut stream, _)) => { + // Asked of the endpoint now, so an address it has learned since + // the last read is in this answer. + let ticket = crate::NestriTicket::new(endpoint.addr(), stream_name.clone()); if let Err(e) = stream.write_all(format!("{ticket}\n").as_bytes()).await { tracing::warn!("could not write ticket to IPC: {e}"); } diff --git a/apps/neshub/src/main.rs b/apps/neshub/src/main.rs index 27636cd1..9cb96bc9 100644 --- a/apps/neshub/src/main.rs +++ b/apps/neshub/src/main.rs @@ -207,7 +207,10 @@ async fn main() -> Result<()> { // ── Accept mode: generate ticket, wait for desktop-app to connect ───── let stream_name = ticket::generate_stream_name(); - let ticket = NestriTicket::new(endpoint_addr, stream_name); + // For the log line below only. What a reader of the socket gets is built + // per read from the endpoint itself, because the addresses this can be + // reached at are not all known yet. + let ticket = NestriTicket::new(endpoint_addr, stream_name.clone()); tracing::info!("╔═══════════════╗"); tracing::info!("║ NESTRI TICKET ║"); @@ -245,7 +248,12 @@ async fn main() -> Result<()> { let ticket_ipc = args.ticket_ipc.clone(); tokio::spawn({ - async move { ipc_listener::run_ticket_ipc_listener(ticket_ipc, ticket).await } + // The endpoint rather than a ticket made from it: the addresses it can + // be reached at are not all known yet, and whoever reads this socket + // re-reads it so that a better one can replace the first. + let endpoint = endpoint.clone(); + let stream_name = stream_name.clone(); + async move { ipc_listener::run_ticket_ipc_listener(ticket_ipc, endpoint, stream_name).await } }); // Accept loop diff --git a/apps/nesinit/src/filesystems.rs b/apps/nesinit/src/filesystems.rs index 8c2768a8..04d03d5e 100644 --- a/apps/nesinit/src/filesystems.rs +++ b/apps/nesinit/src/filesystems.rs @@ -63,14 +63,42 @@ const EARLY: &[Early] = &[ cost: "whatever serves this session's address cannot bind its socket, \ so the session never gets one", }, + // `/run` before anything under it, for the same reason `/proc` comes first: + // a directory cannot be created inside a mount that is not there, and the + // root it would otherwise land on is read-only. + Early { + source: "tmpfs", + target: "/run", + fstype: "tmpfs", + flags: NOSUID_NODEV, + // Octal, and without a leading zero on purpose: the kernel parses a + // tmpfs mode as octal either way, and this is the spelling `mount` + // itself documents. + data: "mode=755", + cost: "there is nowhere for a runtime socket to live, so neither the \ + payload relay nor this session's address can be served", + }, + // The relay's own directory, and it is deliberately **not** in the tree the + // session's shares live in. + // + // It was, and that was wrong in a way no test here would have caught: a + // fresh tmpfs over the share tree hides every directory the image prepared + // underneath it — the install, the user state, the work directory, and the + // mount point the log share is attached to from `fstab`. The box then has a + // socket and none of the places its workload expects to find its files, and + // the exact-path check below cannot notice, because what `fstab` mounts is + // a directory *inside* that tree rather than the tree itself. + // + // Owned by this process and writable by nothing else, which is what makes + // the socket in it unreplaceable. The workload reaches it because the + // directory is traversable and the socket itself is not restricted; see + // `payload::serve`. Early { source: "tmpfs", target: crate::payload::DIRECTORY, fstype: "tmpfs", flags: NOSUID_NODEV | libc::MS_NOEXEC, - // Only this process and the workload it starts, and they are the only - // two that ever have business here. - data: "mode=0770", + data: "mode=755", cost: "the payload relay cannot bind, so nothing reaches the workload \ over the channel", }, @@ -198,6 +226,39 @@ mod tests { assert_eq!(EARLY[0].target, "/proc"); } + /// A mount has to come after whatever it lives inside, or it is a + /// directory created on a read-only root and the mount fails. + #[test] + fn nothing_is_mounted_before_the_mount_it_lives_inside() { + for (i, early) in EARLY.iter().enumerate() { + for other in &EARLY[i + 1..] { + assert!( + !early.target.starts_with(&format!("{}/", other.target)), + "{} is mounted before {}, which contains it", + early.target, + other.target + ); + } + } + } + + /// **Nothing here may be mounted over the tree the session's shares live + /// in.** A fresh tmpfs there hides every directory the image prepared + /// underneath — the install, the user state, the work directory, and the + /// mount point the log share attaches to — and the exact-path check cannot + /// notice, because what is mounted from `fstab` is a directory inside that + /// tree rather than the tree itself. So a box would come up with a socket + /// and without any of the places its workload looks for its files. + #[test] + fn the_share_tree_is_never_mounted_over() { + for early in EARLY { + assert_ne!( + early.target, "/nestri", + "this hides the directories the image prepared for a session" + ); + } + } + /// The relay's directory is the one this cannot hardcode: it belongs to /// `payload`, and a rename there that missed this file would take the /// relay down again in exactly the way this exists to prevent. diff --git a/apps/nesinit/src/main.rs b/apps/nesinit/src/main.rs index 239be9b7..cb44cbed 100644 --- a/apps/nesinit/src/main.rs +++ b/apps/nesinit/src/main.rs @@ -114,10 +114,18 @@ async fn guest(waiters: &Waiters, workload: &mut Process) -> anyhow::Result outcome?, + outcome = session::run(channel, workload, &mut ports, &mut found_rx, &untrusted) => 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 index 944bcbca..cbfebe00 100644 --- a/apps/nesinit/src/payload.rs +++ b/apps/nesinit/src/payload.rs @@ -9,6 +9,7 @@ // anything above it. use std::io; +use std::os::unix::fs::PermissionsExt; use std::path::Path; use nesprotocol::lifecycle::{Payload, from_line, to_line}; @@ -22,10 +23,18 @@ use tokio::sync::mpsc::{Receiver, Sender}; /// is read-only, so this is a tmpfs that `filesystems` puts there, and a /// rename here that did not reach the mount table would take the relay down /// with an `EROFS` that looks like nothing to do with a path. -pub const DIRECTORY: &str = "/nestri"; +/// +/// **Under `/run` rather than in the tree the session's shares live in**, and +/// that was a correction. Putting a fresh tmpfs on the share tree hid every +/// directory the image had prepared underneath it — the install, the user +/// state, the work directory and the log share's own mount point — so a box +/// gained a socket and lost the places its workload was supposed to find its +/// files. `/run` is where a runtime socket belongs, it is a tmpfs already, and +/// nothing else is mounted inside it. +pub const DIRECTORY: &str = "/run/nestri"; /// Where the workload finds the relay. -pub const SOCKET: &str = "/nestri/payload.sock"; +pub const SOCKET: &str = "/run/nestri/payload.sock"; /// The longest envelope this will assemble before giving up on the connection. /// @@ -63,6 +72,17 @@ pub async fn serve( let _ = std::fs::remove_file(path); let listener = UnixListener::bind(path)?; + // The workload does not run as this process does, and it has to be able to + // connect. It reaches the socket through a directory this process owns and + // nothing else may write to, so the permission that matters is on the + // socket rather than on the path: the directory is what stops anybody + // replacing this listener, and this is what lets the workload talk to it. + // + // Without it the workload gets `EACCES` on connect and the payload layer + // is dead in both directions, silently, because nothing in the guest is + // waiting to be told about a relay it cannot reach. + std::fs::set_permissions(path, std::fs::Permissions::from_mode(0o666))?; + loop { let stream = tokio::select! { accepted = listener.accept() => accepted?.0, diff --git a/apps/nesinit/src/session.rs b/apps/nesinit/src/session.rs index 8490f499..a94e0b33 100644 --- a/apps/nesinit/src/session.rs +++ b/apps/nesinit/src/session.rs @@ -41,12 +41,13 @@ pub async fn run( workload: &mut W, payload: &mut Ports, addresses: &mut Receiver, + untrusted: &crate::ticket::Untrusted, ) -> std::io::Result where C: AsyncRead + AsyncWrite, W: Workload, { - match converse(channel, workload, payload, addresses).await { + match converse(channel, workload, payload, addresses, untrusted).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 @@ -73,6 +74,7 @@ async fn converse( workload: &mut W, payload: &mut Ports, addresses: &mut Receiver, + untrusted: &crate::ticket::Untrusted, ) -> std::io::Result where C: AsyncRead + AsyncWrite, @@ -186,6 +188,11 @@ where } } + // Before the workload exists, so there is no window in which + // it is running and something else would still be trusted to + // serve this session's address. + untrusted.is(descriptor.exec.uid); + match workload.start(&descriptor.exec) { Ok(exited) => { send(&mut writer, &GuestToHost::Started).await?; @@ -354,9 +361,15 @@ mod tests { 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, &mut ports, &mut nowhere()) - .await - .unwrap(); + let outcome = run( + guest, + &mut workload, + &mut ports, + &mut nowhere(), + &crate::ticket::Untrusted::unknown(), + ) + .await + .unwrap(); (outcome, workload) }); @@ -385,9 +398,15 @@ mod tests { 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, &mut ports, &mut found_rx) - .await - .unwrap() + run( + guest, + &mut workload, + &mut ports, + &mut found_rx, + &crate::ticket::Untrusted::unknown(), + ) + .await + .unwrap() }); assert_eq!( @@ -436,9 +455,15 @@ mod tests { 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, &mut ports, &mut found_rx) - .await - .unwrap() + run( + guest, + &mut workload, + &mut ports, + &mut found_rx, + &crate::ticket::Untrusted::unknown(), + ) + .await + .unwrap() }); assert_eq!( @@ -462,9 +487,15 @@ mod tests { 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, &mut ports, &mut nowhere()) - .await - .unwrap(); + let outcome = run( + guest, + &mut workload, + &mut ports, + &mut nowhere(), + &crate::ticket::Untrusted::unknown(), + ) + .await + .unwrap(); (outcome, workload) }); @@ -490,9 +521,15 @@ mod tests { 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, &mut ports, &mut nowhere()) - .await - .unwrap(); + let outcome = run( + guest, + &mut workload, + &mut ports, + &mut nowhere(), + &crate::ticket::Untrusted::unknown(), + ) + .await + .unwrap(); (outcome, workload) }); @@ -528,9 +565,15 @@ mod tests { 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, &mut ports, &mut nowhere()) - .await - .unwrap() + run( + guest, + &mut workload, + &mut ports, + &mut nowhere(), + &crate::ticket::Untrusted::unknown(), + ) + .await + .unwrap() }); assert!(matches!(caller.expect().await, GuestToHost::Ready { .. })); @@ -561,9 +604,15 @@ mod tests { 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, &mut ports, &mut nowhere()) - .await - .unwrap(); + let outcome = run( + guest, + &mut workload, + &mut ports, + &mut nowhere(), + &crate::ticket::Untrusted::unknown(), + ) + .await + .unwrap(); (outcome, workload) }); @@ -589,9 +638,15 @@ mod tests { 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, &mut ports, &mut nowhere()) - .await - .unwrap(); + let outcome = run( + guest, + &mut workload, + &mut ports, + &mut nowhere(), + &crate::ticket::Untrusted::unknown(), + ) + .await + .unwrap(); (outcome, workload) }); @@ -626,9 +681,15 @@ mod tests { 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, &mut ports, &mut nowhere()) - .await - .unwrap() + run( + guest, + &mut workload, + &mut ports, + &mut nowhere(), + &crate::ticket::Untrusted::unknown(), + ) + .await + .unwrap() }); assert!(matches!(caller.expect().await, GuestToHost::Ready { .. })); @@ -651,9 +712,15 @@ mod tests { 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, &mut ports, &mut nowhere()) - .await - .unwrap(); + let outcome = run( + guest, + &mut workload, + &mut ports, + &mut nowhere(), + &crate::ticket::Untrusted::unknown(), + ) + .await + .unwrap(); (outcome, workload) }); @@ -685,9 +752,15 @@ mod tests { 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, &mut nowhere()) - .await - .unwrap(); + let outcome = run( + guest, + &mut workload, + &mut ports, + &mut nowhere(), + &crate::ticket::Untrusted::unknown(), + ) + .await + .unwrap(); (outcome, workload) }); @@ -729,9 +802,15 @@ mod tests { from_workload: up_rx, }; let mut workload = Double::exits_when_stopped(Exit::code(0)); - run(guest, &mut workload, &mut ports, &mut nowhere()) - .await - .unwrap() + run( + guest, + &mut workload, + &mut ports, + &mut nowhere(), + &crate::ticket::Untrusted::unknown(), + ) + .await + .unwrap() }); let mut to_relay = down_rx; @@ -780,9 +859,15 @@ mod tests { from_workload: up_rx, }; let mut workload = Double::exits_when_stopped(Exit::code(0)); - run(guest, &mut workload, &mut ports, &mut nowhere()) - .await - .unwrap() + run( + guest, + &mut workload, + &mut ports, + &mut nowhere(), + &crate::ticket::Untrusted::unknown(), + ) + .await + .unwrap() }); assert!(matches!(caller.expect().await, GuestToHost::Ready { .. })); @@ -828,9 +913,15 @@ mod tests { from_workload: up_rx, }; let mut workload = Double::exits_when_stopped(Exit::code(0)); - run(guest, &mut workload, &mut ports, &mut nowhere()) - .await - .unwrap() + run( + guest, + &mut workload, + &mut ports, + &mut nowhere(), + &crate::ticket::Untrusted::unknown(), + ) + .await + .unwrap() }); assert!(matches!(caller.expect().await, GuestToHost::Ready { .. })); diff --git a/apps/nesinit/src/ticket.rs b/apps/nesinit/src/ticket.rs index 991bd8f6..82eebf18 100644 --- a/apps/nesinit/src/ticket.rs +++ b/apps/nesinit/src/ticket.rs @@ -22,8 +22,24 @@ // long-lived one. Dialling also means a server that has not bound yet is an // error this retries, rather than a connection that has to be waited for // without knowing whether it is coming. +// +// # Who is allowed to answer +// +// Dialling a path means trusting whoever is behind it, and an address is the +// capability to reach this session — so the wrong answer here does not break a +// session, it hands one to somebody else. The workload runs arbitrary code, and +// on a writable directory it can unlink whatever bound the socket and bind a +// replacement; every read after that returns an address of its choosing, and +// the client outside connects there instead. +// +// So the peer's credentials are checked and an answer from the workload's own +// user is refused. That check is only as good as the workload having a user of +// its own: run it as the same user as the process serving the address and +// nothing can tell the two apart, which is [`Peer::indistinguishable`] and is +// said out loud at boot rather than discovered later. use std::io; +use std::os::fd::AsRawFd; use std::path::{Path, PathBuf}; use std::time::Duration; @@ -57,16 +73,66 @@ const PATIENCE: Duration = Duration::from_secs(5); /// buffer is the one the kernel has been told not to kill. const LONGEST: u64 = 8 * 1024; +/// Who may not serve this session's address. +/// +/// The workload's user, and nothing else is excluded — this is not an allow +/// list of trusted uids, because init does not know which user an image happens +/// to run its media components as, and inventing one here would be a second +/// place for that to be configured wrongly. +/// +/// Shared and filled in later, because the carrier starts before the descriptor +/// that names the user arrives. That leaves no gap: a workload cannot serve +/// anything before it is started, and it is started from the same descriptor, +/// which sets this first. +#[derive(Clone, Debug, Default)] +pub struct Untrusted(std::sync::Arc); + +/// No workload has been started, so there is no untrusted user yet. +const NOBODY: u32 = u32::MAX; + +impl Untrusted { + pub fn unknown() -> Self { + Self(std::sync::Arc::new(std::sync::atomic::AtomicU32::new( + NOBODY, + ))) + } + + /// Name the user the workload runs as. Called before it is started. + pub fn is(&self, uid: u32) { + self.0.store(uid, std::sync::atomic::Ordering::Release); + if uid == 0 { + tracing::warn!( + "the workload runs as root, so an address it serves cannot be \ + told apart from a real one. Give it a user of its own." + ); + } + } + + /// The uid to refuse, or `None` when refusing anything would be wrong. + /// + /// `None` covers two cases that want the same answer for different reasons: + /// nothing has been started yet, so no peer can be the workload; and the + /// workload runs as `root`, which is every user at once — refusing root + /// would refuse whatever legitimately serves the address as well. The uid + /// being shared is the thing an operator has to fix, and `is` says so. + fn refuse(&self) -> Option { + match self.0.load(std::sync::atomic::Ordering::Acquire) { + NOBODY | 0 => None, + uid => Some(uid), + } + } +} + /// Forward every new address for as long as the session lasts. /// /// Never returns on its own. A server that is not there yet, or has gone away, /// is retried at the next interval — there is nothing here worth ending a /// running session over, and an address that stops being re-offered does not /// stop being correct. -pub async fn carry(path: PathBuf, out: Sender) { +pub async fn carry(path: PathBuf, out: Sender, untrusted: Untrusted) { let mut sent: Option = None; loop { - match look(&path).await { + match look(&path, &untrusted).await { Ok(current) if Some(¤t) != sent.as_ref() => { // The address itself is not logged. It is a capability to reach // this session, and a log inside the guest is the one place it @@ -82,6 +148,14 @@ pub async fn carry(path: PathBuf, out: Sender) { sent = Some(current); } Ok(_) => {} + // Not the same as no address yet, and it must not be logged as + // though it were: this says something *is* serving an address and + // it is the one thing that may not. A session with no address at + // all is a legible failure; a session pointed somewhere else is + // not, so this is the line that has to be found afterwards. + Err(error) if error.kind() == io::ErrorKind::PermissionDenied => { + tracing::error!(%error, "refusing an address for this session"); + } Err(error) => { // Expected until whatever serves the address has bound, so it // is not a warning the first several times. It stays at this @@ -95,8 +169,8 @@ pub async fn carry(path: PathBuf, out: Sender) { } /// One look at the socket, abandoned if it takes longer than [`PATIENCE`]. -async fn look(path: &Path) -> io::Result { - match tokio::time::timeout(PATIENCE, read(path)).await { +async fn look(path: &Path, untrusted: &Untrusted) -> io::Result { + match tokio::time::timeout(PATIENCE, read(path, untrusted)).await { Ok(result) => result, Err(_) => Err(io::Error::new( io::ErrorKind::TimedOut, @@ -106,8 +180,22 @@ async fn look(path: &Path) -> io::Result { } /// One line from the socket, which is the whole protocol. -async fn read(path: &Path) -> io::Result { +async fn read(path: &Path, untrusted: &Untrusted) -> io::Result { let stream = UnixStream::connect(path).await?; + + // Before a byte is read. The kernel answers this about the socket's peer + // rather than about the path, so it cannot be spoofed by whoever holds the + // path — which is the whole reason the check is worth anything. + if let Some(refuse) = untrusted.refuse() + && peer_uid(&stream)? == refuse + { + return Err(io::Error::new( + io::ErrorKind::PermissionDenied, + "the workload is serving this session's address, which would point \ + a client at whatever it chose", + )); + } + let mut line = String::new(); BufReader::new(stream.take(LONGEST)) .read_line(&mut line) @@ -122,6 +210,35 @@ async fn read(path: &Path) -> io::Result { Ok(line) } +/// The uid of the process on the other end of a connected unix socket. +/// +/// From the kernel, at connect time, and not from anything the peer says about +/// itself. `SO_PEERCRED` records who held the other end when it connected, so a +/// process cannot claim a uid it does not have. +fn peer_uid(stream: &UnixStream) -> io::Result { + let mut credentials = libc::ucred { + pid: 0, + uid: u32::MAX, + gid: u32::MAX, + }; + let mut length = std::mem::size_of::() as libc::socklen_t; + // SAFETY: a connected socket this function borrows, and an out-parameter of + // exactly the length being passed. + let rc = unsafe { + libc::getsockopt( + stream.as_raw_fd(), + libc::SOL_SOCKET, + libc::SO_PEERCRED, + (&raw mut credentials).cast::(), + &raw mut length, + ) + }; + if rc != 0 { + return Err(io::Error::last_os_error()); + } + Ok(credentials.uid) +} + #[cfg(test)] mod tests { use super::*; @@ -160,7 +277,7 @@ mod tests { serve(path.clone(), vec![Some("nestri:abc".into())]); let (tx, mut rx) = mpsc::channel(4); - tokio::spawn(carry(path.clone(), tx)); + tokio::spawn(carry(path.clone(), tx, Untrusted::unknown())); let first = tokio::time::timeout(Duration::from_secs(10), rx.recv()) .await @@ -191,7 +308,7 @@ mod tests { ); let (tx, mut rx) = mpsc::channel(4); - tokio::spawn(carry(path.clone(), tx)); + tokio::spawn(carry(path.clone(), tx, Untrusted::unknown())); let mut seen = Vec::new(); while seen.len() < 2 { @@ -205,13 +322,68 @@ mod tests { let _ = std::fs::remove_file(&path); } + /// The peer's user decides whether an address is trusted, and this process + /// is the peer in a test — so naming *it* as the workload is a real refusal + /// of a real connection, not a stubbed one. + /// + /// The attack this closes: the workload unlinks whatever bound the socket, + /// binds its own, and every read afterwards hands the client an address of + /// the workload's choosing. + #[tokio::test] + async fn an_address_served_by_the_workload_is_refused() { + let path = scratch("hostile"); + serve(path.clone(), vec![Some("nestri:attacker".into())]); + + // SAFETY: reading this process's own uid cannot fail. + let ours = unsafe { libc::getuid() }; + let untrusted = Untrusted::unknown(); + untrusted.is(ours); + + let error = look(&path, &untrusted) + .await + .expect_err("an address from the workload's own user was accepted"); + assert_eq!(error.kind(), io::ErrorKind::PermissionDenied); + + // And the same socket is read happily once the workload is somebody + // else, which is what shows the refusal is about the peer and not about + // the socket. + let elsewhere = Untrusted::unknown(); + elsewhere.is(ours + 1); + assert_eq!(look(&path, &elsewhere).await.unwrap(), "nestri:attacker"); + + let _ = std::fs::remove_file(&path); + } + + /// Before a workload is started there is nothing to refuse, and refusing + /// anything then would mean no session ever got an address. + #[tokio::test] + async fn nothing_is_refused_before_a_workload_exists() { + let path = scratch("early"); + serve(path.clone(), vec![Some("nestri:real".into())]); + let untrusted = Untrusted::unknown(); + assert_eq!(untrusted.refuse(), None); + assert_eq!(look(&path, &untrusted).await.unwrap(), "nestri:real"); + let _ = std::fs::remove_file(&path); + } + + /// A workload running as root is every user at once, so refusing root would + /// refuse whatever legitimately serves the address too. The check stands + /// down and `is` warns instead — the uid being shared is the operator's to + /// fix and this is not the place to fail closed over it. + #[tokio::test] + async fn a_root_workload_leaves_nothing_to_tell_apart() { + let untrusted = Untrusted::unknown(); + untrusted.is(0); + assert_eq!(untrusted.refuse(), None); + } + /// Nothing serving the socket yet is the ordinary case at boot, not a /// failure: this starts before whatever binds it. #[tokio::test] async fn a_socket_that_is_not_there_yet_is_waited_out_rather_than_failed() { let path = scratch("late"); let (tx, mut rx) = mpsc::channel(4); - tokio::spawn(carry(path.clone(), tx)); + tokio::spawn(carry(path.clone(), tx, Untrusted::unknown())); tokio::time::sleep(EVERY * 2).await; serve(path.clone(), vec![Some("nestri:late".into())]); @@ -253,7 +425,7 @@ mod tests { } let (tx, mut rx) = mpsc::channel(4); - tokio::spawn(carry(path.clone(), tx)); + tokio::spawn(carry(path.clone(), tx, Untrusted::unknown())); let first = tokio::time::timeout(PATIENCE + EVERY * 4, rx.recv()) .await @@ -271,7 +443,7 @@ mod tests { serve(path.clone(), vec![None, Some("nestri:real".into())]); let (tx, mut rx) = mpsc::channel(4); - tokio::spawn(carry(path.clone(), tx)); + tokio::spawn(carry(path.clone(), tx, Untrusted::unknown())); let first = tokio::time::timeout(Duration::from_secs(10), rx.recv()) .await diff --git a/build/Dockerfile b/build/Dockerfile index 29abc082..0c9be9e3 100644 --- a/build/Dockerfile +++ b/build/Dockerfile @@ -145,9 +145,23 @@ RUN pacman -Syu --noconfirm --needed \ # groupadd -f so this is idempotent whether or not udev's rules already # created these. +# +# **Two users, and they must stay two.** `nestri` runs the services that come +# with this image; `nesplay` is who a workload runs as. Sharing one user between +# them is what lets workload code impersonate a service — it can replace the +# socket a service listens on and answer in its place, and the answer that +# matters is the address a client is told to connect to. Init refuses an address +# served by the workload's own user, so a single shared user does not merely +# weaken that check, it makes every session fail it. +# +# The uid a workload actually runs as is chosen by whoever asks for the box, not +# here; this account exists so that the number has a home, a shell and a name in +# `ps`, and so the separation has somewhere to be written down. RUN groupadd -f audio && groupadd -f video && groupadd -f input && groupadd -f render && \ useradd -m -u 1000 -s /bin/bash nestri && \ - for g in audio video input render; do gpasswd -a nestri "$g" >/dev/null; done + for g in audio video input render; do gpasswd -a nestri "$g" >/dev/null; done && \ + useradd -m -u 1001 -s /bin/bash nesplay && \ + for g in audio video input render; do gpasswd -a nesplay "$g" >/dev/null; done # ───────────────────────────────────────────────────────────