From ea417bcd11235a1ae26b948fc41b3d453e84ea1a Mon Sep 17 00:00:00 2001 From: Andrey Mnatsakanov Date: Sun, 26 Jul 2026 11:05:57 +0200 Subject: [PATCH] fix(mcp): detach SSE keep-alive so it cannot wedge the HTTP server MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The MCP HTTP server serves connections sequentially on the single current-thread runtime (the browser session is !Send). The SSE branch (GET /mcp, Accept: text/event-stream) ran an infinite keep-alive ping loop inline, so handle_connection never returned and the accept loop never ran again — one open SSE stream hung every later POST /mcp tool call indefinitely (functional break + trivial unauthenticated DoS). The keep-alive carries no browser state and its write half is Send + 'static, so spawn the ping loop on its own task and return from the handler. POST handling stays serialized (correct for a single browser session), but a live SSE stream no longer blocks the accept loop. Fixes #489 --- crates/obscura-mcp/src/http.rs | 25 +++-- .../tests/sse_stream_does_not_wedge.rs | 97 +++++++++++++++++++ 2 files changed, 114 insertions(+), 8 deletions(-) create mode 100644 crates/obscura-mcp/tests/sse_stream_does_not_wedge.rs diff --git a/crates/obscura-mcp/src/http.rs b/crates/obscura-mcp/src/http.rs index cd81099..fad3f91 100644 --- a/crates/obscura-mcp/src/http.rs +++ b/crates/obscura-mcp/src/http.rs @@ -169,7 +169,14 @@ async fn handle_connection( } "GET" if accept_sse => { - // SSE stream: hold open and send periodic keep-alive comments + // SSE stream: hold open and send periodic keep-alive comments. + // Connections are served sequentially (the browser session is + // `!Send`), so holding this infinite keep-alive loop inline would + // never return to the accept loop and would wedge every later + // request. The keep-alive touches no browser state and the write + // half is `Send + 'static`, so detach the ping loop onto its own + // task and return — the accept loop stays free while the stream + // lives on independently. let hdr = format!( "HTTP/1.1 200 OK\r\n\ Content-Type: text/event-stream\r\n\ @@ -179,14 +186,16 @@ async fn handle_connection( \r\n" ); writer.write_all(hdr.as_bytes()).await?; - loop { - tokio::time::sleep(tokio::time::Duration::from_secs(15)).await; - if writer.write_all(b": ping\n\n").await.is_err() { - break; + tokio::spawn(async move { + loop { + tokio::time::sleep(tokio::time::Duration::from_secs(15)).await; + if writer.write_all(b": ping\n\n").await.is_err() { + break; + } + let _ = writer.flush().await; } - let _ = writer.flush().await; - } - break; + }); + return Ok(()); } "POST" => { diff --git a/crates/obscura-mcp/tests/sse_stream_does_not_wedge.rs b/crates/obscura-mcp/tests/sse_stream_does_not_wedge.rs new file mode 100644 index 0000000..81c92b2 --- /dev/null +++ b/crates/obscura-mcp/tests/sse_stream_does_not_wedge.rs @@ -0,0 +1,97 @@ +//! Regression test: an open SSE stream (`GET /mcp` with +//! `Accept: text/event-stream`) must not wedge the whole MCP HTTP server. +//! +//! The server serves connections sequentially on the single current-thread +//! runtime (the browser session is `!Send`). The SSE keep-alive is an infinite +//! ping loop, so if it is held inline it never returns and the `accept()` loop +//! never runs again — every later request hangs forever. The keep-alive carries +//! no browser state, so it must be detached and the connection handler must +//! return, leaving the accept loop free. + +use std::net::TcpListener as StdListener; +use std::time::Duration; +use tokio::io::{AsyncReadExt, AsyncWriteExt}; +use tokio::net::TcpStream; +use tokio::task::LocalSet; +use tokio::time::{sleep, timeout}; + +fn pick_free_port() -> u16 { + let l = StdListener::bind("127.0.0.1:0").unwrap(); + let p = l.local_addr().unwrap().port(); + drop(l); + p +} + +#[tokio::test(flavor = "current_thread")] +async fn open_sse_stream_does_not_block_other_requests() { + let port = pick_free_port(); + let local = LocalSet::new(); + + let server = local.spawn_local(async move { + let _ = obscura_mcp::http::run("127.0.0.1".to_string(), port, None, None, false).await; + }); + + local + .run_until(async { + // Wait for the listener to bind. + for _ in 0..40 { + if TcpStream::connect(("127.0.0.1", port)).await.is_ok() { + break; + } + sleep(Duration::from_millis(50)).await; + } + + // Connection A: open the SSE stream and read its headers so the + // server-side handler has entered its keep-alive path. Keep it open. + let mut sse = TcpStream::connect(("127.0.0.1", port)) + .await + .expect("MCP server did not come up"); + let get = b"GET /mcp HTTP/1.1\r\n\ + Host: 127.0.0.1\r\n\ + Accept: text/event-stream\r\n\ + \r\n"; + sse.write_all(get).await.unwrap(); + sse.flush().await.unwrap(); + + let mut hbuf = [0u8; 512]; + let hn = timeout(Duration::from_secs(2), sse.read(&mut hbuf)) + .await + .expect("SSE headers read timed out") + .expect("SSE headers read failed"); + let sse_head = String::from_utf8_lossy(&hbuf[..hn]).to_lowercase(); + assert!( + sse_head.contains("text/event-stream"), + "expected an SSE response on the GET stream, got:\n{sse_head}" + ); + + // Connection B: while A's SSE stream is still open, a second request + // must still be accepted and answered promptly. Pre-fix this hangs + // because the SSE loop never yields the accept loop. + let mut stream = TcpStream::connect(("127.0.0.1", port)) + .await + .expect("second connection refused"); + let req = b"OPTIONS /mcp HTTP/1.1\r\n\ + Host: 127.0.0.1\r\n\ + Origin: https://dashboard.example.com\r\n\ + Access-Control-Request-Method: POST\r\n\ + \r\n"; + stream.write_all(req).await.unwrap(); + stream.flush().await.unwrap(); + + let mut buf = [0u8; 1024]; + let n = timeout(Duration::from_secs(3), stream.read(&mut buf)) + .await + .expect("second request timed out — the SSE stream wedged the server") + .expect("read failed"); + let response = String::from_utf8_lossy(&buf[..n]).to_string(); + + server.abort(); + drop(sse); + + assert!( + response.starts_with("HTTP/1.1 204"), + "expected the second request to be served (204), got:\n{response}" + ); + }) + .await; +}