From 339eb78a9bf7c955cba402738bdeb8e76ac92c9e Mon Sep 17 00:00:00 2001 From: Luskish <76147806+Luskish@users.noreply.github.com> Date: Tue, 6 Oct 2026 22:50:35 -0400 Subject: [PATCH] Control: close the connection on input that isn't a request A line that is not a JSON object with a string `method` (an HTTP request line, other non-JSON text, a JSON array, invalid UTF-8, a line over 4 MiB) now gets one error reply and the connection is closed, so a browser's cross-origin fetch can't smuggle a command in its body. Connections are capped at 16. Same transport as LightCraft (storytold/lightcraft#129) and FilmCraft (storytold/filmcraft#123). --- apps/vectorcraft/src/control_server.rs | 283 ++++++++++++++++++++++--- docs/control-protocol.md | 10 + 2 files changed, 261 insertions(+), 32 deletions(-) diff --git a/apps/vectorcraft/src/control_server.rs b/apps/vectorcraft/src/control_server.rs index f4066b5e..ec67c142 100644 --- a/apps/vectorcraft/src/control_server.rs +++ b/apps/vectorcraft/src/control_server.rs @@ -1,14 +1,28 @@ //! Loopback JSON-lines control server: one request per line, one reply per line. //! This is the transport the MCP server (`vectorcraft mcp --connect`) wraps. +//! +//! The server only ever reads requests: a line that is not a JSON object with a string `method` +//! (an HTTP request line from a browser's cross-origin `fetch`, a stray `nc`, binary junk) gets +//! one error reply and the connection is closed, so nothing after it is executed. Lines are +//! bounded ([`MAX_LINE`]) and so is the number of open connections ([`MAX_CONNECTIONS`]); junk +//! never reaches the UI thread. See `docs/control-protocol.md`. -use std::io::{BufRead, BufReader, Write}; +use std::io::{BufRead, BufReader, Read, Write}; use std::net::{TcpListener, TcpStream}; +use std::sync::Arc; +use std::sync::atomic::{AtomicUsize, Ordering}; use std::sync::mpsc::{Receiver, Sender, channel}; use std::time::Duration; use serde_json::{Value, json}; use vectorcraft_ui_egui::ControlRequest; +/// Longest accepted request line (bytes, without the newline). Requests are small JSON objects; +/// anything longer is refused and the connection closed. +pub const MAX_LINE: usize = 4 * 1024 * 1024; +/// Connections served at once; further ones get an error line and are closed. +pub const MAX_CONNECTIONS: usize = 16; + pub fn start(port: u16, ctx: egui::Context) -> Receiver { let (tx, rx) = channel::(); let listener = match TcpListener::bind(("127.0.0.1", port)) { @@ -19,44 +33,249 @@ pub fn start(port: u16, ctx: egui::Context) -> Receiver { } }; eprintln!("vectorcraft: control server listening on 127.0.0.1:{port}"); - std::thread::spawn(move || { - for stream in listener.incoming().flatten() { - let tx = tx.clone(); - let ctx = ctx.clone(); - std::thread::spawn(move || serve(stream, tx, ctx)); - } - }); + std::thread::spawn(move || accept_loop(listener, tx, ctx)); rx } -fn serve(stream: TcpStream, tx: Sender, ctx: egui::Context) { - let Ok(read) = stream.try_clone() else { return }; - let mut out = stream; - for line in BufReader::new(read).lines() { - let Ok(line) = line else { break }; - if line.trim().is_empty() { +fn accept_loop(listener: TcpListener, tx: Sender, ctx: egui::Context) { + let open = Arc::new(AtomicUsize::new(0)); + for mut stream in listener.incoming().flatten() { + if open.fetch_add(1, Ordering::SeqCst) >= MAX_CONNECTIONS { + open.fetch_sub(1, Ordering::SeqCst); + // Best effort: the connection is dropped either way, and a peer that can't read the reason loses nothing. + let _ = writeln!(stream, "{}", json!({"ok": false, "error": "too many control connections"})); continue; } - let reply = match serde_json::from_str::(&line) { - Ok(msg) => { - let id = msg.get("id").cloned().unwrap_or(Value::Null); - let method = msg.get("method").and_then(Value::as_str).unwrap_or("").to_string(); - let params = msg.get("params").cloned().unwrap_or(json!({})); - let (req, rrx) = ControlRequest::new(method, params); - if tx.send(req).is_err() { - break; - } - ctx.request_repaint(); - let mut r = rrx.recv_timeout(Duration::from_secs(60)).unwrap_or_else(|_| json!({"ok": false, "error": "timeout"})); - if let Some(o) = r.as_object_mut() { - o.insert("id".into(), id); + let tx = tx.clone(); + let ctx = ctx.clone(); + let open = Arc::clone(&open); + std::thread::spawn(move || { + serve(stream, &tx, &ctx); + open.fetch_sub(1, Ordering::SeqCst); + }); + } +} + +fn serve(stream: TcpStream, tx: &Sender, ctx: &egui::Context) { + let Ok(read) = stream.try_clone() else { return }; + serve_lines(read, stream, |method, params| { + let (req, rrx) = ControlRequest::new(method, params); + tx.send(req).ok()?; + ctx.request_repaint(); + Some(rrx.recv_timeout(Duration::from_secs(60)).unwrap_or_else(|_| json!({"ok": false, "error": "timeout"}))) + }); +} + +/// One parsed line. +enum Line { + Request { + id: Value, + method: String, + params: Value, + }, + 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; + } + let msg = match serde_json::from_str::(text) { + Ok(v) => v, + Err(e) => return Line::Reject(format!("bad 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(Value::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(Value::Null), + method: method.to_string(), + params: obj.get("params").cloned().unwrap_or(json!({})), + } +} + +/// 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, 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(); + if buf.last() == Some(&b'\r') { + buf.pop(); + } + return Ok(true); + } + if buf.len() > max { + return Err(format!("request line longer than {max} bytes; closing the connection")); + } + // Last line without a newline, then end of stream. + Ok(true) +} + +/// Serve requests from `read`, answering on `out`, until the peer closes, sends something that is +/// not a request, or `handle` returns `None` (app gone). `handle` gets `(method, params)` and +/// returns the reply object (the request's `id` is added here). +fn serve_lines(read: R, mut out: W, mut handle: impl FnMut(String, Value) -> Option) { + 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) => return, + Err(e) => e, + Ok(true) => match parse_line(&buf) { + Line::Blank => continue, + Line::Reject(e) => e, + Line::Request { id, method, params } => { + let Some(mut r) = handle(method, params) else { return }; + if let Some(o) = r.as_object_mut() { + o.insert("id".into(), id); + } + if writeln!(out, "{r}").and_then(|()| out.flush()).is_err() { + return; + } + continue; } - r - } - Err(e) => json!({"ok": false, "error": format!("bad JSON: {e}")}), + }, }; - if writeln!(out, "{reply}").is_err() { - break; + // Best effort: the connection closes either way, and a peer that can't read the reason loses nothing. + let _ = writeln!(out, "{}", json!({"ok": false, "error": reject})).and_then(|()| out.flush()); + return; + } +} + +#[cfg(test)] +mod tests { + use super::*; + + fn run(input: &[u8]) -> (Vec, Vec) { + let mut out = Vec::new(); + let mut calls = Vec::new(); + serve_lines(input, &mut out, |m, _p| { + calls.push(m); + Some(json!({"ok": true, "result": null})) + }); + let replies = String::from_utf8(out).unwrap().lines().map(|l| serde_json::from_str::(l).unwrap()).collect(); + (replies, calls) + } + + #[test] + fn requests_are_answered_in_order_with_ids() { + let (r, calls) = run(b"{\"id\":1,\"method\":\"a\"}\n\n{\"id\":2,\"method\":\"b\",\"params\":{}}\r\n{\"method\":\"c\"}"); + assert_eq!(calls, ["a", "b", "c"]); + assert_eq!(r.len(), 3); + assert_eq!(r[0]["id"], 1); + assert_eq!(r[1]["id"], 2); + assert_eq!(r[2]["id"], Value::Null); + } + + #[test] + fn http_request_closes_before_the_body_runs() { + // A browser's cross-origin `fetch` POST: the request line is rejected and the + // connection closed, so the JSON body is never executed. + let http = b"POST / HTTP/1.1\r\nHost: 127.0.0.1:7979\r\nOrigin: https://attacker.example\r\nContent-Type: text/plain\r\n\r\n\ + {\"id\":7,\"method\":\"engine.execute\",\"params\":{\"command\":\"file.new\",\"params\":{}}}\n"; + let (r, calls) = run(http); + assert!(calls.is_empty(), "nothing may run: {calls:?}"); + assert_eq!(r.len(), 1); + assert_eq!(r[0]["ok"], false); + assert!(r[0]["error"].as_str().unwrap().contains("closing the connection")); + } + + #[test] + fn non_request_json_closes_the_connection() { + for junk in [ + &b"[1,2]\n{\"method\":\"a\"}\n"[..], + b"{\"id\":1}\n{\"method\":\"a\"}\n", + b"{\"method\":5}\n{\"method\":\"a\"}\n", + b"\xff\xfe\n{\"method\":\"a\"}\n", + ] { + let (r, calls) = run(junk); + assert!(calls.is_empty(), "{calls:?}"); + assert_eq!(r.len(), 1); + assert_eq!(r[0]["ok"], false); } + } + + #[test] + fn a_request_before_junk_still_runs() { + let (r, calls) = run(b"{\"method\":\"a\"}\nGET / HTTP/1.1\n{\"method\":\"b\"}\n"); + assert_eq!(calls, ["a"]); + assert_eq!(r.len(), 2); + assert_eq!(r[1]["ok"], false); + } + + #[test] + fn overlong_line_is_refused() { + // One byte over the limit, then a valid request: neither runs. + let mut input = vec![b'x'; MAX_LINE + 1]; + input.extend_from_slice(b"\n{\"method\":\"a\"}\n"); + let (r, calls) = run(&input); + assert!(calls.is_empty(), "{calls:?}"); + assert_eq!(r.len(), 1); + assert!(r[0]["error"].as_str().unwrap().contains("longer than")); + } + + #[test] + fn line_at_the_limit_is_accepted() { + let mut buf = Vec::new(); + let line = vec![b'a'; 10]; + let mut input = line.clone(); + input.push(b'\n'); + assert_eq!(read_bounded_line(&mut &input[..], &mut buf, 10), Ok(true)); + assert_eq!(buf, line); + assert!(read_bounded_line(&mut &[b'a'; 11][..], &mut buf, 10).is_err()); + } + + #[test] + fn tcp_connection_is_closed_after_junk() { + // End to end over a real loopback socket (the transport the app uses). + let listener = TcpListener::bind(("127.0.0.1", 0)).unwrap(); + let addr = listener.local_addr().unwrap(); + let server = std::thread::spawn(move || { + let (s, _) = listener.accept().unwrap(); + let read = s.try_clone().unwrap(); + serve_lines(read, s, |_, _| Some(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\":7,\"method\":\"x\"}\nhello\n{\"method\":\"y\"}\n").unwrap(); + let mut text = String::new(); + c.read_to_string(&mut text).unwrap(); // returns at EOF: the server closed the connection + server.join().unwrap(); + let lines: Vec<&str> = text.lines().collect(); + assert_eq!(lines.len(), 2, "{text}"); + assert!(lines[0].contains("\"id\":7")); + assert!(lines[1].contains("closing the connection")); + } + + #[test] + fn extra_connections_are_refused() { + let listener = TcpListener::bind(("127.0.0.1", 0)).unwrap(); + let addr = listener.local_addr().unwrap(); + let (tx, _rx) = channel::(); + std::thread::spawn(move || accept_loop(listener, tx, egui::Context::default())); + // Idle connections are accepted in order and held open. + let held: Vec = (0..16).map(|_| TcpStream::connect(addr).unwrap()).collect(); + let mut extra = TcpStream::connect(addr).unwrap(); + extra.set_read_timeout(Some(Duration::from_secs(5))).unwrap(); + let mut line = String::new(); + BufReader::new(&mut extra).read_line(&mut line).unwrap(); + assert!(line.contains("too many control connections"), "{line}"); + drop(held); } } diff --git a/docs/control-protocol.md b/docs/control-protocol.md index 0b394bf5..53e78d8c 100644 --- a/docs/control-protocol.md +++ b/docs/control-protocol.md @@ -3,6 +3,16 @@ `vectorcraft --control ` listens on `127.0.0.1:` (loopback only). One JSON request per line: `{"id": 1, "method": "ui.inspect", "params": {}}` → `{"id": 1, "ok": true, "result": {...}}` or `{"id":1,"ok":false,"error":"..."}`. +**Only requests are read.** Every line must be a JSON object with a string `method` (`id` and `params` +are optional; blank lines are skipped). Anything else gets one error reply +(`{"ok": false, "error": "… closing the connection"}`) and the server **closes the connection**, so +nothing sent after it on that connection runs. That covers text that isn't JSON, a JSON array or number, +an object without `method`, invalid UTF-8, and a line longer than 4 MiB. An HTTP request (for example a +web page's cross-origin `fetch` to `127.0.0.1:`) therefore can't smuggle a command in its body: +its request line is rejected first. At most 16 connections are served at once; further ones get an error +line and are closed. Clients that get an error reply should reconnect. The port has no authentication, +so only enable it while you use it. Transport: `apps/vectorcraft/src/control_server.rs`. + | Method | Params | | |---|---|---| | `engine.execute` | `{command, params}` | run any engine or UI command (see `engine.commands`) |