diff --git a/src/app/automation.rs b/src/app/automation.rs index ed5bdaa58..ca052aab5 100644 --- a/src/app/automation.rs +++ b/src/app/automation.rs @@ -15,7 +15,9 @@ //! once opened/saved) #[cfg(not(target_arch = "wasm32"))] -use std::io::{BufRead, Write}; +use crate::io::line_read::{lines_capped, MAX_LINE_BYTES}; +#[cfg(not(target_arch = "wasm32"))] +use std::io::Write; use std::collections::{BTreeMap, HashSet}; use std::path::PathBuf; @@ -32,7 +34,11 @@ use super::OpenCADStudio; pub fn serve() { let mut app = OpenCADStudio::new(); match port_arg() { - Some(port) => serve_socket(&mut app, port), + Some(port) => serve_socket( + &mut app, + port, + &std::sync::Arc::new(std::sync::atomic::AtomicU16::new(0)), + ), None => serve_stdio(&mut app), } } @@ -106,7 +112,7 @@ fn serve_stdio(app: &mut OpenCADStudio) { let _ = writeln!(o, "{}", ready()); let _ = o.flush(); } - for line in stdin.lock().lines() { + for line in lines_capped(stdin.lock(), MAX_LINE_BYTES) { let Ok(line) = line else { break }; let line = line.trim(); if line.is_empty() { @@ -120,7 +126,11 @@ fn serve_stdio(app: &mut OpenCADStudio) { } #[cfg(not(target_arch = "wasm32"))] -fn serve_socket(app: &mut OpenCADStudio, port: u16) { +fn serve_socket( + app: &mut OpenCADStudio, + port: u16, + bound_port: &std::sync::Arc, +) { let listener = match std::net::TcpListener::bind(("127.0.0.1", port)) { Ok(l) => l, Err(e) => { @@ -128,7 +138,9 @@ fn serve_socket(app: &mut OpenCADStudio, port: u16) { return; } }; - eprintln!("OpenCADStudio --serve listening on 127.0.0.1:{port}"); + let bound = listener.local_addr().map(|a| a.port()).unwrap_or(port); + bound_port.store(bound, std::sync::atomic::Ordering::SeqCst); + eprintln!("OpenCADStudio --serve listening on 127.0.0.1:{bound}"); for stream in listener.incoming().flatten() { let Ok(read_half) = stream.try_clone() else { continue; @@ -137,7 +149,7 @@ fn serve_socket(app: &mut OpenCADStudio, port: u16) { let mut writer = stream; let _ = writeln!(writer, "{}", ready()); let _ = writer.flush(); - for line in reader.lines() { + for line in lines_capped(reader, MAX_LINE_BYTES) { let Ok(line) = line else { break }; let line = line.trim(); if line.is_empty() { @@ -3283,4 +3295,68 @@ mod tests { let _ = app.update(Message::ToggleRibbonDropdown("PROP_COLOR".to_string())); let _ = app.view(wid); } + + /// A JSON-lines client streaming one endless line must not be buffered + /// (or processed) without bound: `BufRead::lines` grows its `String` by + /// doubling until the allocation aborts the process — an abort nobody + /// upstream can catch. The connection is dropped at the cap instead. + #[cfg(not(target_arch = "wasm32"))] + #[test] + fn serve_socket_drops_a_client_streaming_an_oversize_line() { + use std::io::{BufRead, BufReader, Write}; + use std::net::TcpStream; + use std::sync::atomic::{AtomicU16, Ordering}; + use std::sync::Arc; + + let bound = Arc::new(AtomicU16::new(0)); + let listen_bound = bound.clone(); + std::thread::spawn(move || { + let mut app = super::OpenCADStudio::new_for_test(); + super::serve_socket(&mut app, 0, &listen_bound); + }); + let mut port = 0; + for _ in 0..50 { + port = bound.load(Ordering::SeqCst); + if port != 0 { + break; + } + std::thread::sleep(std::time::Duration::from_millis(100)); + } + assert_ne!(port, 0, "listener never bound a port"); + + let client = TcpStream::connect(("127.0.0.1", port)).expect("connect"); + client + .set_read_timeout(Some(std::time::Duration::from_secs(10))) + .expect("read timeout"); + let mut reader = BufReader::new(client.try_clone().expect("clone")); + let mut banner = String::new(); + reader.read_line(&mut banner).expect("ready banner"); + assert!(banner.contains("ready"), "{banner}"); + + // Stream >64 MiB without a newline from its own thread: the payload + // dwarfs any socket buffer, so one thread would deadlock writing + // before the server reads. + let mut writer = client.try_clone().expect("clone"); + let payload = std::thread::spawn(move || { + let chunk = vec![b'A'; 1024 * 1024]; + for _ in 0..65 { + if writer.write_all(&chunk).is_err() { + return; + } + } + let _ = writer.write_all(b"\n"); + let _ = writer.flush(); + }); + + // The server must drop the connection WITHOUT answering: a response + // means the oversize line was read whole and handed to the dispatcher. + let mut response = String::new(); + let read = reader.read_line(&mut response); + assert_eq!( + read.unwrap_or(0), + 0, + "oversize line must be dropped, not processed — got: {response:?}" + ); + payload.join().unwrap(); + } } diff --git a/src/io/line_read.rs b/src/io/line_read.rs new file mode 100644 index 000000000..8465b52b9 --- /dev/null +++ b/src/io/line_read.rs @@ -0,0 +1,106 @@ +//! Bounded line reads for the JSON-lines wire protocols (`--serve`, `--mcp`). +//! +//! `BufRead::read_line` — and the `.lines()` iterator built on it — grow +//! their `String` by doubling with no ceiling, so a peer streaming one line +//! without a newline pushes the allocation until the process dies: an +//! allocation failure is an *abort*, not a catchable panic. The REST +//! transport learned this the hard way (`src/rest.rs`); this module is the +//! shared ceiling for every other line-oriented channel. +//! +//! The cap matches the plugin IPC frame budget +//! (`MAX_MESSAGE_SIZE = 64 MiB` in `crates/ocs_plugin_api`): generous enough +//! that no legitimate message (base64 images included) is refused, small +//! enough that memory stays bounded. + +use std::io::{BufRead, Read}; + +/// Upper bound for one protocol line, newline included. +pub const MAX_LINE_BYTES: usize = 64 * 1024 * 1024; + +/// Marker for a line that exceeded its size cap. Kept as a distinct error +/// payload so callers can tell "peer blew the cap" apart from transport +/// errors (reset, timeout) if they ever need to answer differently. +#[derive(Debug)] +pub struct LineTooLong; + +impl std::fmt::Display for LineTooLong { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + f.write_str("line exceeds the size limit") + } +} + +impl std::error::Error for LineTooLong {} + +/// True for the cap violation [`read_line_capped`] raises. +pub fn is_line_too_long(error: &std::io::Error) -> bool { + error + .get_ref() + .map_or(false, |inner| inner.is::()) +} + +/// `BufRead::read_line` capped at `cap` bytes — the whole line including its +/// newline. The underlying read stops at `cap + 1` bytes, so memory stays +/// bounded no matter how much the peer streams; an over-cap line fails with +/// [`LineTooLong`] instead of growing `buf` forever. +pub fn read_line_capped( + reader: &mut impl BufRead, + buf: &mut String, + cap: usize, +) -> std::io::Result { + buf.clear(); + let read = reader.by_ref().take(cap as u64 + 1).read_line(buf)?; + if read > cap { + return Err(std::io::Error::new( + std::io::ErrorKind::InvalidData, + LineTooLong, + )); + } + Ok(read) +} + +/// Iterator over [`read_line_capped`] results, trimming line endings exactly +/// like [`BufRead::lines`]. Yields `Err` — and then stops — for the first +/// over-cap line, so callers keep their existing read-error handling. +pub struct CappedLines { + reader: R, + cap: usize, + done: bool, +} + +pub fn lines_capped(reader: R, cap: usize) -> CappedLines { + CappedLines { + reader, + cap, + done: false, + } +} + +impl Iterator for CappedLines { + type Item = std::io::Result; + + fn next(&mut self) -> Option { + if self.done { + return None; + } + let mut buf = String::new(); + match read_line_capped(&mut self.reader, &mut buf, self.cap) { + Ok(0) => { + self.done = true; + None + } + Ok(_) => { + if buf.ends_with('\n') { + buf.pop(); + if buf.ends_with('\r') { + buf.pop(); + } + } + Some(Ok(buf)) + } + Err(error) => { + self.done = true; + Some(Err(error)) + } + } + } +} diff --git a/src/io/mod.rs b/src/io/mod.rs index a6c4e96b3..45fbc2f33 100644 --- a/src/io/mod.rs +++ b/src/io/mod.rs @@ -20,6 +20,7 @@ pub mod stl; pub mod xref; pub mod xref_model; pub mod linetypes; +pub mod line_read; pub mod patterns; pub mod update_check; pub mod paper_catalog; diff --git a/src/mcp.rs b/src/mcp.rs index 7ce1348a6..13ab89649 100644 --- a/src/mcp.rs +++ b/src/mcp.rs @@ -9,7 +9,7 @@ use serde_json::{Map, Value, json}; use std::{ collections::{HashMap, VecDeque}, fs::{File, OpenOptions}, - io::{self, BufRead, BufReader, Read, Write}, + io::{self, BufReader, Read, Write}, net::TcpStream, path::{Path, PathBuf}, process::{Child, Command, Stdio}, @@ -1171,7 +1171,10 @@ pub fn run() { let mut output = stdout.lock(); let mut clients = HashMap::new(); let mut tasks = TaskStore::default(); - for line in stdin.lock().lines() { + for line in crate::io::line_read::lines_capped( + stdin.lock(), + crate::io::line_read::MAX_LINE_BYTES, + ) { let response = match line { Ok(line) if !line.trim().is_empty() => match serde_json::from_str::(&line) { Ok(message) => handle_message(message, &mut clients, &mut tasks), diff --git a/src/rest.rs b/src/rest.rs index 8abcf0fb6..d494ad796 100644 --- a/src/rest.rs +++ b/src/rest.rs @@ -23,8 +23,9 @@ //! so both `--http` modes serve the identical surface. use crate::app::OpenCADStudio; +use crate::io::line_read::{is_line_too_long, read_line_capped}; use serde_json::{json, Value}; -use std::io::{BufRead, BufReader, Read, Write}; +use std::io::{BufReader, Read, Write}; use std::net::{TcpListener, TcpStream}; use std::sync::atomic::{AtomicU16, AtomicU64, AtomicUsize, Ordering}; use std::sync::Arc; @@ -206,46 +207,6 @@ fn serve_connection(mut stream: TcpStream, jobs: std::sync::mpsc::Sender) { } } -/// Marker for a line that exceeded its size cap. Kept as a distinct error -/// payload so the parse path can answer `431` for it while transport errors -/// (reset, timeout) and invalid UTF-8 keep their existing silent-close path. -#[derive(Debug)] -struct LineTooLong; - -impl std::fmt::Display for LineTooLong { - fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { - f.write_str("line exceeds the size limit") - } -} - -impl std::error::Error for LineTooLong {} - -fn is_line_too_long(error: &std::io::Error) -> bool { - error - .get_ref() - .map_or(false, |inner| inner.is::()) -} - -/// `BufRead::read_line` capped at `cap` bytes — the whole line including its -/// newline. The underlying read stops at `cap + 1` bytes, so memory stays -/// bounded no matter how much the peer streams; an over-cap line fails with -/// [`LineTooLong`] instead of growing `buf` forever. -fn read_line_capped( - reader: &mut impl BufRead, - buf: &mut String, - cap: usize, -) -> std::io::Result { - buf.clear(); - let read = reader.by_ref().take(cap as u64 + 1).read_line(buf)?; - if read > cap { - return Err(std::io::Error::new( - std::io::ErrorKind::InvalidData, - LineTooLong, - )); - } - Ok(read) -} - /// Answer a peer whose request line or header block blew a size cap — RFC /// 9110's `431` — then hand the caller `Ok(None)` so it closes without /// dispatching. Writing the refusal from here keeps the headless `--http` @@ -764,6 +725,7 @@ pub(crate) fn map_status(response: Value, created: bool) -> u16 { mod tests { use super::*; use crate::app::OpenCADStudio; + use std::io::BufRead; fn request(method: &str, path: &str, body: &str) -> HttpRequest { HttpRequest {