diff --git a/apps/solidcraft/src/control_server.rs b/apps/solidcraft/src/control_server.rs index b3a197edf3..996d7c54a9 100644 --- a/apps/solidcraft/src/control_server.rs +++ b/apps/solidcraft/src/control_server.rs @@ -22,6 +22,10 @@ const MAX_BAD_LINES: usize = 8; /// How long a subscribed connection waits after a `design.changed` line before the next: what /// changes meanwhile (a parameter dragged) goes out as one line, the latest (#6985). const EVENT_GAP: Duration = Duration::from_millis(100); +/// How long a new connection has to show the token. A client connects when it has a request to +/// send, so this is plenty; one that doesn't send it loses the connection, and its seat among +/// [`MAX_CONNECTIONS`] goes to someone else. +const TOKEN_WAIT: Duration = Duration::from_secs(5); static OPEN: AtomicUsize = AtomicUsize::new(0); /// The token files this launch wrote (its control port, the agent chat's endpoint) and the @@ -225,23 +229,53 @@ fn unauthorized() -> Value { /// Refuse the connection: the reply, then the end of what we send, then what the client had /// already sent read and dropped (a little, briefly). Closing a socket with unread input resets /// the connection, and a reset can reach the client before it has read the refusal. -fn refuse(out: &mut TcpStream, reader: &mut BufReader) { +fn refuse(out: &mut TcpStream, reader: &mut BufReader) { use std::io::Read; let _ = send(out, &unauthorized()); let _ = out.shutdown(std::net::Shutdown::Write); - let _ = reader.get_ref().set_read_timeout(Some(Duration::from_millis(500))); + reader.get_mut().deadline = Some(Instant::now() + Duration::from_millis(500)); let _ = std::io::copy(&mut reader.by_ref().take(64 * 1024), &mut std::io::sink()); } +/// What a client sends. Until it shows the token, all of its reads share one deadline, so neither +/// blank lines nor a line sent a byte at a time buy it more time. After that it may stay idle as +/// long as it likes. +struct ClientInput { + socket: TcpStream, + deadline: Option, +} + +impl ClientInput { + /// No deadline any more: the client showed the token. + fn admit(&mut self) -> std::io::Result<()> { + self.deadline = None; + self.socket.set_read_timeout(None) + } +} + +impl std::io::Read for ClientInput { + fn read(&mut self, buf: &mut [u8]) -> std::io::Result { + if let Some(deadline) = self.deadline { + let left = deadline.saturating_duration_since(Instant::now()); + if left.is_zero() { + return Err(std::io::ErrorKind::TimedOut.into()); + } + self.socket.set_read_timeout(Some(left))?; + } + self.socket.read(buf) + } +} + /// How long a request waits for the app's reply: most requests, and a batch (or every request, -/// on a patient connection). +/// on a patient connection); and how long a new connection has to show the token. #[derive(Clone, Copy)] struct Waits { short: Duration, long: Duration, + token: Duration, } -const WAITS: Waits = Waits { short: Duration::from_secs(60), long: Duration::from_secs(3600) }; +const WAITS: Waits = Waits { short: Duration::from_secs(60), long: Duration::from_secs(3600), token: TOKEN_WAIT }; /// A request that timed out but is still running in the app (#1618): its id and method, and where /// its reply will land. @@ -543,7 +577,7 @@ fn serve_with( let connection = running.connections.fetch_add(1, Ordering::Relaxed) + 1; let Ok(read) = stream.try_clone() else { return }; let mut out = stream; - let mut reader = BufReader::new(read); + let mut reader = BufReader::new(ClientInput { socket: read, deadline: Some(Instant::now() + waits.token) }); let mut bad = 0usize; let mut authenticated = false; let mut pending: Option = None; @@ -554,6 +588,10 @@ fn serve_with( let line = match read_line(&mut reader, MAX_LINE) { Ok(Some(l)) => String::from_utf8_lossy(&l).into_owned(), Ok(None) => break, + // Closed without a word: a client that connected early and only now sends its first + // request would read a line here as the reply to it. It finds the connection closed + // and opens another. + Err(e) if !authenticated && matches!(e.kind(), std::io::ErrorKind::TimedOut | std::io::ErrorKind::WouldBlock) => break, Err(e) => { let _ = answer(&mut out, events.as_ref(), &json!({"ok": false, "error": e.to_string()})); break; @@ -565,7 +603,12 @@ fn serve_with( // The first request must carry the token; later ones on the same connection may. if !authenticated { match token_of(&line) { - Some(t) if control_auth::tokens_match(&t, token) => authenticated = true, + Some(t) if control_auth::tokens_match(&t, token) => { + if reader.get_mut().admit().is_err() { + break; + } + authenticated = true; + } _ => { refuse(&mut out, &mut reader); break; @@ -675,6 +718,80 @@ fn serve_with( mod tests { use super::*; + /// A connection that doesn't show the token in time is closed without a word, however it fills + /// the time. One that does keeps working when it idles for longer than that. + #[test] + fn the_token_has_a_time_limit_and_an_authenticated_client_may_idle() { + use std::io::Read; + let token = "0123456789abcdef0123456789abcdef"; + let limit = Duration::from_millis(500); + let listener = TcpListener::bind("127.0.0.1:0").unwrap(); + let addr = listener.local_addr().unwrap(); + let (tx, rx) = channel::(); + std::thread::spawn(move || { + for req in rx { + let _ = req.reply.send(json!({"ok": true, "result": req.method})); + } + }); + static RUNNING: Running = Running::new(); + let waits = Waits { short: Duration::from_secs(10), long: Duration::from_secs(10), token: limit }; + std::thread::spawn(move || { + for stream in listener.incoming().flatten() { + let tx = tx.clone(); + std::thread::spawn(move || serve_with(stream, token, tx, egui::Context::default(), false, waits, &RUNNING, &DESIGN_WATCH)); + } + }); + let open = || { + let s = TcpStream::connect(addr).unwrap(); + s.set_read_timeout(Some(Duration::from_secs(20))).unwrap(); + s + }; + // Sends `drip` every 50 ms until the server closes; what it read, and when it closed. + let closed_after = |drip: &[u8]| { + let mut s = open(); + let start = Instant::now(); + s.set_read_timeout(Some(Duration::from_millis(50))).unwrap(); + let mut got = Vec::new(); + loop { + let mut buf = [0u8; 256]; + match s.read(&mut buf) { + Ok(0) => break, + Ok(n) => got.extend_from_slice(buf.get(..n).unwrap()), + Err(e) if matches!(e.kind(), std::io::ErrorKind::WouldBlock | std::io::ErrorKind::TimedOut) => { + assert!(start.elapsed() < Duration::from_secs(20), "still open"); + let _ = s.write_all(drip); + } + Err(_) => break, + } + } + (got, start.elapsed()) + }; + for drip in [&b""[..], b"\n", b"\r\n \n", b" ", b"{"] { + let (got, took) = closed_after(drip); + assert!(got.is_empty(), "{drip:?} was answered: {}", String::from_utf8_lossy(&got)); + assert!(took >= limit.saturating_sub(Duration::from_millis(50)), "{drip:?} closed after {took:?}"); + } + + let mut s = open(); + let mut lines = BufReader::new(s.try_clone().unwrap()); + let mut ask = |line: Value| -> Value { + writeln!(s, "{line}").unwrap(); + let mut l = String::new(); + lines.read_line(&mut l).unwrap(); + serde_json::from_str(&l).unwrap() + }; + assert_eq!(ask(json!({"id": 1, "method": "ui.view", "token": token}))["ok"], true); + std::thread::sleep(limit * 3); + assert_eq!(ask(json!({"id": 2, "method": "ui.view"}))["ok"], true, "idle for longer than the limit"); + + // A wrong token is still answered, at once. + let mut wrong = open(); + writeln!(wrong, "{}", json!({"id": 1, "method": "ui.view", "token": "nope"})).unwrap(); + let mut l = String::new(); + BufReader::new(wrong).read_line(&mut l).unwrap(); + assert!(l.contains("unauthorized"), "{l}"); + } + /// #1618: a request that outlasts its wait says it is still running; the connection keeps its /// reply for `control.pending`, and refuses other requests as `busy` until it lands, so none /// is queued to run unseen behind it. @@ -698,7 +815,7 @@ mod tests { let _ = req.reply.send(json!({"ok": true, "result": req.method})); } }); - let waits = Waits { short: Duration::from_millis(100), long: Duration::from_millis(100) }; + let waits = Waits { short: Duration::from_millis(100), long: Duration::from_millis(100), token: TOKEN_WAIT }; static RUNNING: Running = Running::new(); std::thread::spawn(move || { if let Some(stream) = listener.incoming().flatten().next() { @@ -764,7 +881,7 @@ mod tests { let _ = req.reply.send(json!({"ok": true, "result": req.method})); } }); - let waits = Waits { short: Duration::from_millis(100), long: Duration::from_secs(60) }; + let waits = Waits { short: Duration::from_millis(100), long: Duration::from_secs(60), token: TOKEN_WAIT }; std::thread::spawn(move || { for stream in listener.incoming().flatten() { let tx = tx.clone(); diff --git a/crates/mcp/src/backend.rs b/crates/mcp/src/backend.rs index 2ad8d52eca..a783853166 100644 --- a/crates/mcp/src/backend.rs +++ b/crates/mcp/src/backend.rs @@ -212,9 +212,13 @@ impl Remote { } } - /// Connect now, if the app is there (`new` does not); the error says why not. + /// See whether the app is there now (`new` does not); the error says why not. The connection + /// doesn't stay open: the app closes one that shows no token within a few seconds, and the + /// first call may be long away. That call connects again. pub fn try_connect(&mut self) -> std::io::Result<()> { - self.reconnect() + self.reconnect()?; + self.drop_conn(); + Ok(()) } /// Failing to reach the app at all is `NotConnected`, so [`Remote::explain`] can tell it from diff --git a/crates/mcp/src/tests.rs b/crates/mcp/src/tests.rs index 4a0e6cc0af..86832a3294 100644 --- a/crates/mcp/src/tests.rs +++ b/crates/mcp/src/tests.rs @@ -1754,6 +1754,39 @@ fn a_bridged_request_carries_the_agents_folders() { echo.join().unwrap(); } +/// A server started before its first tool call keeps no connection waiting (the app closes one +/// that shows no token in time), and a call on a connection the app has closed meanwhile goes out +/// on a new one without an error. +#[test] +fn a_bridge_holds_no_idle_connection_and_survives_one_closed_before_its_first_call() { + use std::io::{BufRead, Read, Write}; + let listener = std::net::TcpListener::bind("127.0.0.1:0").unwrap(); + let addr = listener.local_addr().unwrap().to_string(); + let app = std::thread::spawn(move || { + let (mut probe, _) = listener.accept().unwrap(); + probe.set_read_timeout(Some(std::time::Duration::from_secs(10))).unwrap(); + assert_eq!(probe.read(&mut [0u8; 1]).unwrap(), 0, "the probe closes at once and sends nothing"); + // The connection the second bridge makes first: closed unanswered, as a timed-out one is. + drop(listener.accept().unwrap()); + for _ in 0..2 { + let (stream, _) = listener.accept().unwrap(); + let mut line = String::new(); + std::io::BufReader::new(stream.try_clone().unwrap()).read_line(&mut line).unwrap(); + let req: Value = serde_json::from_str(&line).unwrap(); + writeln!(&stream, "{}", json!({"id": req["id"], "ok": true, "result": {"method": req["method"]}})).unwrap(); + } + }); + let token = "0123456789abcdef0123456789abcdef"; + let mut early = crate::Remote::new(&addr, Some(token)); + early.try_connect().unwrap(); + let mut idle = crate::Remote::connect_with_token(&addr, Some(token)).unwrap(); + let got = crate::Backend::call(&mut early, "ui.view", json!({})).unwrap(); + assert_eq!(got["method"], "ui.view", "{got}"); + let got = crate::Backend::call(&mut idle, "ui.view", json!({})).unwrap(); + assert_eq!(got["method"], "ui.view", "{got}"); + app.join().unwrap(); +} + /// #7181, bridged: once the agent has saved a design through the app, that file's folder goes /// with every later request as `design_folders`, so the app counts it after the design moves on. /// Not before the save succeeded, not for one the app refused, not for a relative path (the app diff --git a/docs/control-protocol.md b/docs/control-protocol.md index 0488f6f31d..63bee380dd 100644 --- a/docs/control-protocol.md +++ b/docs/control-protocol.md @@ -99,7 +99,9 @@ The first request on a connection must carry the token as a top-level `"token"` (`{"id": 1, "method": "ui.inspect", "params": {}, "token": "…"}`); later requests on the same connection may leave it out. A first line without the right token is answered with `{"ok": false, "error": "unauthorized: …"}` and the connection is closed, without running -anything. Tokens are compared in constant time. +anything. Tokens are compared in constant time. A connection that hasn't sent that first request +within five seconds is closed without a reply, and blank lines don't extend the time. So connect +when you have something to send; once authenticated, a connection may stay idle. `solidcraft-cli mcp --connect 127.0.0.1:PORT` reads the token file for that port by itself (`--token T` or `SOLIDCRAFT_CONTROL_TOKEN` give one explicitly). For a test harness, or a