github-mirrors/netris-nestri
1
0
Fork 0
mirror of https://github.com/nestriness/nestri.git synced 2026-09-19 17:25:19 +03:00
Code Issues Packages Projects Releases Wiki Activity

fix(nesinit): the relay may not stall the session, and may not buffer without end

Browse Source 
Three problems in the relay, all of them found in review.

Handing an envelope over waited for room. That loop also carries stop,
shutdown and the workload's exit, so a workload slow to read its own mail — or
one that never connected — could hold the lifecycle layer still behind it. It
never waits now: an envelope that will not fit is dropped, which costs nothing,
because what crosses this layer is re-sent when it changes.

Envelopes were queued for a workload that was not there. The queue filled with
copies that would be stale by the time anyone connected, and filling it was
what stalled the session. Nothing is held while the socket has nobody on it.

A frame had no maximum length. The workload can write for as long as it likes
without ever sending a newline, and the process assembling that is the one the
kernel has been told not to kill, so the memory it takes comes out of
everything else in the guest. Past 64 KiB the connection is dropped and the
relay waits for the next one; the failure says how long the frame got and
nothing about what was in it.

Also, a tag or a mount point with a nul byte in it was quietly turned into an
empty string, so an unmountable descriptor arrived later as a mount failure
about something else, after the mount point had already been created. It is
refused by name now, before anything is created.

The relay's tests grew a harness that waits for the connection to be carried
before sending anything down it, because dropping what arrives with nobody
connected made "connected" something a test has to establish rather than
assume.
This commit is contained in:
KAAL1
2026-09-05 09:48:34 +03:00
committed by Wanjohi
parent 7b99f49f62
commit a94b323edb
4 changed files with 381 additions and 66 deletions
Download Patch File Download Diff File Expand all files Collapse all files

13
apps/nesinit/README.md
View File

@@ -65,6 +65,19 @@ Both directions. Inside the guest an envelope crosses a unix socket at
That socket is a mechanism and expected to change; the envelope is the boundary
and is not.
Nothing is held for a workload that is not on the relay, and nothing waits on
one that is slow to read. An envelope that arrives with nobody connected is
dropped, and so is one that arrives faster than the workload reads: what
crosses this layer is re-sent when it changes, so a queued copy is a stale copy
— and the queue that would hold it is on the same loop that carries stop,
shutdown and the workload's exit, none of which may wait behind it.
A frame is capped at 64 KiB. The workload is on the other end of that socket
and can write for as long as it likes without ever sending a newline; the
process assembling it is the one the kernel has been told not to kill, so an
unbounded buffer there comes out of everything else in the guest. Past the cap
the connection is dropped and the relay waits for the next one.
`body` is a string rather than nested JSON, deliberately. A document nesinit
can index into is a document nesinit can grow to depend on, and then the layer
is no longer opaque and the boundary it exists to draw is gone.

297
apps/nesinit/src/payload.rs
View File

@@ -12,13 +12,22 @@ use std::io;
use std::path::Path;
use nesprotocol::lifecycle::{Payload, from_line, to_line};
use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader};
use tokio::io::{AsyncBufRead, AsyncBufReadExt, AsyncWriteExt, BufReader};
use tokio::net::{UnixListener, UnixStream};
use tokio::sync::mpsc::{Receiver, Sender};
/// Where the workload finds the relay.
pub const SOCKET: &str = "/nestri/payload.sock";
/// The longest envelope this will assemble before giving up on the connection.
///
/// A cap rather than a preference. The workload is on the other end of this
/// socket and can write for as long as it likes without ever sending a
/// newline; without a limit, the process holding the buffer is the one process
/// in the guest the kernel has been told not to kill, so the memory it takes
/// comes out of everything else.
const LONGEST_ENVELOPE: usize = 64 * 1024;
/// The two directions, as the session sees them.
pub struct Ports {
/// Envelopes bound for the workload.
@@ -47,7 +56,25 @@ pub async fn serve(
let listener = UnixListener::bind(path)?;
loop {
let (stream, _) = listener.accept().await?;
let stream = tokio::select! {
accepted = listener.accept() => accepted?.0,
waiting = outbound.recv() => {
// Dropped rather than queued for whoever connects next. What
// crosses this layer is re-sent when it changes, so what a
// queue would hold is a stale copy, and holding it is also
// what would eventually block the session that fills it.
match waiting {
Some(envelope) => {
tracing::warn!(
envelope = %envelope.summary(),
"dropped an envelope: nothing is on the relay",
);
continue;
}
None => return Ok(()), // the session is over
}
}
};
tracing::info!("the workload is on the relay");
match relay(stream, &mut outbound, &inbound).await {
Ok(()) => tracing::info!("the workload left the relay"),
@@ -62,19 +89,27 @@ async fn relay(
inbound: &Sender<Payload>,
) -> io::Result<()> {
let (reader, mut writer) = stream.into_split();
let mut lines = BufReader::new(reader).lines();
let mut reader = BufReader::new(reader);
let mut frame = Vec::new();
loop {
tokio::select! {
line = lines.next_line() => {
let Some(line) = line? else { return Ok(()) };
read = read_capped(&mut reader, &mut frame) => {
if !read? {
return Ok(()); // the workload closed the socket
}
// Deliberately not `{:?}` on the error either: a decoder is
// entitled to quote what it could not read, and what it could
// not read is a body.
let Ok(payload) = from_line::<Payload>(&line) else {
tracing::warn!(bytes = line.len(), "ignoring an envelope that would not decode");
let decoded = std::str::from_utf8(&frame)
.ok()
.and_then(|line| from_line::<Payload>(line).ok());
let Some(payload) = decoded else {
tracing::warn!(bytes = frame.len(), "ignoring an envelope that would not decode");
frame.clear();
continue;
};
frame.clear();
tracing::debug!(envelope = %payload.summary(), "relaying from the workload");
if inbound.send(payload).await.is_err() {
return Ok(()); // the session is over
@@ -91,10 +126,62 @@ async fn relay(
}
}
/// Read one newline-terminated frame into `frame`, refusing to grow it past
/// [`LONGEST_ENVELOPE`]. `false` at end of stream.
///
/// Cancel-safe, which it has to be to sit in a `select!`: bytes are copied out
/// of the reader and consumed together, so a cancelled read leaves the partial
/// frame in `frame` and the rest of it in the socket.
async fn read_capped<R>(reader: &mut R, frame: &mut Vec<u8>) -> io::Result<bool>
where
R: AsyncBufRead + Unpin,
{
loop {
let consumed;
let complete;
{
let available = reader.fill_buf().await?;
if available.is_empty() {
return Ok(false);
}
match available.iter().position(|byte| *byte == b'\n') {
Some(end) => {
within_cap(frame.len() + end)?;
frame.extend_from_slice(&available[..end]);
consumed = end + 1;
complete = true;
}
None => {
within_cap(frame.len() + available.len())?;
frame.extend_from_slice(available);
consumed = available.len();
complete = false;
}
}
}
reader.consume(consumed);
if complete {
return Ok(true);
}
}
}
/// The error says how long, and nothing about what: what did not fit is a
/// body, and a body is not logged even when it is malformed.
fn within_cap(length: usize) -> io::Result<()> {
if length > LONGEST_ENVELOPE {
return Err(io::Error::other(format!(
"an envelope grew past {LONGEST_ENVELOPE} bytes without ending"
)));
}
Ok(())
}
#[cfg(test)]
mod tests {
use super::*;
use std::sync::{Arc, Mutex};
use tokio::net::unix::{OwnedReadHalf, OwnedWriteHalf};
use tokio::sync::mpsc;
fn socket_path(name: &str) -> std::path::PathBuf {
@@ -128,22 +215,79 @@ mod tests {
}
}
#[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);
/// A relay, and the two ends of it a test drives.
struct Relay {
path: std::path::PathBuf,
down: mpsc::Sender<Payload>,
up: mpsc::Receiver<Payload>,
server: tokio::task::JoinHandle<io::Result<()>>,
}
impl Relay {
fn start(name: &str) -> Self {
let path = socket_path(name);
let (down, outbound) = mpsc::channel(8);
let (inbound, up) = mpsc::channel(8);
let server = tokio::spawn({
let path = path.clone();
async move { serve(&path, down_rx, up_tx).await }
async move { serve(&path, outbound, inbound).await }
});
Self {
path,
down,
up,
server,
}
}
let workload = connect(&path).await;
let (reader, mut writer) = workload.into_split();
let mut lines = BufReader::new(reader).lines();
/// Dial the relay and wait until it is demonstrably carrying the
/// connection.
///
/// `serve` drops what arrives while nothing is connected, so a test
/// that sends downward before the accept has completed is racing it. A
/// line upward is the barrier: it can only arrive once the relay is
/// carrying this connection.
async fn workload(&mut self) -> (BufReader<OwnedReadHalf>, OwnedWriteHalf) {
let (reader, mut writer) = self.dial().await.into_split();
writer
.write_all(to_line(&Payload::new("handshake", "")).unwrap().as_bytes())
.await
.unwrap();
assert_eq!(self.up.recv().await.unwrap().channel, "handshake");
(BufReader::new(reader), writer)
}
/// The listener may not be bound the instant the task is spawned.
async fn dial(&self) -> UnixStream {
for _ in 0..100 {
if let Ok(stream) = UnixStream::connect(&self.path).await {
return stream;
}
tokio::time::sleep(std::time::Duration::from_millis(10)).await;
}
panic!("the relay never came up on {}", self.path.display());
}
}
impl Drop for Relay {
fn drop(&mut self) {
self.server.abort();
let _ = std::fs::remove_file(&self.path);
}
}
#[tokio::test]
async fn an_envelope_crosses_in_both_directions_untouched() {
let mut relay = Relay::start("both-ways");
let (reader, mut writer) = relay.workload().await;
let mut lines = reader.lines();
let body = r#"{"looks":"structured"} and is not"#;
down_tx.send(Payload::new("identity", body)).await.unwrap();
relay
.down
.send(Payload::new("identity", body))
.await
.unwrap();
let line = lines.next_line().await.unwrap().unwrap();
let seen: Payload = from_line(&line).unwrap();
assert_eq!(seen.body, body, "the body arrived changed");
@@ -151,11 +295,7 @@ mod tests {
let back = to_line(&Payload::new("identity", "opaque back")).unwrap();
writer.write_all(back.as_bytes()).await.unwrap();
let up = up_rx.recv().await.unwrap();
assert_eq!(up.body, "opaque back");
server.abort();
let _ = std::fs::remove_file(&path);
assert_eq!(relay.up.recv().await.unwrap().body, "opaque back");
}
#[tokio::test]
@@ -168,20 +308,13 @@ mod tests {
.finish();
let _log = tracing::subscriber::set_default(subscriber);
let path = socket_path("logging");
let (down_tx, down_rx) = mpsc::channel(4);
let (up_tx, mut up_rx) = mpsc::channel(4);
let server = tokio::spawn({
let path = path.clone();
async move { serve(&path, down_rx, up_tx).await }
});
let workload = connect(&path).await;
let (reader, mut writer) = workload.into_split();
let mut lines = BufReader::new(reader).lines();
let mut relay = Relay::start("logging");
let (reader, mut writer) = relay.workload().await;
let mut lines = reader.lines();
let secret = "a-credential-nobody-should-read";
down_tx
relay
.down
.send(Payload::new("identity", secret))
.await
.unwrap();
@@ -195,7 +328,7 @@ mod tests {
)
.await
.unwrap();
up_rx.recv().await.unwrap();
relay.up.recv().await.unwrap();
// An envelope that will not decode is the other way a body reaches a
// log line — a decoder is entitled to quote what it could not read —
@@ -219,19 +352,91 @@ mod tests {
logged.contains("bytes=62"),
"the byte count is loggable:\n{logged}"
);
server.abort();
let _ = std::fs::remove_file(&path);
}
/// The listener may not be bound the instant the task is spawned.
async fn connect(path: &Path) -> UnixStream {
for _ in 0..100 {
if let Ok(stream) = UnixStream::connect(path).await {
return stream;
#[tokio::test]
async fn an_envelope_sent_while_nobody_is_connected_is_dropped_rather_than_queued() {
let mut relay = Relay::start("nobody-home");
// Sent into a relay nothing is on. It is not held for whoever connects
// next: what crosses this layer is re-sent when it changes, so a held
// copy is a stale copy — and a queue that fills is what would
// eventually stall the session filling it.
relay
.down
.send(Payload::new("identity", "stale"))
.await
.unwrap();
let (reader, _writer) = relay.workload().await;
let mut lines = reader.lines();
relay
.down
.send(Payload::new("identity", "current"))
.await
.unwrap();
let line = lines.next_line().await.unwrap().unwrap();
let seen: Payload = from_line(&line).unwrap();
assert_eq!(
seen.body, "current",
"a stale envelope was delivered on connect"
);
}
tokio::time::sleep(std::time::Duration::from_millis(10)).await;
}
panic!("the relay never came up on {}", path.display());
#[tokio::test]
async fn a_frame_that_never_ends_costs_the_connection_and_not_the_guest() {
use tokio::io::AsyncReadExt;
let mut relay = Relay::start("unbounded");
let (mut reader, mut writer) = relay.dial().await.into_split();
// Well past the cap, and not a newline in it. Without a limit the
// buffer holding this grows in the one process the kernel has been
// told not to kill, so what it takes comes out of everything else.
let flood = vec![b'a'; LONGEST_ENVELOPE + 4096];
// The write fails once the relay drops the connection, which is the
// outcome under test rather than a problem with it.
let _ = writer.write_all(&flood).await;
// The relay closes on us rather than keeping the buffer. Bounded,
// because the failure being guarded against is a relay that reads for
// as long as the workload writes.
let mut unread = Vec::new();
let closed = tokio::time::timeout(
std::time::Duration::from_secs(5),
reader.read_to_end(&mut unread),
)
.await
.expect("the relay is still assembling a frame that never ends");
assert_eq!(
closed.unwrap(),
0,
"the relay answered a frame it should have refused"
);
// And it is there for whoever connects next.
let (reader, mut writer) =
tokio::time::timeout(std::time::Duration::from_secs(5), relay.workload())
.await
.expect("the relay never came back for the next workload");
let mut lines = reader.lines();
writer
.write_all(
to_line(&Payload::new("identity", "after the flood"))
.unwrap()
.as_bytes(),
)
.await
.unwrap();
assert_eq!(relay.up.recv().await.unwrap().body, "after the flood");
relay
.down
.send(Payload::new("identity", "down"))
.await
.unwrap();
let line = lines.next_line().await.unwrap().unwrap();
assert_eq!(from_line::<Payload>(&line).unwrap().body, "down");
}
}

78
apps/nesinit/src/session.rs
View File

@@ -183,7 +183,7 @@ where
}
}
}
HostToGuest::Payload { payload: envelope } => hand_over(payload, envelope).await,
HostToGuest::Payload { payload: envelope } => hand_over(payload, envelope),
HostToGuest::Stop => workload.signal_stop(),
HostToGuest::Shutdown => return Ok(Outcome::Shutdown),
}
@@ -199,13 +199,24 @@ enum Event {
/// Hand an envelope to the relay, and treat a relay that is not there as the
/// caller's problem rather than a failure of this session.
async fn hand_over(ports: &mut Ports, envelope: Payload) {
tracing::debug!(envelope = %envelope.summary(), "handing an envelope over");
if ports.to_workload.send(envelope).await.is_err() {
// The workload is not on the relay. Dropped rather than queued: what
// crosses here is re-sent when it changes, so a held copy is a stale
// copy.
tracing::warn!("dropped an envelope: nothing is on the relay");
///
/// It never waits. This loop also carries stop, shutdown and the workload's
/// exit, and none of those may be held up by a workload that is slow to read
/// its own mail — or by one that never connected at all. What crosses this
/// layer is re-sent when it changes, so a dropped copy costs less than a
/// stalled session.
fn hand_over(ports: &mut Ports, envelope: Payload) {
use tokio::sync::mpsc::error::TrySendError;
let summary = envelope.summary();
match ports.to_workload.try_send(envelope) {
Ok(()) => tracing::debug!(envelope = %summary, "handed an envelope over"),
Err(TrySendError::Full(_)) => {
tracing::warn!(envelope = %summary, "dropped an envelope: the relay is behind")
}
Err(TrySendError::Closed(_)) => {
tracing::warn!(envelope = %summary, "dropped an envelope: the relay is gone")
}
}
}
@@ -655,4 +666,55 @@ mod tests {
assert_eq!(session.await.unwrap(), Outcome::Shutdown);
}
#[tokio::test]
async fn a_relay_that_is_not_draining_does_not_stall_the_session() {
// Bounded, because the failure is a session that stops rather than one
// that answers wrongly.
tokio::time::timeout(std::time::Duration::from_secs(5), async {
a_backed_up_relay().await
})
.await
.expect("the session stalled on the relay");
}
async fn a_backed_up_relay() {
let (guest, host) = tokio::io::duplex(4096);
let mut caller = Caller::new(host);
let (down_tx, down_rx) = mpsc::channel(1);
let (_up_tx, up_rx) = mpsc::channel::<Payload>(1);
// Held and never read: a workload that is slow to read its own mail,
// or one that connected and stopped.
let _backed_up = down_rx;
let session = tokio::spawn(async move {
let mut ports = Ports {
to_workload: down_tx,
from_workload: up_rx,
};
let mut workload = Double::exits_when_stopped(Exit::code(0));
run(guest, &mut workload, &mut ports).await.unwrap()
});
assert!(matches!(caller.expect().await, GuestToHost::Ready { .. }));
for _ in 0..8 {
caller
.say(&HostToGuest::Payload {
payload: Payload::new("identity", "backlog"),
})
.await;
}
// The lifecycle layer still moves: stop, shutdown and an exit are on
// this loop too, and none of them may wait on the relay.
caller
.say(&HostToGuest::Boot {
descriptor: Box::new(descriptor()),
})
.await;
caller.expect_started().await;
caller.say(&HostToGuest::Shutdown).await;
assert_eq!(session.await.unwrap(), Outcome::Shutdown);
}
}

53
apps/nesinit/src/workload.rs
View File

@@ -192,11 +192,15 @@ impl Workload for Process {
/// channel, so the guest still learns nothing about the filesystem it is being
/// handed a piece of.
fn mount_share(share: &Mount) -> Result<(), Failure> {
// Checked before anything is created: a descriptor this component cannot
// act on should leave no directory behind to confuse whoever reads the
// failure.
let (source, target, flags) = options(share)?;
// The mount point may not exist yet: a share can land anywhere the
// descriptor names, including a directory no image created.
std::fs::create_dir_all(&share.at).map_err(|error| failed(share, error))?;
let (source, target, flags) = options(share);
// SAFETY: mount takes two paths, a filesystem name and a flag word, all
// of which outlive the call, and no options string.
let mounted = unsafe {
@@ -220,7 +224,7 @@ const FSTYPE: &std::ffi::CStr = c"virtiofs";
/// What the mount call is given, split out because this is the part worth
/// asserting: mounting itself needs privileges a test does not have.
fn options(share: &Mount) -> (CString, CString, libc::c_ulong) {
fn options(share: &Mount) -> Result<(CString, CString, libc::c_ulong), Failure> {
// nosuid and nodev on every share, whether or not it is writable. A share
// is data handed to the guest; a setuid binary or a device node appearing
// in one is not something a workload should be able to use, and no
@@ -229,11 +233,23 @@ fn options(share: &Mount) -> (CString, CString, libc::c_ulong) {
if share.ro {
flags |= libc::MS_RDONLY;
}
// Interior nul bytes are the caller's mistake; a path with one cannot be
// mounted under any flags.
let source = CString::new(share.tag.as_str()).unwrap_or_default();
let target = CString::new(share.at.as_str()).unwrap_or_default();
(source, target, flags)
// A nul byte inside a tag or a path is a descriptor that cannot be carried
// out under any flags. Refused by name rather than silently emptied: an
// empty source turns up later as a mount failure about something else
// entirely, which is the wrong thing to go and look at.
let source = CString::new(share.tag.as_str()).map_err(|_| {
Failure::new(format!(
"the share tag contains a nul byte: {:?}",
share.tag
))
})?;
let target = CString::new(share.at.as_str()).map_err(|_| {
Failure::new(format!(
"the mount point contains a nul byte: {:?}",
share.at
))
})?;
Ok((source, target, flags))
}
/// A failure names the path, which is what makes it actionable: a permission
@@ -257,7 +273,7 @@ mod tests {
#[test]
fn a_writable_share_is_still_mounted_without_devices_or_setuid() {
let (source, target, flags) = options(&share(false));
let (source, target, flags) = options(&share(false)).unwrap();
assert_eq!(
source.to_str().unwrap(),
"user",
@@ -271,10 +287,29 @@ mod tests {
#[test]
fn a_read_only_share_is_mounted_read_only() {
let (_, _, flags) = options(&share(true));
let (_, _, flags) = options(&share(true)).unwrap();
assert_eq!(flags & libc::MS_RDONLY, libc::MS_RDONLY);
}
#[test]
fn a_descriptor_with_a_nul_byte_in_it_is_refused_by_name() {
let tagged = Mount {
tag: "us\0er".into(),
at: "/mnt/user".into(),
ro: false,
};
let failure = options(&tagged).expect_err("an empty source would have been mounted");
assert!(failure.reason.contains("tag"), "{}", failure.reason);
let placed = Mount {
tag: "user".into(),
at: "/mnt/us\0er".into(),
ro: false,
};
let failure = options(&placed).expect_err("an empty target would have been mounted");
assert!(failure.reason.contains("mount point"), "{}", failure.reason);
}
#[test]
fn a_failure_names_the_path_it_happened_on() {
let failure = failed(&share(false), io::Error::from_raw_os_error(libc::EACCES));