net/
http.rs

1// SPDX-License-Identifier: Apache-2.0
2//! A minimal blocking HTTP/1.1 client over any `Read + Write`. RFC 0006.
3//!
4//! This is the single highest-leverage minimalism decision (RFC 0002): one
5//! ~250-line module replaces the `ureq`/`url`→IDNA→ICU dependency tax. It
6//! carries the intelligence wire over TCP, TLS, unix sockets, and vsock
7//! alike — the transport is just the stream.
8//!
9//! Two request paths: [`send`] buffers the whole response (the LLM/intelligence
10//! path), and [`send_streaming`] returns the status + headers plus a live reader
11//! so the caller can either buffer it (`application/json`) or pump it as an **SSE**
12//! stream ([`SseReader`]) — the MCP Streamable HTTP transport, where a response may
13//! be a single JSON body or a `text/event-stream`, and a long-lived GET carries
14//! server→client notifications.
15//! `connect_tcp` is intentionally unguarded, and dials by *name*: it exists for
16//! the *operator-configured* endpoints (the intelligence dial, the auth/token
17//! endpoints), which the model cannot influence. A model/agent/peer-supplied URL
18//! must NOT be dialled through it — not even with a guard wrapped around it,
19//! since the guard's lookup and this function's lookup are two resolutions and a
20//! hostile nameserver can answer them differently. Those surfaces use
21//! [`crate::ssrf::connect_vetted`], which resolves once and dials the address it
22//! vetted.
23
24use std::io::{self, BufRead, BufReader, Read, Write};
25use std::net::TcpStream;
26use std::time::Duration;
27
28/// Response body cap. LLM responses can be large; 8 MiB is generous without
29/// being an unbounded allocation from a hostile peer.
30pub const MAX_RESPONSE: usize = 8 * 1024 * 1024;
31
32/// Any bidirectional byte stream the HTTP client can run over. `Box<dyn Stream>`
33/// is itself `Read + Write` (via std's `impl<R: Read + ?Sized> Read for Box<R>`),
34/// so an OWNED boxed stream can be handed to [`send_streaming`] by value — used
35/// by the long-lived MCP notification SSE reader.
36pub trait Stream: Read + Write {}
37impl<T: Read + Write> Stream for T {}
38
39/// A parsed absolute URL (the subset we need: scheme/host/port/path).
40#[derive(Debug, Clone, PartialEq, Eq)]
41pub struct Url {
42    pub scheme: String,
43    pub host: String,
44    pub port: u16,
45    /// Path + query, always starting with `/`.
46    pub path: String,
47}
48
49impl Url {
50    /// Parse `http(s)://host[:port][/path][?query]`. No `url` crate — we only
51    /// support the absolute http/https forms agentd actually issues.
52    pub fn parse(s: &str) -> Result<Url, String> {
53        let (scheme, rest) = s
54            .split_once("://")
55            .ok_or_else(|| format!("not an absolute URL: {s}"))?;
56        let scheme = scheme.to_ascii_lowercase();
57        let default_port = match scheme.as_str() {
58            "http" => 80,
59            "https" => 443,
60            other => return Err(format!("unsupported scheme: {other}")),
61        };
62        let (authority, path) = match rest.find('/') {
63            Some(i) => (&rest[..i], &rest[i..]),
64            None => (rest, "/"),
65        };
66        if authority.is_empty() {
67            return Err(format!("missing host in URL: {s}"));
68        }
69        let (host, port) = match authority.rsplit_once(':') {
70            // ':' only counts as a port separator if what follows is numeric
71            // (guards against IPv6 literals, which we don't expect here).
72            Some((h, p)) if p.chars().all(|c| c.is_ascii_digit()) && !p.is_empty() => (
73                h.to_string(),
74                p.parse().map_err(|_| format!("bad port in {s}"))?,
75            ),
76            _ => (authority.to_string(), default_port),
77        };
78        Ok(Url {
79            scheme,
80            host,
81            port,
82            path: path.to_string(),
83        })
84    }
85
86    pub fn is_tls(&self) -> bool {
87        self.scheme == "https"
88    }
89
90    /// The `Host:` header value (includes a non-default port).
91    pub fn host_header(&self) -> String {
92        let default = if self.is_tls() { 443 } else { 80 };
93        if self.port == default {
94            self.host.clone()
95        } else {
96            format!("{}:{}", self.host, self.port)
97        }
98    }
99}
100
101/// A parsed HTTP response.
102#[derive(Debug, Clone)]
103pub struct Response {
104    pub status: u16,
105    pub headers: Vec<(String, String)>,
106    pub body: Vec<u8>,
107}
108
109impl Response {
110    /// Case-insensitive header lookup (header names are stored lowercased).
111    pub fn header(&self, name: &str) -> Option<&str> {
112        let name = name.to_ascii_lowercase();
113        self.headers
114            .iter()
115            .find(|(k, _)| *k == name)
116            .map(|(_, v)| v.as_str())
117    }
118
119    pub fn is_success(&self) -> bool {
120        (200..300).contains(&self.status)
121    }
122
123    pub fn body_str(&self) -> std::borrow::Cow<'_, str> {
124        String::from_utf8_lossy(&self.body)
125    }
126}
127
128/// Whether `host` names the local loopback — the dev/test carve-out for
129/// plaintext `http://` (production transports are TLS-only). Accepts the IPv4
130/// loopback block (`127.0.0.0/8`), the IPv6 loopback (`::1`, bare or
131/// bracketed), and the literal name `localhost`. A resolvable-but-unresolved
132/// name is NOT loopback — this classifies the written form, without DNS.
133pub fn is_loopback_host(host: &str) -> bool {
134    let h = host.trim_start_matches('[').trim_end_matches(']');
135    if h.eq_ignore_ascii_case("localhost") {
136        return true;
137    }
138    h.parse::<std::net::IpAddr>()
139        .map(|ip| ip.is_loopback())
140        .unwrap_or(false)
141}
142
143/// Connect a plain TCP stream with connect + read/write timeouts. Intentionally
144/// unguarded, and resolves the name itself — for operator-configured endpoints
145/// only. A model/agent/peer-supplied URL belongs on
146/// [`crate::ssrf::connect_vetted`]: composing `ssrf::guard_host` around *this*
147/// function resolves twice and is the DNS-rebinding hole, not a fix for it.
148pub fn connect_tcp(host: &str, port: u16, timeout: Duration) -> io::Result<TcpStream> {
149    use std::net::ToSocketAddrs;
150    let addr = (host, port).to_socket_addrs()?.next().ok_or_else(|| {
151        io::Error::new(
152            io::ErrorKind::NotFound,
153            format!("cannot resolve {host}:{port}"),
154        )
155    })?;
156    let stream = TcpStream::connect_timeout(&addr, timeout)?;
157    stream.set_read_timeout(Some(timeout))?;
158    stream.set_write_timeout(Some(timeout))?;
159    stream.set_nodelay(true).ok();
160    Ok(stream)
161}
162
163/// Issue one request over `stream` and read the full response. Adds `Host`,
164/// `Connection: close`, and `Content-Length`; the caller supplies any other
165/// headers (e.g. `Authorization`, `Content-Type`).
166pub fn send<S: Read + Write + ?Sized>(
167    stream: &mut S,
168    host_header: &str,
169    method: &str,
170    path: &str,
171    headers: &[(&str, &str)],
172    body: &[u8],
173) -> io::Result<Response> {
174    // The request TARGET is caller-supplied too — a templated endpoint path or
175    // a peer-supplied A2A push-notification URL — and CR/LF there splits the
176    // request line exactly as it splits a header, letting an injected
177    // `Authorization:` shadow the operator's real one further down. Scanned
178    // before anything is written so the framing layer stays closed on both
179    // caller surfaces, not just headers (RFC 0012).
180    let mut req: Vec<u8> = Vec::with_capacity(256 + body.len());
181    if path.contains(['\r', '\n']) {
182        return Err(io::Error::new(
183            io::ErrorKind::InvalidInput,
184            "CR/LF in request target",
185        ));
186    }
187    write!(req, "{method} {path} HTTP/1.1\r\n")?;
188    write!(req, "Host: {host_header}\r\n")?;
189    req.extend_from_slice(b"Connection: close\r\n");
190    for (k, v) in headers {
191        // Reject CR/LF injection in caller-supplied headers (RFC 0012).
192        if k.contains(['\r', '\n']) || v.contains(['\r', '\n']) {
193            return Err(io::Error::new(
194                io::ErrorKind::InvalidInput,
195                "CR/LF in header",
196            ));
197        }
198        write!(req, "{k}: {v}\r\n")?;
199    }
200    write!(req, "Content-Length: {}\r\n\r\n", body.len())?;
201    req.extend_from_slice(body);
202    stream.write_all(&req)?;
203    stream.flush()?;
204
205    let mut reader = BufReader::new(stream);
206    read_response(&mut reader)
207}
208
209/// Read the status line + headers off a response, leaving the reader positioned
210/// at the start of the body. Header names are lowercased.
211fn read_head<R: BufRead>(r: &mut R) -> io::Result<(u16, Vec<(String, String)>)> {
212    let mut status_line = String::new();
213    r.read_line(&mut status_line)?;
214    let status = parse_status(&status_line)?;
215
216    let mut headers = Vec::new();
217    loop {
218        let mut line = String::new();
219        if r.read_line(&mut line)? == 0 {
220            break;
221        }
222        let line = line.trim_end_matches(['\r', '\n']);
223        if line.is_empty() {
224            break;
225        }
226        if let Some((k, v)) = line.split_once(':') {
227            headers.push((k.trim().to_ascii_lowercase(), v.trim().to_string()));
228        }
229    }
230    Ok((status, headers))
231}
232
233fn read_response<R: BufRead>(r: &mut R) -> io::Result<Response> {
234    let (status, headers) = read_head(r)?;
235
236    let content_length = headers
237        .iter()
238        .find(|(k, _)| k == "content-length")
239        .and_then(|(_, v)| v.parse::<usize>().ok());
240    let chunked = headers
241        .iter()
242        .any(|(k, v)| k == "transfer-encoding" && v.to_ascii_lowercase().contains("chunked"));
243
244    let body = if chunked {
245        read_chunked(r)?
246    } else if let Some(n) = content_length {
247        read_exact_capped(r, n)?
248    } else {
249        // Connection: close — read to EOF, capped.
250        read_to_end_capped(r)?
251    };
252
253    Ok(Response {
254        status,
255        headers,
256        body,
257    })
258}
259
260fn parse_status(line: &str) -> io::Result<u16> {
261    // "HTTP/1.1 200 OK"
262    line.split_whitespace()
263        .nth(1)
264        .and_then(|s| s.parse().ok())
265        .ok_or_else(|| {
266            io::Error::new(
267                io::ErrorKind::InvalidData,
268                format!("bad status line: {line:?}"),
269            )
270        })
271}
272
273fn read_exact_capped<R: Read>(r: &mut R, n: usize) -> io::Result<Vec<u8>> {
274    if n > MAX_RESPONSE {
275        return Err(io::Error::new(
276            io::ErrorKind::InvalidData,
277            "response exceeds cap",
278        ));
279    }
280    let mut buf = vec![0u8; n];
281    r.read_exact(&mut buf)?;
282    Ok(buf)
283}
284
285fn read_to_end_capped<R: Read>(r: &mut R) -> io::Result<Vec<u8>> {
286    let mut buf = Vec::new();
287    r.take(MAX_RESPONSE as u64 + 1).read_to_end(&mut buf)?;
288    if buf.len() > MAX_RESPONSE {
289        return Err(io::Error::new(
290            io::ErrorKind::InvalidData,
291            "response exceeds cap",
292        ));
293    }
294    Ok(buf)
295}
296
297fn read_chunked<R: BufRead>(r: &mut R) -> io::Result<Vec<u8>> {
298    let mut body = Vec::new();
299    loop {
300        let mut size_line = String::new();
301        r.read_line(&mut size_line)?;
302        let size_hex = size_line.trim_end_matches(['\r', '\n']);
303        // A chunk extension (`;name=val`) may follow the size.
304        let size_hex = size_hex.split(';').next().unwrap_or("").trim();
305        let size = usize::from_str_radix(size_hex, 16)
306            .map_err(|_| io::Error::new(io::ErrorKind::InvalidData, "bad chunk size"))?;
307        if size == 0 {
308            // Consume the trailing CRLF (and any trailers) until blank line.
309            loop {
310                let mut t = String::new();
311                if r.read_line(&mut t)? == 0 || t.trim_end_matches(['\r', '\n']).is_empty() {
312                    break;
313                }
314            }
315            break;
316        }
317        // The declared size is peer-controlled and unverified until the bytes
318        // actually arrive, so it must never be trusted with arithmetic OR with
319        // an allocation. `size > MAX_RESPONSE` is checked first and the running
320        // total uses `saturating_add`: the plain `body.len() + size` this
321        // replaces wrapped on a size near `usize::MAX` — panicking the process
322        // in debug, and in release wrapping *below* the cap so the check passed
323        // and the allocation below aborted on a multi-exabyte request. That is
324        // remotely reachable: `runtime::mod` dials every configured MCP server
325        // at startup, so a hostile server could kill the daemon at connect time,
326        // defeating the "a down server is only logged" containment.
327        if size > MAX_RESPONSE || body.len().saturating_add(size) > MAX_RESPONSE {
328            return Err(io::Error::new(
329                io::ErrorKind::InvalidData,
330                "chunked response exceeds cap",
331            ));
332        }
333        // Read the chunk incrementally onto `body` instead of pre-allocating
334        // `size` bytes, so even under the cap a lying header buys the peer an
335        // allocation only as large as the bytes it really sends. A short read
336        // means the peer framed a chunk it never delivered — that is a
337        // truncated response, not an empty one, so it must not parse.
338        let before = body.len();
339        r.by_ref().take(size as u64).read_to_end(&mut body)?;
340        if body.len() - before != size {
341            return Err(io::Error::new(
342                io::ErrorKind::UnexpectedEof,
343                "truncated chunk",
344            ));
345        }
346        // Trailing CRLF after the chunk.
347        let mut crlf = [0u8; 2];
348        r.read_exact(&mut crlf)?;
349    }
350    Ok(body)
351}
352
353/// A streamed response: status + headers, plus the reader positioned at the body.
354/// The caller decides how to drain it — [`into_body`] to buffer (`application/json`)
355/// or [`sse`] to pump it as an SSE event stream (`text/event-stream`). Owns the
356/// underlying stream, matching the MCP client's per-request connection model.
357pub struct StreamingResponse<S: Read + Write> {
358    pub status: u16,
359    pub headers: Vec<(String, String)>,
360    reader: BufReader<S>,
361}
362
363impl<S: Read + Write> StreamingResponse<S> {
364    pub fn header(&self, name: &str) -> Option<&str> {
365        let name = name.to_ascii_lowercase();
366        self.headers
367            .iter()
368            .find(|(k, _)| *k == name)
369            .map(|(_, v)| v.as_str())
370    }
371    pub fn is_success(&self) -> bool {
372        (200..300).contains(&self.status)
373    }
374    /// The lowercased `Content-Type` (media type only, params stripped).
375    pub fn content_type(&self) -> Option<&str> {
376        self.header("content-type")
377            .map(|v| v.split(';').next().unwrap_or(v).trim())
378    }
379    /// `true` when the body is `text/event-stream` (Streamable HTTP SSE).
380    pub fn is_event_stream(&self) -> bool {
381        self.content_type() == Some("text/event-stream")
382    }
383    /// Buffer the whole body (capped), honoring `Content-Length`/`chunked`/close.
384    pub fn into_body(mut self) -> io::Result<Vec<u8>> {
385        let content_length = self
386            .headers
387            .iter()
388            .find(|(k, _)| k == "content-length")
389            .and_then(|(_, v)| v.parse::<usize>().ok());
390        let chunked = self
391            .headers
392            .iter()
393            .any(|(k, v)| k == "transfer-encoding" && v.to_ascii_lowercase().contains("chunked"));
394        if chunked {
395            read_chunked(&mut self.reader)
396        } else if let Some(n) = content_length {
397            read_exact_capped(&mut self.reader, n)
398        } else {
399            read_to_end_capped(&mut self.reader)
400        }
401    }
402    /// Consume into an [`SseReader`] to pump `text/event-stream` events.
403    pub fn sse(self) -> SseReader<BufReader<S>> {
404        SseReader::new(self.reader)
405    }
406
407    /// Consume into the raw body reader — for a response that turned out NOT to
408    /// be an event stream (e.g. a peer that answered `application/json`).
409    pub fn into_reader(self) -> BufReader<S> {
410        self.reader
411    }
412}
413
414/// Issue one request over an OWNED `stream` and return the status + headers +
415/// body reader WITHOUT draining the body (unlike [`send`]). Adds `Host`,
416/// `Connection: close`, and `Content-Length`; the caller supplies the rest
417/// (`Accept`, `Authorization`, `Content-Type`, `Mcp-Session-Id`, …).
418pub fn send_streaming<S: Read + Write>(
419    mut stream: S,
420    host_header: &str,
421    method: &str,
422    path: &str,
423    headers: &[(&str, &str)],
424    body: &[u8],
425) -> io::Result<StreamingResponse<S>> {
426    // Same request-target scan as [`send`] — this is the MCP path, where the
427    // target comes from server-advertised endpoint metadata.
428    let mut req: Vec<u8> = Vec::with_capacity(256 + body.len());
429    if path.contains(['\r', '\n']) {
430        return Err(io::Error::new(
431            io::ErrorKind::InvalidInput,
432            "CR/LF in request target",
433        ));
434    }
435    write!(req, "{method} {path} HTTP/1.1\r\n")?;
436    write!(req, "Host: {host_header}\r\n")?;
437    req.extend_from_slice(b"Connection: close\r\n");
438    for (k, v) in headers {
439        if k.contains(['\r', '\n']) || v.contains(['\r', '\n']) {
440            return Err(io::Error::new(
441                io::ErrorKind::InvalidInput,
442                "CR/LF in header",
443            ));
444        }
445        write!(req, "{k}: {v}\r\n")?;
446    }
447    write!(req, "Content-Length: {}\r\n\r\n", body.len())?;
448    req.extend_from_slice(body);
449    stream.write_all(&req)?;
450    stream.flush()?;
451
452    let mut reader = BufReader::new(stream);
453    let (status, headers) = read_head(&mut reader)?;
454    Ok(StreamingResponse {
455        status,
456        headers,
457        reader,
458    })
459}
460
461/// One parsed SSE event (RFC 6455-style `text/event-stream`). For MCP, `data` is
462/// a JSON-RPC message; `event`/`id` are the optional SSE field lines.
463#[derive(Debug, Clone, Default, PartialEq, Eq)]
464pub struct SseEvent {
465    pub event: Option<String>,
466    pub data: String,
467    pub id: Option<String>,
468}
469
470/// A blocking, line-based SSE reader. `next_event` accumulates `field: value`
471/// lines and emits one [`SseEvent`] per blank-line separator (multiple `data:`
472/// lines join with `\n`), returning `Ok(None)` at end of stream. Bounded per
473/// event by [`MAX_RESPONSE`] so a hostile stream cannot exhaust memory.
474pub struct SseReader<R: BufRead> {
475    r: R,
476}
477
478impl<R: BufRead> SseReader<R> {
479    pub fn new(r: R) -> SseReader<R> {
480        SseReader { r }
481    }
482
483    /// Read the next event, or `Ok(None)` at EOF. Comment lines (`:` prefix) and
484    /// unknown fields are ignored per the SSE spec.
485    pub fn next_event(&mut self) -> io::Result<Option<SseEvent>> {
486        let mut ev = SseEvent::default();
487        let mut saw_field = false;
488        let mut total = 0usize;
489        loop {
490            let mut line = String::new();
491            let n = self.r.read_line(&mut line)?;
492            if n == 0 {
493                // EOF: flush a pending event if one was in progress.
494                return Ok(if saw_field { Some(ev) } else { None });
495            }
496            total += n;
497            if total > MAX_RESPONSE {
498                return Err(io::Error::new(
499                    io::ErrorKind::InvalidData,
500                    "SSE event exceeds cap",
501                ));
502            }
503            let line = line.trim_end_matches(['\r', '\n']);
504            if line.is_empty() {
505                // Blank line dispatches the accumulated event.
506                if saw_field {
507                    return Ok(Some(ev));
508                }
509                continue; // stray blank line between events
510            }
511            if line.starts_with(':') {
512                continue; // comment
513            }
514            let (field, value) = match line.split_once(':') {
515                Some((f, v)) => (f, v.strip_prefix(' ').unwrap_or(v)),
516                None => (line, ""), // a bare field name with empty value
517            };
518            saw_field = true;
519            match field {
520                "event" => ev.event = Some(value.to_string()),
521                "id" => ev.id = Some(value.to_string()),
522                "data" => {
523                    if !ev.data.is_empty() {
524                        ev.data.push('\n');
525                    }
526                    ev.data.push_str(value);
527                }
528                _ => {} // retry/unknown — ignore
529            }
530        }
531    }
532}
533
534#[cfg(test)]
535mod tests {
536    use super::*;
537    use std::io::Cursor;
538
539    #[test]
540    fn url_parse_https_default_port() {
541        let u = Url::parse("https://api.openai.com/v1/chat/completions").unwrap();
542        assert_eq!(u.scheme, "https");
543        assert_eq!(u.host, "api.openai.com");
544        assert_eq!(u.port, 443);
545        assert_eq!(u.path, "/v1/chat/completions");
546        assert_eq!(u.host_header(), "api.openai.com");
547        assert!(u.is_tls());
548    }
549
550    #[test]
551    fn url_parse_http_with_port_and_no_path() {
552        let u = Url::parse("http://localhost:8080").unwrap();
553        assert_eq!(u.port, 8080);
554        assert_eq!(u.path, "/");
555        assert_eq!(u.host_header(), "localhost:8080");
556        assert!(!u.is_tls());
557    }
558
559    #[test]
560    fn url_rejects_bad_scheme() {
561        assert!(Url::parse("ftp://x/").is_err());
562        assert!(Url::parse("no-scheme").is_err());
563    }
564
565    #[test]
566    fn response_content_length() {
567        let raw = "HTTP/1.1 200 OK\r\nContent-Type: application/json\r\nContent-Length: 13\r\n\r\n{\"ok\":true}!!";
568        let mut cur = Cursor::new(raw.as_bytes().to_vec());
569        let resp = read_response(&mut cur).unwrap();
570        assert_eq!(resp.status, 200);
571        assert_eq!(resp.header("content-type"), Some("application/json"));
572        assert_eq!(resp.body, b"{\"ok\":true}!!");
573        assert!(resp.is_success());
574    }
575
576    #[test]
577    fn response_chunked() {
578        let raw = "HTTP/1.1 200 OK\r\nTransfer-Encoding: chunked\r\n\r\n5\r\nhello\r\n6\r\n world\r\n0\r\n\r\n";
579        let mut cur = Cursor::new(raw.as_bytes().to_vec());
580        let resp = read_response(&mut cur).unwrap();
581        assert_eq!(resp.body, b"hello world");
582    }
583
584    #[test]
585    fn cr_lf_header_injection_rejected() {
586        let mut sink: Vec<u8> = Vec::new();
587        // a write-only fake stream: Cursor over Vec implements Write+Read
588        let mut stream = Cursor::new(Vec::new());
589        let _ = &mut sink;
590        let err = send(&mut stream, "h", "POST", "/", &[("X", "a\r\nEvil: 1")], b"").unwrap_err();
591        assert_eq!(err.kind(), io::ErrorKind::InvalidInput);
592    }
593
594    /// A fake duplex stream: reads return a canned server response, writes are
595    /// captured (so a request/response round-trip is testable without sockets).
596    struct FakeStream {
597        resp: Cursor<Vec<u8>>,
598        sink: Vec<u8>,
599    }
600    impl Read for FakeStream {
601        fn read(&mut self, buf: &mut [u8]) -> io::Result<usize> {
602            self.resp.read(buf)
603        }
604    }
605    impl Write for FakeStream {
606        fn write(&mut self, buf: &[u8]) -> io::Result<usize> {
607            self.sink.extend_from_slice(buf);
608            Ok(buf.len())
609        }
610        fn flush(&mut self) -> io::Result<()> {
611            Ok(())
612        }
613    }
614
615    fn sse_events(body: &str) -> Vec<SseEvent> {
616        let mut r = SseReader::new(BufReader::new(Cursor::new(body.as_bytes().to_vec())));
617        let mut out = Vec::new();
618        while let Some(e) = r.next_event().unwrap() {
619            out.push(e);
620        }
621        out
622    }
623
624    #[test]
625    fn sse_parses_events_with_event_id_and_data() {
626        let body = "event: message\nid: 7\ndata: {\"jsonrpc\":\"2.0\",\"id\":1,\"result\":{}}\n\n";
627        let evs = sse_events(body);
628        assert_eq!(evs.len(), 1);
629        assert_eq!(evs[0].event.as_deref(), Some("message"));
630        assert_eq!(evs[0].id.as_deref(), Some("7"));
631        assert_eq!(evs[0].data, "{\"jsonrpc\":\"2.0\",\"id\":1,\"result\":{}}");
632    }
633
634    #[test]
635    fn sse_joins_multi_data_lines_and_ignores_comments() {
636        // Comment line, then an event whose data spans two `data:` lines.
637        let body = ": keep-alive\ndata: line1\ndata: line2\n\ndata: second\n\n";
638        let evs = sse_events(body);
639        assert_eq!(evs.len(), 2);
640        assert_eq!(evs[0].data, "line1\nline2");
641        assert_eq!(evs[1].data, "second");
642    }
643
644    #[test]
645    fn sse_flushes_trailing_event_without_final_blank_line() {
646        let evs = sse_events("data: only\n");
647        assert_eq!(evs.len(), 1);
648        assert_eq!(evs[0].data, "only");
649    }
650
651    #[test]
652    fn send_streaming_reads_head_then_buffers_json_body() {
653        let raw = "HTTP/1.1 200 OK\r\nContent-Type: application/json\r\nMcp-Session-Id: abc123\r\nContent-Length: 11\r\n\r\n{\"ok\":true}".to_string();
654        let stream = FakeStream {
655            resp: Cursor::new(raw.into_bytes()),
656            sink: Vec::new(),
657        };
658        let resp = send_streaming(
659            stream,
660            "h",
661            "POST",
662            "/mcp",
663            &[("Accept", "application/json")],
664            b"{}",
665        )
666        .unwrap();
667        assert_eq!(resp.status, 200);
668        assert!(resp.is_success());
669        assert_eq!(resp.content_type(), Some("application/json"));
670        assert!(!resp.is_event_stream());
671        assert_eq!(resp.header("mcp-session-id"), Some("abc123"));
672        assert_eq!(resp.into_body().unwrap(), b"{\"ok\":true}");
673    }
674
675    #[test]
676    fn send_streaming_detects_event_stream_and_pumps_sse() {
677        let raw = "HTTP/1.1 200 OK\r\nContent-Type: text/event-stream\r\n\r\ndata: {\"jsonrpc\":\"2.0\",\"id\":1,\"result\":{\"x\":1}}\n\n".to_string();
678        let stream = FakeStream {
679            resp: Cursor::new(raw.into_bytes()),
680            sink: Vec::new(),
681        };
682        let resp = send_streaming(stream, "h", "POST", "/mcp", &[], b"{}").unwrap();
683        assert!(resp.is_event_stream());
684        let mut sse = resp.sse();
685        let ev = sse.next_event().unwrap().expect("one event");
686        assert!(ev.data.contains("\"id\":1"));
687        assert!(sse.next_event().unwrap().is_none());
688    }
689}
net/http.rs

net/
http.rs
