From 90511840eb2935525faaa0c36c13ca1633d91c1e Mon Sep 17 00:00:00 2001 From: Ariadne Conill Date: Wed, 12 Aug 2026 16:06:57 -0700 Subject: [PATCH 1/2] fix(mux): tolerate non-blocking delegated descriptors A caller may delegate a descriptor it also drives with an async runtime, so it arrives here non-blocking -- and `dup` shares the open file description, so the sender-side pump inherits that whether it wants to or not. The pumps assumed otherwise. `write_all_blocking` treated `EAGAIN` as fatal, so the first time a peer wrote faster than the far end read, the writer gave up and dropped the rest of the stream with no error reported anywhere. The reader had a milder version: a readiness report is not a guarantee, and a bare `EAGAIN` after `poll` was taken for a broken descriptor and closed the stream. Now `EAGAIN` means wait and retry, which is what a blocking descriptor would have done for us. The writer waits for `POLLOUT`, treating a hangup as `EPIPE` so it cannot wait forever on a descriptor that will never drain; the reader goes back to waiting. Blocking threads stay: caller descriptors may be regular files or ttys, which epoll cannot watch. This only makes them indifferent to the mode of the descriptor they were handed. The test delegates a non-blocking socket as stdout and drains it late, so the buffer is full while the daemon still has output to deliver. Without the fix it receives exactly one socket buffer of a four megabyte payload. --- CLAUDE.md | 5 ++ crates/capsudo-core/tests/mux_session.rs | 74 ++++++++++++++++++++++++ crates/capsudo-transport/src/mux.rs | 39 +++++++++++++ 3 files changed, 118 insertions(+) diff --git a/CLAUDE.md b/CLAUDE.md index 9cd2ff6..95b2b51 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -111,6 +111,11 @@ messages (window-size updates) concurrently with a parked `recv()`. - Sender-side mux pumps use blocking threads (caller fds may be regular files or ttys, which epoll can't watch); receiver-side pumps are async on the socket pairs the transport creates. This asymmetry is intentional — see `mux.rs`. + The blocking pumps still handle a *non-blocking* descriptor: a caller may + delegate a socket it also drives with an async runtime, and `dup` shares the + open file description, so `O_NONBLOCK` comes with it. `EAGAIN` therefore means + "wait and retry", never "give up" — treating it as fatal silently truncated a + stream as soon as a peer outran the far end. - SELinux/`SO_PEERSEC` and a successful pwauth login can only be exercised on an SELinux host / with root + a real password; the rest is covered by `cargo test` and the binaries' smoke paths. diff --git a/crates/capsudo-core/tests/mux_session.rs b/crates/capsudo-core/tests/mux_session.rs index 9fcbcc0..56a24b4 100644 --- a/crates/capsudo-core/tests/mux_session.rs +++ b/crates/capsudo-core/tests/mux_session.rs @@ -69,3 +69,77 @@ async fn cat_round_trips_stdio_over_multiplexer() { assert_eq!(&read, b"hello\n"); } + +/// A caller may delegate descriptors it also drives with an async runtime, so +/// they arrive here non-blocking — and `dup` shares the open file description, +/// so the pump inherits that. Writing more than the descriptor can absorb must +/// therefore wait for room rather than give up: treating `EAGAIN` as fatal +/// truncated the stream silently, with no error anywhere. +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn output_survives_a_non_blocking_stdout_that_fills_up() { + // Comfortably more than a socket buffer, so the pump cannot possibly write + // it all in one go. + const PAYLOAD: usize = 4 * 1024 * 1024; + + let (client_io, server_io) = tokio::io::duplex(64 * 1024); + let mut server = MuxTransport::new(server_io, Side::Listener); + let mut client = MuxTransport::new(client_io, Side::Dialer); + + let server_task = tokio::spawn(async move { + serve_connection(&mut server, &DaemonConfig::default()) + .await + .unwrap(); + }); + + let (mut sink, stdout_w) = std::os::unix::net::UnixStream::pair().unwrap(); + stdout_w.set_nonblocking(true).unwrap(); + let stdin_r = std::fs::File::open("/dev/null").unwrap(); + let devnull = std::fs::OpenOptions::new() + .write(true) + .open("/dev/null") + .unwrap(); + + // Deliberately late: let the socket fill first, so the pump has to cope + // with a descriptor that cannot take another byte. + let drain = std::thread::spawn(move || { + std::thread::sleep(Duration::from_millis(250)); + let mut received = Vec::new(); + sink.read_to_end(&mut received).unwrap(); + received.len() + }); + + let request = ClientRequest { + args: vec![ + "head".to_owned(), + "-c".to_owned(), + PAYLOAD.to_string(), + "/dev/zero".to_owned(), + ], + env: Vec::new(), + session_type: SessionType::NonInteractive, + winsize: None, + }; + + let stdio = [stdin_r.as_fd(), stdout_w.as_fd(), devnull.as_fd()]; + let code = run_client(&mut client, &request, stdio).await.unwrap(); + assert_eq!(code, 0, "head should exit cleanly"); + server_task.await.unwrap(); + + // Close both ends of the write side — ours here, and the pump's `dup` when + // dropping the transport ends it — so the drain reaches EOF. + drop(client); + drop(stdout_w); + + let received = tokio::time::timeout( + Duration::from_secs(30), + tokio::task::spawn_blocking(move || drain.join().unwrap()), + ) + .await + .expect("draining stdout timed out") + .unwrap(); + + assert_eq!( + received, PAYLOAD, + "every byte should arrive; a short read means the pump gave up on a full descriptor" + ); +} diff --git a/crates/capsudo-transport/src/mux.rs b/crates/capsudo-transport/src/mux.rs index 26213bf..90fe84f 100644 --- a/crates/capsudo-transport/src/mux.rs +++ b/crates/capsudo-transport/src/mux.rs @@ -401,6 +401,14 @@ fn spawn_blocking_reader(fd: OwnedFd, id: u32, frame_tx: mpsc::Sender, } } Err(nix::errno::Errno::EINTR) => continue, + // The descriptor may be non-blocking: a caller can hand us + // one it also drives with an async runtime, and `dup` shares + // the open file description, so `O_NONBLOCK` comes with it. + // A readiness report is not a guarantee — the data can be + // gone by the time we read, e.g. consumed by another reader + // of the same description. Wait again rather than treating + // it as end of stream. + Err(nix::errno::Errno::EAGAIN) => continue, Err(_) => { let _ = frame_tx.blocking_send(OutFrame::StreamClose(id)); break; @@ -429,18 +437,49 @@ fn spawn_blocking_writer(fd: OwnedFd, mut inbound: mpsc::Receiver) { }); } +/// Writes every byte to `fd`, whether or not the descriptor is blocking. +/// +/// A caller may hand us a descriptor it also drives with an async runtime, and +/// `dup` shares the open file description, so `O_NONBLOCK` comes along with it. +/// Treating the resulting `EAGAIN` as fatal would silently truncate a stream the +/// first time a peer wrote faster than the far end read — so wait for writability +/// and carry on, which is what a blocking descriptor would have done for us. fn write_all_blocking(fd: &OwnedFd, mut bytes: &[u8]) -> nix::Result<()> { while !bytes.is_empty() { match nix::unistd::write(fd.as_fd(), bytes) { Ok(0) => return Err(nix::errno::Errno::EIO), Ok(n) => bytes = &bytes[n..], Err(nix::errno::Errno::EINTR) => continue, + Err(nix::errno::Errno::EAGAIN) => wait_writable(fd)?, Err(e) => return Err(e), } } Ok(()) } +/// Blocks until `fd` can accept more bytes, or the peer has gone. +fn wait_writable(fd: &OwnedFd) -> nix::Result<()> { + loop { + let mut poll_fds = [PollFd::new(fd.as_fd(), PollFlags::POLLOUT)]; + match poll(&mut poll_fds, PollTimeout::NONE) { + Ok(_) => {} + Err(nix::errno::Errno::EINTR) => continue, + Err(e) => return Err(e), + } + let Some(revents) = poll_fds[0].revents() else { + continue; + }; + // A hangup means nobody will ever read what is left; report it rather + // than waiting forever on a descriptor that will never drain. + if revents.intersects(PollFlags::POLLERR | PollFlags::POLLHUP | PollFlags::POLLNVAL) { + return Err(nix::errno::Errno::EPIPE); + } + if revents.intersects(PollFlags::POLLOUT) { + return Ok(()); + } + } +} + // ---- receiver-side pumps (async, used on socket pairs we create) ------------ async fn async_reader(fd: Arc>, id: u32, frame_tx: mpsc::Sender) { From cde2a8ff687c1892dff7dd027eab7309118bebaa Mon Sep 17 00:00:00 2001 From: Ariadne Conill Date: Wed, 12 Aug 2026 16:07:18 -0700 Subject: [PATCH 2/2] refactor(core): hand the daemon's error to the caller instead of printing The client printed a daemon's `Error` message to the process's stderr and then dropped it. For the `capsudo` binary that happened to be right; for anything embedding this library it is not. Protect's daemon serves object capabilities in-process, so a refused session printed a bare line outside its logging, unattributed to the capability that produced it, and the caller never saw the text at all. `SessionOutcome::Failed` now carries the message alongside the exit code. It is kept distinct from `Exited` because the two mean different things: one is a program that ran and failed, the other a program that never ran. `run_client`, which reports an exit status, turns it into the new `CoreError::Refused` for the same reason. Printing is the binary's job, and it still does it. Also stops printing on unexpected message types: a peer is free to send messages a given version does not know about, and saying so on stderr is noise, not diagnosis. Note for callers: `SessionOutcome` gained a variant, so exhaustive matches need an arm. --- CLAUDE.md | 4 ++ crates/capsudo-core/src/client.rs | 42 +++++++++++++---- crates/capsudo-core/src/error.rs | 4 ++ crates/capsudo-core/tests/mux_session.rs | 57 +++++++++++++++++++++++- crates/capsudo/src/main.rs | 6 +++ 5 files changed, 104 insertions(+), 9 deletions(-) diff --git a/CLAUDE.md b/CLAUDE.md index 95b2b51..8c4d248 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -116,6 +116,10 @@ messages (window-size updates) concurrently with a parked `recv()`. open file description, so `O_NONBLOCK` comes with it. `EAGAIN` therefore means "wait and retry", never "give up" — treating it as fatal silently truncated a stream as soon as a peer outran the far end. +- The client never prints. A daemon's `Error` message is carried back as + `SessionOutcome::Failed`, because an embedder (Protect's daemon) needs to log + and attribute it, not have it appear on the process's stderr. Printing it is + the `capsudo` binary's job. - SELinux/`SO_PEERSEC` and a successful pwauth login can only be exercised on an SELinux host / with root + a real password; the rest is covered by `cargo test` and the binaries' smoke paths. diff --git a/crates/capsudo-core/src/client.rs b/crates/capsudo-core/src/client.rs index 0f26542..a6812a0 100644 --- a/crates/capsudo-core/src/client.rs +++ b/crates/capsudo-core/src/client.rs @@ -37,6 +37,11 @@ pub struct ClientRequest { pub enum SessionOutcome { /// The program ran and exited with this status. Exited(i32), + /// The daemon reported why it would not run the program, and the session + /// ended with `code`. Kept distinct from [`SessionOutcome::Exited`] so a + /// caller can tell "the program ran and failed" from "the program never + /// ran", and can put the daemon's own words somewhere useful. + Failed { code: i32, message: String }, /// The daemon (or an authenticating front-end) requires a secret; the /// payload is the prompt to show the user. The caller should obtain a /// secret, reconnect, and retry with it. @@ -53,6 +58,9 @@ pub async fn run_client( ) -> Result { match run_session(transport, request, stdio, None).await? { SessionOutcome::Exited(code) => Ok(code), + // There is no exit status to report: the program never ran, and the + // daemon said why. + SessionOutcome::Failed { message, .. } => Err(CoreError::Refused(message)), SessionOutcome::Unauthorized(_) => Err(CoreError::Protocol("authentication required")), } } @@ -100,27 +108,45 @@ pub async fn run_session( } /// Waits for the daemon's terminal message (exit, or an auth challenge). +/// +/// An `Error` is not itself terminal — the daemon sends it and then an `Exit` — +/// so it is carried until the exit arrives and returned with it. This is a +/// library: writing the daemon's message to the process's stderr would put it +/// somewhere an embedder cannot capture, redirect, or attribute, and lose it +/// for the caller entirely. async fn await_outcome(transport: &mut dyn Transport) -> Result { + let mut failure: Option = None; loop { let Some(received) = transport.recv().await? else { return Err(CoreError::DaemonClosed); }; match received.message.field_type() { - FieldType::Exit => return Ok(SessionOutcome::Exited(received.message.as_i32()?)), + FieldType::Exit => { + let code = received.message.as_i32()?; + return Ok(match failure { + Some(message) => SessionOutcome::Failed { code, message }, + None => SessionOutcome::Exited(code), + }); + } FieldType::Unauthorized => { let prompt = received.message.as_str().unwrap_or("password: ").to_owned(); return Ok(SessionOutcome::Unauthorized(prompt)); } FieldType::Error => { - eprintln!( - "capsudo: error: {}", - received.message.as_str().unwrap_or("") - ); - } - other => { - eprintln!("capsudo: ignoring unexpected message {other:?}"); + // Keep the first: it is the one that describes why the session + // failed, and anything after it is likely a consequence. + failure.get_or_insert_with(|| { + received + .message + .as_str() + .unwrap_or("") + .to_owned() + }); } + // Nothing else is meaningful here, and a peer is free to send + // messages this version does not know about. + _ => {} } } } diff --git a/crates/capsudo-core/src/error.rs b/crates/capsudo-core/src/error.rs index 06df933..5c98b66 100644 --- a/crates/capsudo-core/src/error.rs +++ b/crates/capsudo-core/src/error.rs @@ -24,6 +24,10 @@ pub enum CoreError { /// The peer sent something that violates the protocol contract. #[error("protocol violation: {0}")] Protocol(&'static str), + + /// The daemon refused to run the program, in its own words. + #[error("{0}")] + Refused(String), } /// Convenience result alias for core operations. diff --git a/crates/capsudo-core/tests/mux_session.rs b/crates/capsudo-core/tests/mux_session.rs index 56a24b4..aa5cdd4 100644 --- a/crates/capsudo-core/tests/mux_session.rs +++ b/crates/capsudo-core/tests/mux_session.rs @@ -9,7 +9,9 @@ use std::io::{Read, Write}; use std::os::fd::AsFd; use std::time::Duration; -use capsudo_core::{run_client, serve_connection, ClientRequest, DaemonConfig}; +use capsudo_core::{ + run_client, run_session, serve_connection, ClientRequest, DaemonConfig, SessionOutcome, +}; use capsudo_proto::SessionType; use capsudo_transport::mux::{MuxTransport, Side}; @@ -143,3 +145,56 @@ async fn output_survives_a_non_blocking_stdout_that_fills_up() { "every byte should arrive; a short read means the pump gave up on a full descriptor" ); } + +/// The daemon's explanation for refusing a session belongs to the caller. The +/// client library used to print it to the process's stderr, where an embedder +/// could neither capture nor attribute it, and where the caller never saw it at +/// all. +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn a_refusal_carries_the_daemons_message_to_the_caller() { + let (client_io, server_io) = tokio::io::duplex(64 * 1024); + let mut server = MuxTransport::new(server_io, Side::Listener); + let mut client = MuxTransport::new(client_io, Side::Dialer); + + let server_task = tokio::spawn(async move { + serve_connection(&mut server, &DaemonConfig::default()) + .await + .unwrap(); + }); + + let (stdin_r, stdin_w) = std::io::pipe().unwrap(); + drop(stdin_w); + let devnull = std::fs::OpenOptions::new() + .write(true) + .open("/dev/null") + .unwrap(); + + let request = ClientRequest { + args: vec!["/nonexistent/program".to_owned()], + env: Vec::new(), + session_type: SessionType::NonInteractive, + winsize: None, + }; + + let stdio = [stdin_r.as_fd(), devnull.as_fd(), devnull.as_fd()]; + let outcome = run_session(&mut client, &request, stdio, None) + .await + .unwrap(); + server_task.await.unwrap(); + + match outcome { + SessionOutcome::Failed { code, message } => { + // The conventional "command not found" status, as `exit.rs` sends + // for a program that could not be started. + assert_eq!(code, 127); + assert!( + !message.is_empty(), + "the daemon's explanation should reach the caller, not stderr" + ); + } + SessionOutcome::Exited(_) => { + panic!("a program that cannot start must not look like one that ran") + } + SessionOutcome::Unauthorized(_) => panic!("no authentication was involved"), + } +} diff --git a/crates/capsudo/src/main.rs b/crates/capsudo/src/main.rs index 57dbd83..0961d54 100644 --- a/crates/capsudo/src/main.rs +++ b/crates/capsudo/src/main.rs @@ -172,6 +172,12 @@ async fn main() -> ExitCode { Ok(SessionOutcome::Exited(code)) => { return ExitCode::from((code & 0xff) as u8); } + // The library hands the daemon's explanation back rather than + // printing it, so telling the user is this binary's job. + Ok(SessionOutcome::Failed { code, message }) => { + eprintln!("capsudo: error: {message}"); + return ExitCode::from((code & 0xff) as u8); + } Ok(SessionOutcome::Unauthorized(prompt)) => { if secret.is_some() { eprintln!("capsudo: authentication failed");