diff --git a/crates/bxctl/src/admin.rs b/crates/bxctl/src/admin.rs index 1165ef7..cde00b0 100644 --- a/crates/bxctl/src/admin.rs +++ b/crates/bxctl/src/admin.rs @@ -12,10 +12,27 @@ use proto::{ use crate::chat::code_name; use crate::escape::escape_json_text; -// Ask brokerd for one message and read one answer. +/// How long `bxctl` waits for `brokerd` to take or answer an admin request. +pub const ADMIN_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(30); + +// Ask brokerd for one message and read one answer, waiting at most `ADMIN_TIMEOUT`. pub fn request(socket: &Path, msg: Message) -> Result { + request_with_timeout(socket, msg, ADMIN_TIMEOUT) +} + +/// `request` with another limit on each read and write, so a `brokerd` that accepts and never +/// answers cannot hang `bxctl`, or a `chat` turn waiting on an approval. +pub fn request_with_timeout( + socket: &Path, + msg: Message, + timeout: std::time::Duration, +) -> Result { let mut stream = UnixStream::connect(socket).map_err(|e| AdminError::Connect(socket.to_path_buf(), e))?; + stream + .set_read_timeout(Some(timeout)) + .and_then(|()| stream.set_write_timeout(Some(timeout))) + .map_err(|e| AdminError::Connect(socket.to_path_buf(), e))?; let env = Envelope { v: PROTOCOL_VERSION, id: 1, @@ -51,6 +68,8 @@ pub enum AdminError { Frame(FrameError), Refused(WireError), Protocol(String), + /// Writing the output failed; the caller stops rather than write again. + Io(std::io::Error), } impl std::fmt::Display for AdminError { @@ -58,16 +77,20 @@ impl std::fmt::Display for AdminError { match self { AdminError::Connect(p, e) => write!(f, "cannot reach brokerd at {}: {e}", p.display()), AdminError::Frame(e) => write!(f, "{e}"), - AdminError::Refused(w) => write!(f, "{}: {}", code_name(w.code), w.detail), + // The detail may carry text from the inference server or a tool: escape it. + AdminError::Refused(w) => { + write!(f, "{}: {}", code_name(w.code), escape_json_text(&w.detail)) + } AdminError::Protocol(s) => write!(f, "{s}"), + AdminError::Io(e) => write!(f, "{e}"), } } } -// An error writing the output is reported like any other failure, with the message preserved. +// An error writing the output is its own kind, so the caller can stop instead of writing again. impl From for AdminError { fn from(e: std::io::Error) -> Self { - AdminError::Protocol(e.to_string()) + AdminError::Io(e) } } diff --git a/crates/bxctl/src/chat.rs b/crates/bxctl/src/chat.rs index 818994f..11548a6 100644 --- a/crates/bxctl/src/chat.rs +++ b/crates/bxctl/src/chat.rs @@ -10,7 +10,7 @@ use proto::{ }; use crate::admin; -use crate::escape::escape_model_text; +use crate::escape::{escape_json_text, escape_model_text}; #[derive(Debug)] pub enum ChatError { @@ -25,7 +25,10 @@ impl std::fmt::Display for ChatError { match self { ChatError::Connect(e) => write!(f, "{e}"), ChatError::Frame(e) => write!(f, "{e}"), - ChatError::Refused(w) => write!(f, "{}: {}", code_name(w.code), w.detail), + // The detail may carry the inference server's body: escape it like model text. + ChatError::Refused(w) => { + write!(f, "{}: {}", code_name(w.code), escape_json_text(&w.detail)) + } ChatError::Protocol(s) => write!(f, "{s}"), } } @@ -188,7 +191,8 @@ impl Printer { self.close_dimmed(out)?; writeln!( out, - "[retrying: attempt {attempt} in {after_ms} ms: {error}]" + "[retrying: attempt {attempt} in {after_ms} ms: {}]", + escape_json_text(error) )?; } TurnEvent::ThinkingCapped { tokens } => { @@ -302,6 +306,8 @@ pub fn handle_pending( }; match result { Ok(_) => Ok(()), + // The output itself failed: writing another line to it would fail the same way. + Err(admin::AdminError::Io(e)) => Err(e), Err(e) => { writeln!(out, "approval {approval}: {e}")?; Ok(()) diff --git a/crates/bxctl/src/cli.rs b/crates/bxctl/src/cli.rs index aae55c3..9f1948c 100644 --- a/crates/bxctl/src/cli.rs +++ b/crates/bxctl/src/cli.rs @@ -25,7 +25,7 @@ usage: bxctl [options] Check that the grants load. bxctl audit verify [--home DIR] - Verify the audit log against the grants."; + Check the audit log's hash chain, reading the files directly."; // The options every `chat` call carries. `socket` is the loop socket; `admin_socket` is the admin // socket the other subcommands use. Both default to the paths under `$BOXMAKER_HOME`. diff --git a/crates/bxctl/tests/admin_timeout.rs b/crates/bxctl/tests/admin_timeout.rs new file mode 100644 index 0000000..f716c0d --- /dev/null +++ b/crates/bxctl/tests/admin_timeout.rs @@ -0,0 +1,38 @@ +//! A `brokerd` that accepts and never answers does not hang `bxctl` (M3a review finding 9). + +use std::os::unix::net::UnixListener; +use std::time::{Duration, Instant}; + +use bxctl::admin::{ADMIN_TIMEOUT, request_with_timeout}; +use proto::{Empty, Message}; + +#[test] +fn the_default_is_thirty_seconds() { + assert_eq!(ADMIN_TIMEOUT, Duration::from_secs(30)); +} + +#[test] +fn a_silent_brokerd_is_an_error_after_the_timeout() { + let dir = std::env::temp_dir().join(format!("bx-admin-timeout-{}", std::process::id())); + let _ = std::fs::remove_dir_all(&dir); + std::fs::create_dir_all(&dir).unwrap(); + let socket = dir.join("admin.sock"); + let listener = UnixListener::bind(&socket).unwrap(); + let held = std::thread::spawn(move || { + let (stream, _) = listener.accept().unwrap(); + std::thread::sleep(Duration::from_millis(2_000)); + drop(stream); + }); + + let started = Instant::now(); + let got = request_with_timeout( + &socket, + Message::Approvals(Empty {}), + Duration::from_millis(200), + ); + let took = started.elapsed(); + assert!(got.is_err(), "{got:?}"); + assert!(took < Duration::from_millis(1_500), "waited {took:?}"); + held.join().unwrap(); + let _ = std::fs::remove_dir_all(&dir); +} diff --git a/crates/bxctl/tests/escape_details.rs b/crates/bxctl/tests/escape_details.rs new file mode 100644 index 0000000..587c993 --- /dev/null +++ b/crates/bxctl/tests/escape_details.rs @@ -0,0 +1,57 @@ +//! Text from `brokerd` or the inference server is escaped like model text before it reaches the +//! terminal: a `retrying` error and every error detail (M3a review finding 10). + +use bxctl::admin::AdminError; +use bxctl::chat::{ChatError, Printer}; +use proto::{ErrorCode, TurnEvent, WireError}; + +const ESC: char = '\u{1b}'; + +fn hostile() -> String { + format!("the server responded with 503: {ESC}[2J{ESC}]0;owned\u{7}\u{202e}") +} + +fn clean(text: &str) { + for c in text.chars() { + let cp = u32::from(c); + assert!( + cp >= 0x20 && !(0x7f..=0x9f).contains(&cp) && !(0x2028..=0x202e).contains(&cp) + || c == '\n', + "raw {cp:#x} in {text:?}" + ); + } +} + +#[test] +fn a_retrying_error_is_escaped() { + let mut out = Vec::new(); + let mut printer = Printer::new(true, false); + printer + .event( + &mut out, + &TurnEvent::Retrying { + attempt: 1, + after_ms: 2000, + error: hostile(), + }, + ) + .unwrap(); + let text = String::from_utf8(out).unwrap(); + clean(&text); + assert!(text.contains("\\u001b[2J"), "{text}"); +} + +#[test] +fn an_error_detail_is_escaped_in_chat_and_admin_errors() { + let wire = WireError { + code: ErrorCode::Inference, + detail: hostile(), + }; + for text in [ + ChatError::Refused(wire.clone()).to_string(), + AdminError::Refused(wire).to_string(), + ] { + clean(&text); + assert!(text.contains("\\u202e"), "{text}"); + } +}