hive-c0re: bound request lines on the agent and manager sockets

serve() read each request with read_line into an unbounded String, so
size checks such as the 4 KiB Send body limit ran only after the whole
line was buffered. Any process in an agent container can write to
/run/hive/mcp.sock, and one stream with no newline grew hive-c0re's heap
until the host OOM-killed it.

Read at most 16 MiB per line. The largest request a client sends is an
OperatorMsg from the agent web UI's /send, whose body axum's default
limit caps at 2 MiB (the gateway's nginx admits 10 MiB); JSON escaping
can roughly double that, and the cap is 4x the result. A longer line
gets a Response::Err, the same shape as the parse-error path, and the
connection is closed because the rest of the line is still unread.

The line is read as bytes and parsed with serde_json::from_slice, so a
line with invalid UTF-8 now gets a parse-error response instead of the
connection being dropped.

Closes #4718

hive-c0re/src/socket_server/mod.rs

@@ -15,7 +15,7 @@ use anyhow::{Context, Result};
 use hive_core_agent_sock::{Request, Response};
 use hive_sh4re::inbox::Message;
 use hive_sh4re::manager::MANAGER_AGENT;
-use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader};
+use tokio::io::{AsyncBufRead, AsyncBufReadExt, AsyncReadExt, AsyncWriteExt, BufReader};
 use tokio::net::{UnixListener, UnixStream};
 use tokio::task::JoinHandle;
 
@@ -128,26 +128,77 @@ pub fn start_manager(coord: Arc<Coordinator>) -> Result<()> {
     Ok(())
 }
 
+/// Longest request line accepted on the agent and manager sockets, newline
+/// included. The largest request a client sends is an `OperatorMsg` from the
+/// agent web UI's `/send` (`hive-agent/src/web_ui/actions.rs:21`), whose body
+/// axum's default limit caps at 2 MiB; JSON escaping can roughly double that to
+/// 4 MiB, and this is 4× that.
+const REQUEST_LINE_MAX_BYTES: usize = 16 * 1024 * 1024;
+
+enum RequestLine {
+    Eof,
+    Complete,
+    TooLong,
+}
+
+/// Read one request line into `buf`, consuming at most
+/// [`REQUEST_LINE_MAX_BYTES`] from `reader`.
+async fn read_request_line<R: AsyncBufRead + Unpin>(
+    reader: &mut R,
+    buf: &mut Vec<u8>,
+) -> std::io::Result<RequestLine> {
+    buf.clear();
+    let n = (&mut *reader)
+        .take(REQUEST_LINE_MAX_BYTES as u64)
+        .read_until(b'\n', buf)
+        .await?;
+    Ok(if n == 0 {
+        RequestLine::Eof
+    } else if n == REQUEST_LINE_MAX_BYTES && !buf.ends_with(b"\n") {
+        RequestLine::TooLong
+    } else {
+        RequestLine::Complete
+    })
+}
+
+/// Answer each request line on `stream` until EOF. An over-long line gets a
+/// `Response::Err` and then the connection is closed, because the rest of the
+/// line is still unread.
 async fn serve(stream: UnixStream, agent: String, coord: Arc<Coordinator>) -> Result<()> {
     let (read, mut write) = stream.into_split();
     let mut reader = BufReader::new(read);
-    let mut line = String::new();
+    let mut line = Vec::new();
     loop {
-        line.clear();
+        let (resp, too_long) = match read_request_line(&mut reader, &mut line).await? {
-        let n = reader.read_line(&mut line).await?;
+            RequestLine::Eof => return Ok(()),
-        if n == 0 {
+            RequestLine::TooLong => (
-            return Ok(());
+                Response::Err {
-        }
+                    message: format!(
-        let resp = match serde_json::from_str::<Request>(line.trim()) {
+                        "request too long: no newline within {REQUEST_LINE_MAX_BYTES} bytes; \
-            Ok(req) => dispatch(&req, &agent, &coord).await,
+                         connection closed"
-            Err(e) => Response::Err {
+                    ),
+                },
+                true,
+            ),
+            RequestLine::Complete => match serde_json::from_slice::<Request>(&line) {
+                Ok(req) => (dispatch(&req, &agent, &coord).await, false),
+                Err(e) => (
+                    Response::Err {
                         message: format!("parse error: {e}"),
                     },
+                    false,
+                ),
+            },
         };
         let mut payload = serde_json::to_string(&resp)?;
         payload.push('\n');
         write.write_all(payload.as_bytes()).await?;
         write.flush().await?;
+        if too_long {
+            anyhow::bail!(
+                "`{agent}` sent a request line over {REQUEST_LINE_MAX_BYTES} bytes; closed"
+            );
+        }
     }
 }
 
@@ -791,4 +842,55 @@ mod tests {
         let err = check_can_cancel_approval("nobody-with-no-groups").unwrap_err();
         assert!(err.contains("approvals` tool group"), "{err}");
     }
+
+    #[tokio::test]
+    async fn over_long_line_is_rejected_after_reading_only_the_bound() {
+        let total = 2 * REQUEST_LINE_MAX_BYTES as u64;
+        let mut reader = BufReader::new(tokio::io::repeat(b'x').take(total));
+        let mut buf = Vec::new();
+        let outcome = read_request_line(&mut reader, &mut buf)
+            .await
+            .expect("read");
+        // At most one `BufReader` fill past the bound left the source.
+        let unread = reader.into_inner().limit();
+        let consumed = total - unread;
+        assert!(
+            consumed <= (REQUEST_LINE_MAX_BYTES + 8 * 1024) as u64,
+            "consumed {consumed} of {total} bytes"
+        );
+        assert_eq!(buf.len(), REQUEST_LINE_MAX_BYTES);
+        assert!(matches!(outcome, RequestLine::TooLong));
+    }
+
+    #[tokio::test]
+    async fn serve_answers_an_over_long_line_with_an_error_and_closes() {
+        let (_dir, coord) = schedules::tests::coordinator();
+        let (client, server) = UnixStream::pair().expect("socketpair");
+        let handler = tokio::spawn(serve(server, "iris".to_owned(), coord));
+        let (read, mut write) = client.into_split();
+        let writer = tokio::spawn(async move {
+            let chunk = vec![b'x'; 64 * 1024];
+            for _ in 0..2 * REQUEST_LINE_MAX_BYTES / chunk.len() {
+                if write.write_all(&chunk).await.is_err() {
+                    return;
+                }
+            }
+            let _ = write.shutdown().await;
+        });
+        let mut reply = String::new();
+        BufReader::new(read)
+            .read_line(&mut reply)
+            .await
+            .expect("reply");
+        let resp: Response = serde_json::from_str(&reply).expect("response json");
+        assert!(
+            matches!(&resp, Response::Err { message } if message.starts_with("request too long")),
+            "{reply}"
+        );
+        assert!(
+            handler.await.expect("join").is_err(),
+            "connection not closed"
+        );
+        writer.await.expect("join");
+    }
 }

hive-c0re/src/socket_server/schedules.rs

@@ -302,7 +302,7 @@ fn schedule_to_wire(s: crate::scheduled_prompts::Schedule) -> hive_sh4re::schedu
 }
 
 #[cfg(test)]
-mod tests {
+pub(super) mod tests {
     use super::*;
 
     fn target(name: &str) -> hive_sh4re::schedule::WireScheduleTarget {
@@ -377,10 +377,10 @@ mod tests {
     /// anybody's ancestor or descendant.
     const STRANGER: &str = "requester-with-no-relations";
 
-    /// A real `Coordinator` over a throwaway sqlite dir. The schedule
+    /// A real `Coordinator` over a throwaway sqlite dir. The socket-server
     /// handlers take `&Arc<Coordinator>`, so there is no lighter way in;
     /// `open` touches nothing outside the db path it is handed.
-    fn coordinator() -> (tempfile::TempDir, Arc<Coordinator>) {
+    pub(in crate::socket_server) fn coordinator() -> (tempfile::TempDir, Arc<Coordinator>) {
         let dir = tempfile::tempdir().expect("tempdir");
         let coord = Coordinator::open(
             &dir.path().join("broker.sqlite"),