forked from 1b-agent/opencadstudio-upstream
fix(io): cap line reads on the serve and MCP transports
Bug: The --serve automation loop (stdin/socket) and the MCP stdio loop read input with unbounded read_line/lines. A single line over 64 MiB - a huge paste or a hostile client - grows the buffer until the allocator aborts, the same class of failure as the HTTP path, on the non-HTTP transports. Reproduce: 1. Start --serve on a socket (or the MCP server on stdio). 2. Send one 65 MiB line with no terminating newline anywhere in it. 3. Before: RSS grows to the full line, then "memory allocation failed" (or a multi-second stall while the line is copied). 4. After: the read fails at the cap and the existing EOF arm closes the connection/stdio promptly. Fix: New shared src/io/line_read.rs module: lines_capped(reader, 64 MiB) yields Err for an over-cap line, which the serve and MCP loops already treat as EOF/close. rest.rs was refactored onto the same helpers so the cap logic lives in one place. A regression test streams a 65 MiB line at the serve socket and asserts the connection ends without a response.
This commit is contained in:
1 parent
6c3c266475
commit
be8a7b95c4
5 files changed
+197
-49
No files matched your search
+82
-6
@@ -15,7 +15,9 @@
|
|||||||
//! once opened/saved)
|
//! once opened/saved)
|
||||||
|
|
||||||
#[cfg(not(target_arch = "wasm32"))]
|
#[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::collections::{BTreeMap, HashSet};
|
||||||
use std::path::PathBuf;
|
use std::path::PathBuf;
|
||||||
|
|
||||||
@@ -32,7 +34,11 @@ use super::OpenCADStudio;
|
|||||||
pub fn serve() {
|
pub fn serve() {
|
||||||
let mut app = OpenCADStudio::new();
|
let mut app = OpenCADStudio::new();
|
||||||
match port_arg() {
|
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),
|
None => serve_stdio(&mut app),
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
@@ -106,7 +112,7 @@ fn serve_stdio(app: &mut OpenCADStudio) {
|
|||||||
let _ = writeln!(o, "{}", ready());
|
let _ = writeln!(o, "{}", ready());
|
||||||
let _ = o.flush();
|
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 Ok(line) = line else { break };
|
||||||
let line = line.trim();
|
let line = line.trim();
|
||||||
if line.is_empty() {
|
if line.is_empty() {
|
||||||
@@ -120,7 +126,11 @@ fn serve_stdio(app: &mut OpenCADStudio) {
|
|||||||
}
|
}
|
||||||
|
|
||||||
#[cfg(not(target_arch = "wasm32"))]
|
#[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<std::sync::atomic::AtomicU16>,
|
||||||
|
) {
|
||||||
let listener = match std::net::TcpListener::bind(("127.0.0.1", port)) {
|
let listener = match std::net::TcpListener::bind(("127.0.0.1", port)) {
|
||||||
Ok(l) => l,
|
Ok(l) => l,
|
||||||
Err(e) => {
|
Err(e) => {
|
||||||
@@ -128,7 +138,9 @@ fn serve_socket(app: &mut OpenCADStudio, port: u16) {
|
|||||||
return;
|
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() {
|
for stream in listener.incoming().flatten() {
|
||||||
let Ok(read_half) = stream.try_clone() else {
|
let Ok(read_half) = stream.try_clone() else {
|
||||||
continue;
|
continue;
|
||||||
@@ -137,7 +149,7 @@ fn serve_socket(app: &mut OpenCADStudio, port: u16) {
|
|||||||
let mut writer = stream;
|
let mut writer = stream;
|
||||||
let _ = writeln!(writer, "{}", ready());
|
let _ = writeln!(writer, "{}", ready());
|
||||||
let _ = writer.flush();
|
let _ = writer.flush();
|
||||||
for line in reader.lines() {
|
for line in lines_capped(reader, MAX_LINE_BYTES) {
|
||||||
let Ok(line) = line else { break };
|
let Ok(line) = line else { break };
|
||||||
let line = line.trim();
|
let line = line.trim();
|
||||||
if line.is_empty() {
|
if line.is_empty() {
|
||||||
@@ -3283,4 +3295,68 @@ mod tests {
|
|||||||
let _ = app.update(Message::ToggleRibbonDropdown("PROP_COLOR".to_string()));
|
let _ = app.update(Message::ToggleRibbonDropdown("PROP_COLOR".to_string()));
|
||||||
let _ = app.view(wid);
|
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();
|
||||||
|
}
|
||||||
}
|
}
|
||||||
@@ -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::<LineTooLong>())
|
||||||
|
}
|
||||||
|
|
||||||
|
/// `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<usize> {
|
||||||
|
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<R> {
|
||||||
|
reader: R,
|
||||||
|
cap: usize,
|
||||||
|
done: bool,
|
||||||
|
}
|
||||||
|
|
||||||
|
pub fn lines_capped<R: BufRead>(reader: R, cap: usize) -> CappedLines<R> {
|
||||||
|
CappedLines {
|
||||||
|
reader,
|
||||||
|
cap,
|
||||||
|
done: false,
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
impl<R: BufRead> Iterator for CappedLines<R> {
|
||||||
|
type Item = std::io::Result<String>;
|
||||||
|
|
||||||
|
fn next(&mut self) -> Option<Self::Item> {
|
||||||
|
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))
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
@@ -20,6 +20,7 @@ pub mod stl;
|
|||||||
pub mod xref;
|
pub mod xref;
|
||||||
pub mod xref_model;
|
pub mod xref_model;
|
||||||
pub mod linetypes;
|
pub mod linetypes;
|
||||||
|
pub mod line_read;
|
||||||
pub mod patterns;
|
pub mod patterns;
|
||||||
pub mod update_check;
|
pub mod update_check;
|
||||||
pub mod paper_catalog;
|
pub mod paper_catalog;
|
||||||
|
|||||||
+5
-2
@@ -9,7 +9,7 @@ use serde_json::{Map, Value, json};
|
|||||||
use std::{
|
use std::{
|
||||||
collections::{HashMap, VecDeque},
|
collections::{HashMap, VecDeque},
|
||||||
fs::{File, OpenOptions},
|
fs::{File, OpenOptions},
|
||||||
io::{self, BufRead, BufReader, Read, Write},
|
io::{self, BufReader, Read, Write},
|
||||||
net::TcpStream,
|
net::TcpStream,
|
||||||
path::{Path, PathBuf},
|
path::{Path, PathBuf},
|
||||||
process::{Child, Command, Stdio},
|
process::{Child, Command, Stdio},
|
||||||
@@ -1171,7 +1171,10 @@ pub fn run() {
|
|||||||
let mut output = stdout.lock();
|
let mut output = stdout.lock();
|
||||||
let mut clients = HashMap::new();
|
let mut clients = HashMap::new();
|
||||||
let mut tasks = TaskStore::default();
|
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 {
|
let response = match line {
|
||||||
Ok(line) if !line.trim().is_empty() => match serde_json::from_str::<Value>(&line) {
|
Ok(line) if !line.trim().is_empty() => match serde_json::from_str::<Value>(&line) {
|
||||||
Ok(message) => handle_message(message, &mut clients, &mut tasks),
|
Ok(message) => handle_message(message, &mut clients, &mut tasks),
|
||||||
|
|||||||
+3
-41
@@ -23,8 +23,9 @@
|
|||||||
//! so both `--http` modes serve the identical surface.
|
//! so both `--http` modes serve the identical surface.
|
||||||
|
|
||||||
use crate::app::OpenCADStudio;
|
use crate::app::OpenCADStudio;
|
||||||
|
use crate::io::line_read::{is_line_too_long, read_line_capped};
|
||||||
use serde_json::{json, Value};
|
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::net::{TcpListener, TcpStream};
|
||||||
use std::sync::atomic::{AtomicU16, AtomicU64, AtomicUsize, Ordering};
|
use std::sync::atomic::{AtomicU16, AtomicU64, AtomicUsize, Ordering};
|
||||||
use std::sync::Arc;
|
use std::sync::Arc;
|
||||||
@@ -206,46 +207,6 @@ fn serve_connection(mut stream: TcpStream, jobs: std::sync::mpsc::Sender<Job>) {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
/// 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::<LineTooLong>())
|
|
||||||
}
|
|
||||||
|
|
||||||
/// `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<usize> {
|
|
||||||
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
|
/// 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
|
/// 9110's `431` — then hand the caller `Ok(None)` so it closes without
|
||||||
/// dispatching. Writing the refusal from here keeps the headless `--http`
|
/// 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 {
|
mod tests {
|
||||||
use super::*;
|
use super::*;
|
||||||
use crate::app::OpenCADStudio;
|
use crate::app::OpenCADStudio;
|
||||||
|
use std::io::BufRead;
|
||||||
|
|
||||||
fn request(method: &str, path: &str, body: &str) -> HttpRequest {
|
fn request(method: &str, path: &str, body: &str) -> HttpRequest {
|
||||||
HttpRequest {
|
HttpRequest {
|
||||||
|
|||||||
Reference in new issue
Block a user