Close control connections on the first line that is not a request

A line that is not a JSON object with a string method, including an HTTP request line, bytes that are not UTF-8 or a line over 64 MiB, gets one error reply and the connection closes before anything after it is read. Request lines are read with a bound instead of buffering them whole.
2026-10-24 10:46:50 -06:00 · 2026-10-16 16:26:31 +00:00
parent f1894c6da6
commit 27932a2ae4
2 changed files with 186 additions and 34 deletions
@@ -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<Receiver<ControlRequest>> {
    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<Receiver<ControlRequest>> {
    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<Receiver<ControlR
            let _ = std::thread::Builder::new().name("control-conn".into()).spawn(move || serve(stream, tx, ctx));
        }
    })?;
-    log::info!("control channel listening on 127.0.0.1:{port}");
    Ok(rx)
 }

+/// Read one `\n`-terminated line of at most `max` bytes into `buf` (newline stripped).
+/// `Ok(false)` at end of stream; `Err` when the line is too long or the read fails.
+fn read_bounded_line(reader: &mut impl BufRead, buf: &mut Vec<u8>, max: usize) -> Result<bool, String> {
+    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::<Json>(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<ControlRequest>, 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<Json> {
+        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<ControlRequest>) {
+        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::<Json>(&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");
    }
 }
@@ -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 |