bxctl: escape error details, time out on admin.sock, AdminError::Io
M3a review findings 9, 10, 12, 13. A retrying error and every error detail can carry the inference server's body, so they are escaped like model text. Admin requests wait at most 30 s, so a stuck brokerd cannot hang bxctl or a chat turn. A failed write is AdminError::Io and stops handle_pending instead of being answered with another write. The usage line says what audit verify checks. Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
This commit is contained in:
@@ -12,10 +12,27 @@ use proto::{
|
|||||||
use crate::chat::code_name;
|
use crate::chat::code_name;
|
||||||
use crate::escape::escape_json_text;
|
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<Message, AdminError> {
|
pub fn request(socket: &Path, msg: Message) -> Result<Message, AdminError> {
|
||||||
|
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<Message, AdminError> {
|
||||||
let mut stream =
|
let mut stream =
|
||||||
UnixStream::connect(socket).map_err(|e| AdminError::Connect(socket.to_path_buf(), e))?;
|
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 {
|
let env = Envelope {
|
||||||
v: PROTOCOL_VERSION,
|
v: PROTOCOL_VERSION,
|
||||||
id: 1,
|
id: 1,
|
||||||
@@ -51,6 +68,8 @@ pub enum AdminError {
|
|||||||
Frame(FrameError),
|
Frame(FrameError),
|
||||||
Refused(WireError),
|
Refused(WireError),
|
||||||
Protocol(String),
|
Protocol(String),
|
||||||
|
/// Writing the output failed; the caller stops rather than write again.
|
||||||
|
Io(std::io::Error),
|
||||||
}
|
}
|
||||||
|
|
||||||
impl std::fmt::Display for AdminError {
|
impl std::fmt::Display for AdminError {
|
||||||
@@ -58,16 +77,20 @@ impl std::fmt::Display for AdminError {
|
|||||||
match self {
|
match self {
|
||||||
AdminError::Connect(p, e) => write!(f, "cannot reach brokerd at {}: {e}", p.display()),
|
AdminError::Connect(p, e) => write!(f, "cannot reach brokerd at {}: {e}", p.display()),
|
||||||
AdminError::Frame(e) => write!(f, "{e}"),
|
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::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<std::io::Error> for AdminError {
|
impl From<std::io::Error> for AdminError {
|
||||||
fn from(e: std::io::Error) -> Self {
|
fn from(e: std::io::Error) -> Self {
|
||||||
AdminError::Protocol(e.to_string())
|
AdminError::Io(e)
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|||||||
@@ -10,7 +10,7 @@ use proto::{
|
|||||||
};
|
};
|
||||||
|
|
||||||
use crate::admin;
|
use crate::admin;
|
||||||
use crate::escape::escape_model_text;
|
use crate::escape::{escape_json_text, escape_model_text};
|
||||||
|
|
||||||
#[derive(Debug)]
|
#[derive(Debug)]
|
||||||
pub enum ChatError {
|
pub enum ChatError {
|
||||||
@@ -25,7 +25,10 @@ impl std::fmt::Display for ChatError {
|
|||||||
match self {
|
match self {
|
||||||
ChatError::Connect(e) => write!(f, "{e}"),
|
ChatError::Connect(e) => write!(f, "{e}"),
|
||||||
ChatError::Frame(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}"),
|
ChatError::Protocol(s) => write!(f, "{s}"),
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
@@ -188,7 +191,8 @@ impl Printer {
|
|||||||
self.close_dimmed(out)?;
|
self.close_dimmed(out)?;
|
||||||
writeln!(
|
writeln!(
|
||||||
out,
|
out,
|
||||||
"[retrying: attempt {attempt} in {after_ms} ms: {error}]"
|
"[retrying: attempt {attempt} in {after_ms} ms: {}]",
|
||||||
|
escape_json_text(error)
|
||||||
)?;
|
)?;
|
||||||
}
|
}
|
||||||
TurnEvent::ThinkingCapped { tokens } => {
|
TurnEvent::ThinkingCapped { tokens } => {
|
||||||
@@ -302,6 +306,8 @@ pub fn handle_pending(
|
|||||||
};
|
};
|
||||||
match result {
|
match result {
|
||||||
Ok(_) => Ok(()),
|
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) => {
|
Err(e) => {
|
||||||
writeln!(out, "approval {approval}: {e}")?;
|
writeln!(out, "approval {approval}: {e}")?;
|
||||||
Ok(())
|
Ok(())
|
||||||
|
|||||||
@@ -25,7 +25,7 @@ usage: bxctl <command> [options]
|
|||||||
Check that the grants load.
|
Check that the grants load.
|
||||||
|
|
||||||
bxctl audit verify [--home DIR]
|
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
|
// 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`.
|
// socket the other subcommands use. Both default to the paths under `$BOXMAKER_HOME`.
|
||||||
|
|||||||
@@ -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);
|
||||||
|
}
|
||||||
@@ -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}");
|
||||||
|
}
|
||||||
|
}
|
||||||
Reference in New Issue
Block a user