fix: nonblocking delegated fds and correct error logging

CLAUDE.md

@@ -111,6 +111,15 @@ 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.
+- 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.

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<i32> {
     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<SessionOutcome> {
+    let mut failure: Option<String> = 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("<malformed error>")
-                );
-            }
-            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("<malformed error>")
+                        .to_owned()
+                });
             }
+            // Nothing else is meaningful here, and a peer is free to send
+            // messages this version does not know about.
+            _ => {}
         }
     }
 }

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.

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};
 
@@ -69,3 +71,130 @@ 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"
+    );
+}
+
+/// 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"),
+    }
+}

crates/capsudo-transport/src/mux.rs

@@ -401,6 +401,14 @@ fn spawn_blocking_reader(fd: OwnedFd, id: u32, frame_tx: mpsc::Sender<OutFrame>,
                         }
                     }
                     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<StreamMsg>) {
     });
 }
 
+/// 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<AsyncFd<OwnedFd>>, id: u32, frame_tx: mpsc::Sender<OutFrame>) {

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");