diff --git a/CLAUDE.md b/CLAUDE.md index 9cd2ff6..8c4d248 100644 --- a/CLAUDE.md +++ b/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. 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 9fcbcc0..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}; @@ -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"), + } +} 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) { 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");