From 21fc384e4089f4de9ce7fb563f2be07871d4a2af Mon Sep 17 00:00:00 2001 From: bherbruck Date: Mon, 2 Nov 2026 04:23:23 +0000 Subject: [PATCH] Control channel: a new connection has five seconds to show its token (#10961) A client could connect and send nothing, or only blank lines, and hold one of the 16 connection seats for as long as it liked. All reads before the token now share one five-second deadline; the connection is then closed without a reply. An authenticated connection may idle as before. `solidcraft-cli mcp --connect` used to open a connection at start and hold it until the first tool call. It now only checks that the app is there, and the first call connects. Co-authored-by: Claude Opus 5.5 --- apps/solidcraft/src/control_server.rs | 133 ++++++++++++++++++++++++-- crates/mcp/src/backend.rs | 8 +- crates/mcp/src/tests.rs | 33 +++++++ docs/control-protocol.md | 4 +- 4 files changed, 167 insertions(+), 11 deletions(-) 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