Merge branch 'fix/issue-94-control-close' into integrate/safety

# Conflicts:
#	docs/whats-new.md

apps/lightcraft/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 (`lightcraft-cli 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 lightcraft_ui_egui::ControlRequest;
 use serde_json::{Value, json};
 
+/// 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<ControlRequest> {
     let (tx, rx) = channel::<ControlRequest>();
     let listener = match TcpListener::bind(("127.0.0.1", port)) {
@@ -19,44 +33,235 @@ pub fn start(port: u16, ctx: egui::Context) -> Receiver<ControlRequest> {
         }
     };
     eprintln!("lightcraft: 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<ControlRequest>, 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<ControlRequest>, 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);
+            let _ = writeln!(stream, "{}", json!({"ok": false, "error": "too many control connections"}));
             continue;
         }
-        let reply = match serde_json::from_str::<Value>(&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<ControlRequest>, 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::<Value>(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<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();
+        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<R: Read, W: Write>(read: R, mut out: W, mut handle: impl FnMut(String, Value) -> Option<Value>) {
+    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;
+        let _ = writeln!(out, "{}", json!({"ok": false, "error": reject})).and_then(|()| out.flush());
+        return;
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    fn run(input: &[u8]) -> (Vec<Value>, Vec<String>) {
+        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::<Value>(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:7980\r\nContent-Type: text/plain\r\n\r\n\
+            {\"id\":1,\"method\":\"engine.execute\",\"params\":{\"command\":\"folder.move\"}}\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());
+            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_without_buffering_it_all() {
+        // An endless line without a newline: reading stops after MAX_LINE + 1 bytes.
+        let endless = std::io::repeat(b'x');
+        let mut out = Vec::new();
+        let mut ran = false;
+        serve_lines(endless, &mut out, |_, _| {
+            ran = true;
+            None
+        });
+        assert!(!ran);
+        let reply: Value = serde_json::from_slice(out.strip_suffix(b"\n").unwrap()).unwrap();
+        assert!(reply["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(10))).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"));
     }
 }

docs/control-protocol.md

@@ -9,7 +9,16 @@
 ← {"id": 2, "ok": false, "error": "unknown command `nope`"}
 ```
 
-Requests are answered on the UI thread between frames (timeout 60 s). The MCP server's connect
+Requests are answered on the UI thread between frames (timeout 60 s).
+
+**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 — text that isn't JSON, a JSON array or number, an object without
+`method`, invalid UTF-8, or a line longer than 4 MiB — gets one error reply
+(`{"ok": false, "error": "… closing the connection"}`) and the server **closes the connection**; nothing sent after
+it on that connection runs. This keeps an HTTP request (for example a web page's cross-origin `fetch` to
+`127.0.0.1:<port>`) from smuggling a command in its body: its request line is rejected first. Junk never reaches the
+UI thread, and at most 16 connections are served at once (further ones get an error line and are closed). Clients
+that hit an error reply should reconnect. The port has no authentication, so only enable it when you need it. The MCP server's connect
 mode ([mcp.md](mcp.md)) is a thin layer over this channel. Implementation:
 `crates/ui-egui/src/control.rs` (methods) and `apps/lightcraft/src/control_server.rs` (transport).
 

docs/whats-new.md

@@ -39,6 +39,9 @@
   written by a newer LightCraft is refused with "this library was written by a newer version of LightCraft" and left
   untouched — older versions no longer read part of it as a damaged log. Once this version has opened a library,
   LightCraft 0.2.0 and older refuse it ("unsupported format … v2").
+- The control port (`--control`) closes a connection as soon as it receives anything that isn't a JSON request
+  (issue #94): an HTTP request from a web page can no longer carry a command in its body. Lines are capped at
+  4 MiB and connections at 16.
 - Canon CR2 photos from the EOS 7D, 50D, 60D, 550D, 600D, 1200D, 1300D, 5D Mark II and 1D Mark IV (and other
   models whose sensor starts on a green-blue row) no longer come out magenta (issue #85): the colour-filter
   layout is read from each file instead of assumed.