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

feat(nesinit): mount what the descriptor names, and relay the layer it cannot read (#320)

Browse Source 
Stacked on #319, which has PID 1, the channel and the trait but mounts
nothing.
Review that one first; this PR is the descriptor half.

- **The shares are mounted.** A tag names an export, the descriptor
names where
it lands, and every share goes on `nosuid` and `nodev` whether or not it
is
writable — a share is data handed to the guest, and no descriptor has a
way
  to ask for a setuid binary or a device node in one. Mounting needs
privileges a test does not have, so the arguments and flags are derived
by a
  function the tests assert; that is where the read-only decision lives.
- **Progress is two messages, not one.** `mounted` / `mount_failed` stay
apart
from `started` / `start_failed`, because a share that did not appear and
a
command that did not run want different things looked at. A failure
carries
the reason the operating system gave, verbatim, and the path it happened
on.
- **The second layer is relayed and never read.** Envelopes cross a unix
socket
to the workload and come back the same way. `body` is a string rather
than
  nested JSON on purpose: a document this component can index into is a
document it can grow to depend on, and then the layer is not opaque any
more
  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 a byte count are the whole of what may be
said
  about one. `Payload`'s `Debug` is written by hand for the same reason.
- **A write to a channel nobody reads now ends the session** the same
way a
closed read does, and stops the workload. A caller that stopped
listening has
  also stopped being able to say stop; that was two outcomes and is one.

## The tests, failing first

Progress reporting, with the mount result dropped (the state this branch
started from):

```
running 11 tests
test session::tests::an_unreadable_line_does_not_end_a_session ... ok
test session::tests::a_stop_is_idempotent_and_does_not_end_the_session ... ok
test session::tests::the_guest_speaks_first_and_says_its_version ... ok
test session::tests::a_closed_channel_stops_the_workload ... ok
test session::tests::the_descriptor_mounts_and_starts_what_it_names ... ok
test session::tests::an_envelope_crosses_the_session_in_both_directions_unread ... ok
test session::tests::a_share_that_will_not_mount_is_refused_before_anything_starts ... ok
test session::tests::an_exit_is_reported_and_the_workload_is_not_started_again ... FAILED
test session::tests::a_command_that_will_not_run_is_reported_apart_from_a_share_that_will_not_mount ... FAILED
test session::tests::a_signalled_workload_is_reported_as_signalled ... FAILED
test session::tests::a_relay_nothing_is_on_does_not_end_a_session ... FAILED

failures:

---- session::tests::an_exit_is_reported_and_the_workload_is_not_started_again stdout ----

thread 'session::tests::an_exit_is_reported_and_the_workload_is_not_started_again' (312969) panicked at apps/nesinit/src/session.rs:248:13:
assertion `left == right` failed
  left: Started
 right: Mounted

---- session::tests::a_command_that_will_not_run_is_reported_apart_from_a_share_that_will_not_mount stdout ----

thread 'session::tests::a_command_that_will_not_run_is_reported_apart_from_a_share_that_will_not_mount' (312963) panicked at apps/nesinit/src/session.rs:460:9:
assertion `left == right` failed
  left: StartFailed { reason: "ENOENT: /usr/bin/workload" }
 right: Mounted
note: run with `RUST_BACKTRACE=1` environment variable to display a backtrace

---- session::tests::a_signalled_workload_is_reported_as_signalled stdout ----

thread 'session::tests::a_signalled_workload_is_reported_as_signalled' (312966) panicked at apps/nesinit/src/session.rs:248:13:
assertion `left == right` failed
  left: Started
 right: Mounted

---- session::tests::a_relay_nothing_is_on_does_not_end_a_session stdout ----

thread 'session::tests::a_relay_nothing_is_on_does_not_end_a_session' (312964) panicked at apps/nesinit/src/session.rs:248:13:
assertion `left == right` failed
  left: Started
 right: Mounted

failures:
    session::tests::a_command_that_will_not_run_is_reported_apart_from_a_share_that_will_not_mount
    session::tests::a_relay_nothing_is_on_does_not_end_a_session
    session::tests::a_signalled_workload_is_reported_as_signalled
    session::tests::an_exit_is_reported_and_the_workload_is_not_started_again
```

The read-only flag, ignored:

```
running 9 tests
test shutdown::tests::a_workload_that_leaves_in_time_is_not_killed ... ok
test shutdown::tests::the_workload_stops_before_anything_else_and_the_disks_flush_before_power ... ok
test shutdown::tests::a_workload_that_overstays_its_grace_is_killed_and_shutdown_still_finishes ... ok
test workload::tests::a_failure_names_the_path_it_happened_on ... ok
test workload::tests::a_writable_share_is_still_mounted_without_devices_or_setuid ... ok
test workload::tests::a_read_only_share_is_mounted_read_only ... FAILED
test session::tests::a_closed_channel_stops_the_workload ... ok
test session::tests::an_exit_is_reported_and_the_workload_is_not_started_again ... ok
test session::tests::a_signalled_workload_is_reported_as_signalled ... ok

failures:

---- workload::tests::a_read_only_share_is_mounted_read_only stdout ----

thread 'workload::tests::a_read_only_share_is_mounted_read_only' (311963) panicked at apps/nesinit/src/workload.rs:212:9:
assertion `left == right` failed
  left: 0
 right: 1
note: run with `RUST_BACKTRACE=1` environment variable to display a backtrace

failures:
    workload::tests::a_read_only_share_is_mounted_read_only

```

The relay quoting a line it could not decode — which is the second way a
body
reaches a log line, and the reason the failure path logs a length and
nothing
else:

```
running 2 tests
test payload::tests::an_envelope_crosses_in_both_directions_untouched ... ok
test payload::tests::nothing_the_relay_logs_contains_a_body ... FAILED

failures:

---- payload::tests::nothing_the_relay_logs_contains_a_body stdout ----

thread 'payload::tests::nothing_the_relay_logs_contains_a_body' (312716) panicked at apps/nesinit/src/payload.rs:202:9:
a body reached a log line:
2026-09-04T21:11:56.619025Z  WARN nesinit::payload: ignoring an envelope that would not decode bytes=62 line={"channel":"identity","body":"a-credential-nobody-should-read"

note: run with `RUST_BACKTRACE=1` environment variable to display a backtrace

failures:
    payload::tests::nothing_the_relay_logs_contains_a_body

test result: FAILED. 1 passed; 1 failed; 0 ignored; 0 measured; 17 filtered out; finished in 0.06s

```

And the first way: a derived `Debug` instead of the hand-written one.

```
running 6 tests
test lifecycle::tests::a_mount_failure_keeps_its_reason_verbatim ... ok
test lifecycle::tests::a_signalled_exit_is_not_a_zero_exit ... ok
test lifecycle::tests::a_line_round_trips ... ok
test lifecycle::tests::defaults_cover_what_a_caller_may_leave_out ... ok
test lifecycle::tests::an_envelope_does_not_print_its_body ... FAILED
test lifecycle::tests::an_envelope_body_stays_a_string_in_both_directions ... ok

failures:

---- lifecycle::tests::an_envelope_does_not_print_its_body stdout ----

thread 'lifecycle::tests::an_envelope_does_not_print_its_body' (313791) panicked at crates/nesprotocol/src/lifecycle.rs:290:9:
the body reached a log line: Payload { channel: "identity", body: "a-credential-nobody-should-read" }
note: run with `RUST_BACKTRACE=1` environment variable to display a backtrace

failures:
    lifecycle::tests::an_envelope_does_not_print_its_body

test result: FAILED. 5 passed; 1 failed; 0 ignored; 0 measured; 9 filtered out; finished in 0.00s

error: test failed, to rerun pass `-p nesprotocol --lib`
```

All green after: 19 unit tests, 3 against real forked children, 15 in
`nesprotocol`.

## What this does not verify

- **Nothing has been mounted.** No `virtiofs` share has been mounted by
this
code, in a VM or anywhere else. What is tested is the source, target and
flag
word handed to the mount call; that the call succeeds against a real
virtio
transport, that the mount point is where a workload then finds its
files, and
that a `uid` mismatch surfaces as the permission error this is written
to
  produce, are all unverified.
- **The relay has never carried a real workload's traffic.** Two
processes, a
real unix socket and bytes that come back unchanged is what the test
shows.
Whether the socket path is the right mechanism is openly a guess, and it
is
  meant to be replaceable without anything above it moving.
- **"Never logged" is enforced more narrowly than it reads.** The
hand-written
`Debug` and the failure path's length-only line are both tested. The
capture
test cannot reliably see debug-level lines: callsite interest is cached
process-wide, so a line another test in the same binary reached first
never
arrives in the capture. A future `{:?}` on a whole envelope at debug
level
would not necessarily be caught by these tests, only by the `Debug` impl
  keeping its shape.
- **`geometry` is parsed and carried, and nothing consumes it.** This
component
does not start the guest's own services yet. `ticket` exists as a
message
  with no producer wired to it.
- **Still no VM, still no vsock, still no `uid` drop**, as in #319, and
no
  number in this PR is measured.
- **No third-party workload has gone through any of this.** The claim
that a
descriptor plus a set of shares is enough to run something we did not
write
is untested, and our own workload is the weakest possible witness for
it.

## Since review

`d4d473f` — the relay may not stall the session and may not buffer
without end,
plus a descriptor with a nul byte in it is refused by name. Failing
first, in
order:

The session held still behind a workload that was not reading:

```
running 1 test
test session::tests::a_relay_that_is_not_draining_does_not_stall_the_session ... FAILED

failures:

---- session::tests::a_relay_that_is_not_draining_does_not_stall_the_session stdout ----

thread 'session::tests::a_relay_that_is_not_draining_does_not_stall_the_session' (326881) panicked at apps/nesinit/src/session.rs:670:10:
the session stalled on the relay: Elapsed(())
note: run with `RUST_BACKTRACE=1` environment variable to display a backtrace

failures:
    session::tests::a_relay_that_is_not_draining_does_not_stall_the_session

test result: FAILED. 0 passed; 1 failed; 0 ignored; 0 measured; 22 filtered out; finished in 5.00s

error: test failed, to rerun pass `-p nesinit --lib`
```

A frame with no end to it:

```
running 1 test
test payload::tests::a_frame_that_never_ends_costs_the_connection_and_not_the_guest ... FAILED

failures:

---- payload::tests::a_frame_that_never_ends_costs_the_connection_and_not_the_guest stdout ----

thread 'payload::tests::a_frame_that_never_ends_costs_the_connection_and_not_the_guest' (327888) panicked at apps/nesinit/src/payload.rs:375:10:
the relay is still assembling a frame that never ends: Elapsed(())
note: run with `RUST_BACKTRACE=1` environment variable to display a backtrace

failures:
    payload::tests::a_frame_that_never_ends_costs_the_connection_and_not_the_guest

test result: FAILED. 0 passed; 1 failed; 0 ignored; 0 measured; 22 filtered out; finished in 5.01s

error: test failed, to rerun pass `-p nesinit --lib`
```

And a tag that was quietly emptied instead of refused:

```
running 1 test
test workload::tests::a_descriptor_with_a_nul_byte_in_it_is_refused_by_name ... FAILED

failures:

---- workload::tests::a_descriptor_with_a_nul_byte_in_it_is_refused_by_name stdout ----

thread 'workload::tests::a_descriptor_with_a_nul_byte_in_it_is_refused_by_name' (327380) panicked at apps/nesinit/src/workload.rs:274:40:
an empty source would have been mounted: ("", "/mnt/user", 6)
note: run with `RUST_BACKTRACE=1` environment variable to display a backtrace

failures:
    workload::tests::a_descriptor_with_a_nul_byte_in_it_is_refused_by_name

test result: FAILED. 0 passed; 1 failed; 0 ignored; 0 measured; 22 filtered out; finished in 0.00s

error: test failed, to rerun pass `-p nesinit --lib`
```

One behaviour changed rather than only hardened, and it is worth a
reviewer's
eye: **nothing is queued for a workload that is not on the relay.** An
envelope
that arrives with nobody connected is dropped, as is one that arrives
faster
than the workload reads. That follows the layer's own rule — what
crosses it is
re-sent when it changes, so a held copy is a stale copy — but it does
mean a
sender that assumes delivery is wrong to. Nothing here retries, and
nothing
tells the far end that a particular envelope was dropped.

23 unit tests, 5 against real forked children, 15 in `nesprotocol`.

<!-- greptile_comment -->

<h3>Greptile Summary</h3>

This PR mounts descriptor-defined virtiofs shares, reports mount and
process-start progress independently, and relays opaque payload
envelopes between the host channel and workload Unix socket. Changes
since the previous review also bound relay frames, prevent relay
backpressure from stalling lifecycle handling, reject descriptor strings
containing NUL bytes, and track whether a reaped PID remains valid.
- Mounts shares at descriptor-selected targets with `nosuid`, `nodev`,
and optional read-only flags.
- Adds bidirectional opaque payload forwarding with bounded,
non-blocking queues and body-safe logging.
- Adds distinct mounted/start lifecycle responses and failure reporting.
- Adds capped newline-delimited relay frames and drops stale or
backpressured envelopes.
- Reworks workload tracking to avoid signaling a PID after its exit has
been delivered.

<h3>Confidence Score: 5/5</h3>

The reviewed changes appear safe to merge, with no accepted new findings
or outstanding previous root-thread findings.

The resolved relay-stall, unbounded-frame, and invalid-NUL findings are
addressed by non-blocking delivery, capped frame assembly, and explicit
descriptor validation. The protocol-version concern was correctly
withdrawn under the coordinated version-2 deployment model. The
remaining PID check-to-signal race duplicates an existing prior comment
and therefore is not reposted or counted as a new finding.

<h3>Important Files Changed</h3>

| Filename | Overview |
|----------|----------|
| apps/nesinit/src/payload.rs | Adds the bounded, bidirectional
Unix-socket payload relay with non-blocking delivery and body-safe
logging. |
| apps/nesinit/src/session.rs | Integrates payload events with lifecycle
handling and separately reports mount and process-start outcomes. |
| apps/nesinit/src/workload.rs | Implements descriptor-driven virtiofs
mounts and switches process signaling to tracked reaper state. |
| apps/nesinit/src/reap.rs | Adds shared reaped-state tracking so
callers stop treating a delivered PID as the workload. |
| crates/nesprotocol/src/lifecycle.rs | Extends lifecycle messages with
mount progress and opaque payload envelopes while redacting payload
bodies from Debug output. |
| apps/nesinit/src/main.rs | Starts the payload relay before the
workload session and wires bounded relay ports into session handling. |
| apps/nesinit/README.md | Documents mount behavior, payload opacity,
delivery semantics, frame limits, and progress reporting. |

<h3>Sequence Diagram</h3>

```mermaid
sequenceDiagram
    participant H as Host
    participant N as nesinit
    participant M as virtiofs mounts
    participant W as Workload
    H->>N: Boot descriptor
    N->>M: Mount descriptor shares
    M-->>N: Success or failure
    N-->>H: mounted / mount_failed
    N->>W: Start command
    N-->>H: started / start_failed
    H->>N: Payload envelope
    N-->>W: Non-blocking Unix-socket relay
    W->>N: Payload envelope
    N-->>H: Payload envelope
    W-->>N: Exit
    N-->>H: workload_exited
```

<sub>Reviews (3): Last reviewed commit: ["fix(nesinit): the relay may
not stall
th..."](e94ea00593)
| [Re-trigger
Greptile](https://app.greptile.com/api/retrigger?id=60585116)</sub>

**Context used:**

- Knowledge Base — [Streaming appliance
build](https://app.greptile.com/nestri/-/custom-context/knowledge-base/nestrilabs/nestri/-/docs/appliance-build.md)

<!-- /greptile_comment -->
This commit is contained in:
KAAL1 (Bingus)
2026-09-05 10:08:36 +03:00
committed by GitHub
parent 217e82a9a8 a94b323edb
commit 2faf7d77db
7 changed files with 1160 additions and 52 deletions
Download Patch File Download Diff File Expand all files Collapse all files

83
apps/nesinit/README.md
View File

@@ -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,69 @@ 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": "<name>", "body": "<opaque string>" }
```
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.
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.
**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 +119,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 +129,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.

1
apps/nesinit/src/lib.rs
View File

@@ -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;

24
apps/nesinit/src/main.rs
View File

@@ -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<Outc
// waiting will not fix.
let channel = VsockStream::connect(address).await?;
// The relay is up before the workload is started, so a workload that
// dials in as its first act finds it there.
let (down_tx, down_rx) = tokio::sync::mpsc::channel(RELAY_DEPTH);
let (up_tx, up_rx) = tokio::sync::mpsc::channel(RELAY_DEPTH);
tokio::spawn(async move {
if let Err(error) = payload::serve(Path::new(payload::SOCKET), down_rx, up_tx).await {
tracing::error!(%error, "the relay is not running");
}
});
let mut ports = Ports {
to_workload: down_tx,
from_workload: up_rx,
};
let outcome = tokio::select! {
outcome = session::run(channel, workload) => outcome?,
outcome = session::run(channel, workload, &mut ports) => outcome?,
signal = asked_to_stop() => {
signal?;
tracing::info!("asked to stop");

442
apps/nesinit/src/payload.rs Normal file
View File

@@ -0,0 +1,442 @@
// 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::{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.
pub to_workload: Sender<Payload>,
/// Envelopes the workload sent.
pub from_workload: Receiver<Payload>,
}
/// 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<Payload>,
inbound: Sender<Payload>,
) -> 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 = 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"),
Err(error) => tracing::warn!(%error, "the relay connection ended"),
}
}
}
async fn relay(
stream: UnixStream,
outbound: &mut Receiver<Payload>,
inbound: &Sender<Payload>,
) -> io::Result<()> {
let (reader, mut writer) = stream.into_split();
let mut reader = BufReader::new(reader);
let mut frame = Vec::new();
loop {
tokio::select! {
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 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
}
}
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?;
}
}
}
}
/// 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 {
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<Mutex<Vec<u8>>>);
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<usize> {
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()
}
}
/// 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, outbound, inbound).await }
});
Self {
path,
down,
up,
server,
}
}
/// Dial the relay and wait until it is demonstrably carrying the
/// connection.
///
/// `serve` drops what arrives while nothing is connected, so a test
/// that sends downward before the accept has completed is racing it. A
/// line upward is the barrier: it can only arrive once the relay is
/// carrying this connection.
async fn workload(&mut self) -> (BufReader<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"#;
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");
assert_eq!(seen.channel, "identity");
let back = to_line(&Payload::new("identity", "opaque back")).unwrap();
writer.write_all(back.as_bytes()).await.unwrap();
assert_eq!(relay.up.recv().await.unwrap().body, "opaque back");
}
#[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 mut relay = Relay::start("logging");
let (reader, mut writer) = relay.workload().await;
let mut lines = reader.lines();
let secret = "a-credential-nobody-should-read";
relay
.down
.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();
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 —
// 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}"
);
}
#[tokio::test]
async fn an_envelope_sent_while_nobody_is_connected_is_dropped_rather_than_queued() {
let mut relay = Relay::start("nobody-home");
// Sent into a relay nothing is on. It is not held for whoever connects
// next: what crosses this layer is re-sent when it changes, so a held
// copy is a stale copy — and a queue that fills is what would
// eventually stall the session filling it.
relay
.down
.send(Payload::new("identity", "stale"))
.await
.unwrap();
let (reader, _writer) = relay.workload().await;
let mut lines = reader.lines();
relay
.down
.send(Payload::new("identity", "current"))
.await
.unwrap();
let line = lines.next_line().await.unwrap().unwrap();
let seen: Payload = from_line(&line).unwrap();
assert_eq!(
seen.body, "current",
"a stale envelope was delivered on connect"
);
}
#[tokio::test]
async fn a_frame_that_never_ends_costs_the_connection_and_not_the_guest() {
use tokio::io::AsyncReadExt;
let mut relay = Relay::start("unbounded");
let (mut reader, mut writer) = relay.dial().await.into_split();
// Well past the cap, and not a newline in it. Without a limit the
// buffer holding this grows in the one process the kernel has been
// told not to kill, so what it takes comes out of everything else.
let flood = vec![b'a'; LONGEST_ENVELOPE + 4096];
// The write fails once the relay drops the connection, which is the
// outcome under test rather than a problem with it.
let _ = writer.write_all(&flood).await;
// The relay closes on us rather than keeping the buffer. Bounded,
// because the failure being guarded against is a relay that reads for
// as long as the workload writes.
let mut unread = Vec::new();
let closed = tokio::time::timeout(
std::time::Duration::from_secs(5),
reader.read_to_end(&mut unread),
)
.await
.expect("the relay is still assembling a frame that never ends");
assert_eq!(
closed.unwrap(),
0,
"the relay answered a frame it should have refused"
);
// And it is there for whoever connects next.
let (reader, mut writer) =
tokio::time::timeout(std::time::Duration::from_secs(5), relay.workload())
.await
.expect("the relay never came back for the next workload");
let mut lines = reader.lines();
writer
.write_all(
to_line(&Payload::new("identity", "after the flood"))
.unwrap()
.as_bytes(),
)
.await
.unwrap();
assert_eq!(relay.up.recv().await.unwrap().body, "after the flood");
relay
.down
.send(Payload::new("identity", "down"))
.await
.unwrap();
let line = lines.next_line().await.unwrap().unwrap();
assert_eq!(from_line::<Payload>(&line).unwrap().body, "down");
}
}

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

@@ -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<C, W>(channel: C, workload: &mut W) -> std::io::Result<Outcome>
pub async fn run<C, W>(
channel: C,
workload: &mut W,
payload: &mut Ports,
) -> std::io::Result<Outcome>
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<C, W>(
channel: C,
workload: &mut W,
payload: &mut Ports,
) -> std::io::Result<Outcome>
where
C: AsyncRead + AsyncWrite,
W: Workload,
@@ -53,19 +89,39 @@ where
)
.await?;
let mut running: Option<crate::workload::Exited> = None;
let mut running: Option<Exited> = 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?;
ended = exited => Event::Ended(ended?),
line = lines.next_line() => Event::Line(line?),
up = payload.from_workload.recv(), if relay_open => Event::FromWorkload(up),
},
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));
}
line = lines.next_line() => line?,
},
None => lines.next_line().await?,
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,76 @@ 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),
HostToGuest::Stop => workload.signal_stop(),
HostToGuest::Shutdown => return Ok(Outcome::Shutdown),
}
}
}
/// Carry out a descriptor: shares first, then the command.
/// What the session is waiting on, and there are only three things.
enum Event {
Line(Option<String>),
Ended(Exit),
FromWorkload(Option<Payload>),
}
/// 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.
///
/// 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<W: Workload>(
descriptor: &BootDescriptor,
workload: &mut W,
) -> Result<crate::workload::Exited, Failure> {
workload.mount(&descriptor.mounts)?;
workload.start(&descriptor.exec)
/// 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")
}
}
}
async fn send<W>(writer: &mut W, message: &GuestToHost) -> std::io::Result<()>
@@ -132,8 +233,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 +261,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<Payload>, mpsc::Sender<Payload>) {
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<BufReader<DuplexStream>>,
@@ -181,6 +298,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 +320,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 +345,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 +371,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 +383,7 @@ mod tests {
descriptor: Box::new(descriptor()),
})
.await;
caller.expect_started().await;
assert_eq!(
caller.expect().await,
@@ -280,8 +407,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 +418,7 @@ mod tests {
descriptor: Box::new(descriptor()),
})
.await;
caller.expect_started().await;
assert_eq!(
caller.expect().await,
@@ -309,8 +438,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 +463,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 +477,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 +499,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 +522,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 +546,175 @@ 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::<Payload>(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);
}
#[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);
}
}

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

@@ -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<Exited, Failure> {
@@ -188,6 +186,146 @@ 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> {
// 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))?;
// 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) -> 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
// descriptor has a way to ask for it.
let mut flags = libc::MS_NOSUID | libc::MS_NODEV;
if share.ro {
flags |= libc::MS_RDONLY;
}
// 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
/// 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)).unwrap();
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)).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));
assert!(
failure.reason.starts_with("/mnt/user: "),
"{}",
failure.reason
);
assert!(
failure.reason.contains("ermission denied"),
"{}",
failure.reason
);
}
}
#[cfg(test)]
pub mod double {
use super::*;

126
crates/nesprotocol/src/lifecycle.rs
View File

@@ -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<String>, body: impl Into<String>) -> 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},