diff --git a/apps/gridcraft/src/control_server.rs b/apps/gridcraft/src/control_server.rs index e9e05a2e6..4c57ccf41 100644 --- a/apps/gridcraft/src/control_server.rs +++ b/apps/gridcraft/src/control_server.rs @@ -1,7 +1,11 @@ //! Localhost JSON-lines control server. Each connection may send many requests; each line //! `{"id":…, "method":…, "params":…}` gets one reply line `{"id":…, "ok":…, …}`. +//! +//! The server only reads requests: a line that is not a JSON object with a string `method` (an +//! HTTP request line, a stray `nc`, binary data) gets one error reply and the connection is +//! closed, so nothing sent after it runs. -use std::io::{BufRead, BufReader, Write}; +use std::io::{BufRead, BufReader, Read, Write}; use std::net::{TcpListener, TcpStream}; use std::sync::mpsc::{Receiver, Sender, channel}; use std::time::Duration; @@ -13,6 +17,13 @@ const MAX_LINE: usize = 64 * 1024 * 1024; pub fn start(port: u16, ctx: egui::Context) -> std::io::Result> { let listener = TcpListener::bind(("127.0.0.1", port))?; + let rx = serve_listener(listener, ctx)?; + log::info!("control channel listening on 127.0.0.1:{port}"); + Ok(rx) +} + +/// Accept connections on `listener`; requests arrive on the returned receiver. +fn serve_listener(listener: TcpListener, ctx: egui::Context) -> std::io::Result> { let (tx, rx) = channel(); std::thread::Builder::new().name("control".into()).spawn(move || { for stream in listener.incoming().flatten() { @@ -21,47 +32,184 @@ pub fn start(port: u16, ctx: egui::Context) -> std::io::Result, max: usize) -> Result { + buf.clear(); + // `take` bounds the bytes buffered for one line: a peer can't make us allocate more than `max`. + let limit = u64::try_from(max).unwrap_or(u64::MAX).saturating_add(1); + let n = reader.take(limit).read_until(b'\n', buf).map_err(|e| e.to_string())?; + if n == 0 { + return Ok(false); + } + if buf.last() == Some(&b'\n') { + buf.pop(); + return Ok(true); + } + if buf.len() > max { + return Err(format!("request too large (over {max} bytes); closing the connection")); + } + // Last line without a newline, then end of stream. + Ok(true) +} + +/// One request line, parsed. +enum Line { + Request { + id: Json, + method: String, + params: Json, + }, + Blank, + /// Not a request: reply with this error and close the connection. + Reject(String), +} + +fn parse_line(bytes: &[u8]) -> Line { + let Ok(text) = std::str::from_utf8(bytes) else { + return Line::Reject("not a control request (invalid UTF-8); closing the connection".into()); + }; + let text = text.trim(); + if text.is_empty() { + return Line::Blank; + } + if is_http_request_line(text) { + return Line::Reject("HTTP is not served here: this port takes JSON lines; closing the connection".into()); + } + let msg = match serde_json::from_str::(text) { + Ok(v) => v, + Err(e) => return Line::Reject(format!("invalid JSON ({e}); expected one JSON request object per line; closing the connection")), + }; + let Some(obj) = msg.as_object() else { + return Line::Reject("not a control request (expected a JSON object); closing the connection".into()); + }; + let Some(method) = obj.get("method").and_then(Json::as_str) else { + return Line::Reject("not a control request (missing string `method`); closing the connection".into()); + }; + Line::Request { + id: obj.get("id").cloned().unwrap_or(Json::Null), + method: method.to_string(), + params: obj.get("params").cloned().unwrap_or(json!({})), + } +} + +/// An HTTP/1 request line (`POST / HTTP/1.1`) or the HTTP/2 preface (`PRI * HTTP/2.0`). +fn is_http_request_line(text: &str) -> bool { + const METHODS: [&str; 10] = ["GET", "HEAD", "POST", "PUT", "DELETE", "CONNECT", "OPTIONS", "TRACE", "PATCH", "PRI"]; + text.split_once(' ').is_some_and(|(method, _)| METHODS.contains(&method)) + || text.rsplit_once(' ').is_some_and(|(_, version)| version.starts_with("HTTP/")) +} + fn serve(stream: TcpStream, tx: Sender, ctx: egui::Context) { let Ok(read) = stream.try_clone() else { return }; let mut out = stream; - let reader = BufReader::new(read); - for line in reader.lines() { - let Ok(line) = line else { break }; - if line.trim().is_empty() { - continue; - } - if line.len() > MAX_LINE { - let _ = writeln!(out, "{}", json!({"ok": false, "error": "request too large"})); - continue; - } - let msg: Json = match serde_json::from_str(&line) { - Ok(v) => v, - Err(e) => { - let _ = writeln!(out, "{}", json!({"ok": false, "error": format!("invalid JSON: {e}")})); - continue; - } + let mut reader = BufReader::new(read); + let mut buf = Vec::new(); + loop { + let reject = match read_bounded_line(&mut reader, &mut buf, MAX_LINE) { + Ok(false) => break, + // The peer is gone before sending a line: nobody to tell. + Err(_) if buf.is_empty() => break, + Err(e) => e, + Ok(true) => match parse_line(&buf) { + Line::Blank => continue, + Line::Reject(e) => e, + Line::Request { id, method, params } => { + let (req, rx) = ControlRequest::new(method, params); + if tx.send(req).is_err() { + break; + } + // Wake the UI thread: egui only runs frames when something happens. + ctx.request_repaint(); + let mut reply = rx + .recv_timeout(Duration::from_secs(60)) + .unwrap_or_else(|_| json!({"ok": false, "error": "timed out waiting for the UI (is the window minimised?)"})); + if let Some(o) = reply.as_object_mut() { + o.insert("id".into(), id); + } + if writeln!(out, "{reply}").is_err() { + break; + } + continue; + } + }, }; - let id = msg.get("id").cloned().unwrap_or(Json::Null); - let method = msg.get("method").and_then(Json::as_str).unwrap_or("").to_string(); - let params = msg.get("params").cloned().unwrap_or(json!({})); - let (req, rx) = ControlRequest::new(method, params); - if tx.send(req).is_err() { - break; + // The connection closes either way; a peer that can't read the reason loses nothing. + let _ = writeln!(out, "{}", json!({"ok": false, "error": reject})); + break; + } +} + +#[cfg(test)] +mod tests { + use super::*; + + /// Send `input` on a new connection and read reply lines until the server closes it. + fn exchange(addr: std::net::SocketAddr, input: &[u8]) -> Vec { + let mut c = TcpStream::connect(addr).unwrap(); + c.set_read_timeout(Some(Duration::from_secs(5))).unwrap(); + c.write_all(input).unwrap(); + // A read timeout ends the list too: the server kept the connection open. + BufReader::new(c).lines().map_while(Result::ok).map(|l| serde_json::from_str(&l).unwrap()).collect() + } + + fn serve_on_loopback() -> (std::net::SocketAddr, Receiver) { + let listener = TcpListener::bind(("127.0.0.1", 0)).unwrap(); + let addr = listener.local_addr().unwrap(); + (addr, serve_listener(listener, egui::Context::default()).unwrap()) + } + + /// A browser can POST to the port: its request line closes the connection before the JSON + /// line in its body is read, so nothing is dispatched. + #[test] + fn an_http_request_closes_the_connection_before_its_body_is_dispatched() { + let (addr, rx) = serve_on_loopback(); + let http = b"POST / HTTP/1.1\r\nHost: 127.0.0.1:7979\r\nContent-Type: text/plain\r\n\r\n\ + {\"id\":1,\"method\":\"engine.execute\",\"params\":{\"command\":\"cell.set\",\"params\":{\"cell\":\"A1\",\"input\":\"x\"}}}\n"; + let replies = exchange(addr, http); + assert_eq!(replies.len(), 1, "{replies:?}"); + assert!(replies[0]["error"].as_str().unwrap().contains("closing the connection"), "{replies:?}"); + assert!(rx.try_recv().is_err(), "the body was dispatched"); + } + + #[test] + fn a_line_that_is_not_a_request_closes_the_connection() { + let (addr, rx) = serve_on_loopback(); + std::thread::spawn(move || { + for req in rx { + let _ = req.reply.send(json!({"ok": true})); + } + }); + let mut c = TcpStream::connect(addr).unwrap(); + c.set_read_timeout(Some(Duration::from_secs(5))).unwrap(); + c.write_all(b"{\"id\":1,\"method\":\"a\"}\r\n\n{\"id\":2,\"method\":\"b\"}\n").unwrap(); + let mut r = BufReader::new(c); + for id in [1, 2] { + let mut line = String::new(); + r.read_line(&mut line).unwrap(); + assert_eq!(serde_json::from_str::(&line).unwrap()["id"], id, "{line}"); } - // Wake the UI thread: egui only runs frames when something happens. - ctx.request_repaint(); - let mut reply = rx - .recv_timeout(Duration::from_secs(60)) - .unwrap_or_else(|_| json!({"ok": false, "error": "timed out waiting for the UI (is the window minimised?)"})); - if let Some(o) = reply.as_object_mut() { - o.insert("id".into(), id); + for junk in [&b"hello\n"[..], b"[1,2]\n", b"{\"id\":1}\n", b"{\"method\":5}\n", b"\xff\xfe\n", b"GET /x HTTP/1.1\n"] { + let mut input = junk.to_vec(); + input.extend_from_slice(b"{\"id\":2,\"method\":\"b\"}\n"); + let replies = exchange(addr, &input); + assert_eq!(replies.len(), 1, "{replies:?}"); + assert_eq!(replies[0]["ok"], false); } - if writeln!(out, "{reply}").is_err() { - break; - } + assert!(is_http_request_line("OPTIONS * HTTP/1.1")); + assert!(is_http_request_line("PRI * HTTP/2.0")); + assert!(!is_http_request_line("{\"method\":\"GET /\"}")); + } + + #[test] + fn an_overlong_line_is_refused_without_reading_the_rest() { + let mut buf = Vec::new(); + let long = [b'x'; 100]; + let mut input: &[u8] = &long; + assert!(read_bounded_line(&mut input, &mut buf, 10).is_err()); + assert_eq!(input.len(), 100 - 11, "the rest of the line is left unread"); } } diff --git a/docs/control-protocol.md b/docs/control-protocol.md index 7cc6ec230..e82ae7299 100644 --- a/docs/control-protocol.md +++ b/docs/control-protocol.md @@ -15,6 +15,10 @@ Then send newline-delimited JSON to `127.0.0.1:7979`. Every request line gets on Errors come back as `{"id": …, "ok": false, "error": "…"}`. +A line that is not a JSON object with a string `method` (an HTTP request, text that isn't JSON, +bytes that aren't UTF-8, a line over 64 MiB) gets one error reply and the server closes the +connection; nothing sent after it runs. Blank lines are skipped. + ## Methods | Method | Params | What it does |