diff --git a/src/conn_actor.rs b/src/conn_actor.rs index 70b6a35..36ea5f1 100644 --- a/src/conn_actor.rs +++ b/src/conn_actor.rs @@ -133,6 +133,8 @@ pub fn run_connection( let _ = registry.cast(Cast::ConnBusy(me)); // ----- 2. Read body. ----- + // Content-Length pre-check only applies to fixed bodies; a chunked + // body is bounded incrementally by the decoder. let body_len = parsed.content_length.unwrap_or(0); if body_len > limits.max_body_bytes { let _ = write_all( @@ -152,11 +154,41 @@ pub fn run_connection( } } - let body = match read_body(raw, &mut buf, parsed.head_len, body_len, request_deadline) { - Ok(b) => b, - // Timeout mid-body (and any other body io error) -> just close; - // there's no point talking HTTP to a client this far gone. - Err(_) => return, + // `consumed_past_head`: how many RAW bytes of `buf` past the head + // this request's body occupied — for chunked bodies that is framing + // included, NOT the decoded length. The keep-alive drain at the + // bottom of the loop must drop exactly this much to land on the + // next pipelined request. + let (body, consumed_past_head) = if parsed.chunked { + match read_chunked_body(raw, &mut buf, parsed.head_len, &limits, request_deadline) { + Ok(ok) => ok, + Err(ChunkedBodyErr::TooLarge) => { + let _ = write_all( + raw, + b"HTTP/1.1 413 Payload Too Large\r\ncontent-length: 0\r\nconnection: close\r\n\r\n", + Instant::now() + limits.write_timeout, + ); + return; + } + Err(ChunkedBodyErr::Malformed) => { + let _ = write_all( + raw, + b"HTTP/1.1 400 Bad Request\r\ncontent-length: 0\r\nconnection: close\r\n\r\n", + Instant::now() + limits.write_timeout, + ); + return; + } + // Timeout mid-body (and any other io error) -> just close. + Err(ChunkedBodyErr::Io(_)) => return, + } + } else { + match read_body(raw, &mut buf, parsed.head_len, body_len, request_deadline) { + Ok(b) => (b, body_len), + // Timeout mid-body (and any other body io error) -> just + // close; there's no point talking HTTP to a client this far + // gone. + Err(_) => return, + } }; let keep_alive = parsed.keep_alive; @@ -236,9 +268,9 @@ pub fn run_connection( // catch us, which is acceptable: drain means no new work.) let _ = registry.cast(Cast::ConnIdle(me)); - // Drop the request bytes (head + body) from `buf`; anything past - // them is the start of the next pipelined request. - let consumed = head_len + body_len; + // Drop the request bytes (head + raw body framing) from `buf`; + // anything past them is the start of the next pipelined request. + let consumed = head_len + consumed_past_head; buf.drain(..consumed); } } @@ -364,6 +396,135 @@ fn read_body( Ok(buf[head_len..head_len + body_len].to_vec()) } +// --------------------------------------------------------------------------- +// read_chunked_body — incremental chunked transfer-decoding (request side). +// --------------------------------------------------------------------------- +// +// Decodes `Transfer-Encoding: chunked` from `buf[head_len..]`, reading more +// from the socket as needed on the SAME request deadline the head was read +// under. Returns (decoded_body, raw_bytes_consumed_past_head) — the raw +// count includes all framing and the trailer section, so the caller's +// keep-alive drain lands exactly on the next pipelined request. +// +// Bounds: the DECODED size is capped at max_body_bytes (-> TooLarge/413); +// a single size line (incl. chunk extensions, which are ignored) is capped +// at MAX_SIZE_LINE and the trailer section at MAX_TRAILER_BYTES (-> +// Malformed/400) so framing spam can't grow `buf` unboundedly. Trailers +// are consumed and discarded — nothing in the pipeline wants them yet. + +const MAX_SIZE_LINE: usize = 128; +const MAX_TRAILER_BYTES: usize = 8 * 1024; + +#[allow(dead_code)] // io::Error is captured for future logging +enum ChunkedBodyErr { + Io(io::Error), + Malformed, + TooLarge, +} + +fn read_chunked_body( + fd: RawFd, + buf: &mut Vec, + head_len: usize, + limits: &ConnLimits, + deadline: Instant, +) -> Result<(Vec, usize), ChunkedBodyErr> { + // Ensure `buf` holds at least `until` bytes, reading on the request + // deadline. Io(TimedOut) on expiry, UnexpectedEof on early close. + fn fill_to( + fd: RawFd, + buf: &mut Vec, + until: usize, + deadline: Instant, + ) -> Result<(), ChunkedBodyErr> { + while buf.len() < until { + match read_some(fd, buf, 8 * 1024, deadline) { + Ok(0) => { + return Err(ChunkedBodyErr::Io(io::Error::new( + ErrorKind::UnexpectedEof, + "client closed during chunked body", + ))) + } + Ok(_) => {} + Err(e) => return Err(ChunkedBodyErr::Io(e)), + } + } + Ok(()) + } + + // Find "\r\n" in buf[from..], reading more as needed; the line may be + // at most `max_line` bytes (terminator excluded). Returns the index of + // the '\r'. + fn find_crlf( + fd: RawFd, + buf: &mut Vec, + from: usize, + max_line: usize, + deadline: Instant, + ) -> Result { + let mut scan = from; + loop { + while scan + 1 < buf.len() { + if buf[scan] == b'\r' && buf[scan + 1] == b'\n' { + return Ok(scan); + } + scan += 1; + if scan - from > max_line { + return Err(ChunkedBodyErr::Malformed); + } + } + fill_to(fd, buf, buf.len() + 1, deadline)?; + } + } + + let mut pos = head_len; + let mut decoded: Vec = Vec::new(); + + loop { + // ----- size line: HEX[;extensions]\r\n ----- + let line_end = find_crlf(fd, buf, pos, MAX_SIZE_LINE, deadline)?; + let line = &buf[pos..line_end]; + let size_str = match line.iter().position(|&b| b == b';') { + Some(i) => &line[..i], // chunk extensions: ignored + None => line, + }; + let size_str = std::str::from_utf8(size_str) + .map_err(|_| ChunkedBodyErr::Malformed)? + .trim(); + let size = usize::from_str_radix(size_str, 16) + .map_err(|_| ChunkedBodyErr::Malformed)?; + pos = line_end + 2; + + if size == 0 { + // ----- trailer section: zero or more header lines, then CRLF ----- + let trailer_start = pos; + loop { + let t_end = find_crlf(fd, buf, pos, MAX_SIZE_LINE.max(1024), deadline)?; + let empty = t_end == pos; + pos = t_end + 2; + if empty { + return Ok((decoded, pos - head_len)); + } + if pos - trailer_start > MAX_TRAILER_BYTES { + return Err(ChunkedBodyErr::Malformed); + } + } + } + + if decoded.len() + size > limits.max_body_bytes { + return Err(ChunkedBodyErr::TooLarge); + } + + // ----- chunk payload + trailing CRLF ----- + fill_to(fd, buf, pos + size + 2, deadline)?; + decoded.extend_from_slice(&buf[pos..pos + size]); + if &buf[pos + size..pos + size + 2] != b"\r\n" { + return Err(ChunkedBodyErr::Malformed); + } + pos += size + 2; + } +} + // --------------------------------------------------------------------------- // read_some — single epoll-park + read loop, bounded by a deadline. // --------------------------------------------------------------------------- diff --git a/src/parser.rs b/src/parser.rs index 7c93352..8ba83ea 100644 --- a/src/parser.rs +++ b/src/parser.rs @@ -4,14 +4,13 @@ //! copy, battle-tested). Body framing, keep-alive logic and response writing //! are ours. //! -//! v1 body framing: +//! Body framing: //! - `Content-Length: N` — read exactly N bytes. //! - No body header — empty body. -//! - `Transfer-Encoding: chunked` — deferred. Returns ParseError::Unsupported -//! and the connection actor responds 411 Length Required + close. -//! -//! Keep it stupid simple. Chunked decoding lands when something actually -//! requests it. +//! - `Transfer-Encoding: chunked` (HTTP/1.1) — flagged in `ParsedHead`; +//! the connection actor decodes incrementally (`read_chunked_body`). +//! Chunked + Content-Length together, or chunked on HTTP/1.0, is +//! Malformed (request-smuggling ambiguity; RFC 7230 §3.3.3). use crate::conn::{Body, Conn, HeaderMap, HttpVersion, Method, RespBody}; @@ -30,8 +29,9 @@ pub enum ParseError { TooManyHeaders, /// `Content-Length` header could not be parsed as an integer. BadContentLength, - /// A wire feature we haven't implemented yet (e.g. chunked encoding). - /// Connection actor responds 411 + close. + /// A wire feature we haven't implemented. Currently unconstructed + /// (chunked decoding landed in v0.3); kept for future unsupported + /// framings. Connection actor responds 411 + close. Unsupported, } @@ -50,6 +50,10 @@ pub struct ParsedHead { pub version: HttpVersion, pub headers: HeaderMap, pub content_length: Option, + /// `Transfer-Encoding: chunked` — the body is chunked-framed and the + /// connection actor decodes it (`read_chunked_body`). Mutually + /// exclusive with `content_length` (rejected as Malformed). + pub chunked: bool, pub keep_alive: bool, pub expect_100: bool, } @@ -130,7 +134,13 @@ pub fn parse_head(buf: &[u8], max_headers: usize) -> Result Result {} - _ => panic!("expected Unsupported for chunked"), + Err(ParseError::Malformed) => {} + _ => panic!("expected Malformed for CL + chunked"), + } + } + + #[test] + fn parse_chunked_on_http10_is_malformed() { + let req = b"POST /a HTTP/1.0\r\nHost: x\r\nTransfer-Encoding: chunked\r\n\r\n"; + match parse_head(req, 64) { + Err(ParseError::Malformed) => {} + _ => panic!("expected Malformed for chunked on 1.0"), } } diff --git a/tests/integration.rs b/tests/integration.rs index dfc06d8..55c8995 100644 --- a/tests/integration.rs +++ b/tests/integration.rs @@ -772,3 +772,136 @@ fn stalled_reader_killed_at_write_timeout() { .recv_timeout(Duration::from_secs(10)) .expect("producer never observed conn death: write_timeout not enforced"); } + +// --------------------------------------------------------------------------- +// Chunked request bodies (v0.3 chunk 2) +// --------------------------------------------------------------------------- + +fn echo_pipeline() -> Pipeline { + Pipeline::new().plug( + Router::new() + .get("/", |c: Conn, _n: Next| c.put_status(200).put_body("hello")) + .post("/echo", |c: Conn, _n: Next| { + let body = c.body.as_bytes().to_vec(); + c.put_status(200).put_body(body) + }), + ) +} + +/// A chunked request body (with chunk extensions, which must be ignored) +/// is decoded before the pipeline runs; the handler sees the joined bytes. +#[test] +fn chunked_request_body_is_decoded() { + let port = spawn_server(echo_pipeline()); + let resp = send_request( + port, + b"POST /echo HTTP/1.1\r\nHost: x\r\nTransfer-Encoding: chunked\r\nConnection: close\r\n\r\n\ + 5\r\nhello\r\n6;ext=1\r\n world\r\n0\r\n\r\n", + ); + assert_eq!(http_status(&resp), 200); + assert_eq!(http_body(&resp), b"hello world"); +} + +/// Trailer headers after the 0-chunk are consumed and discarded; the +/// connection stays usable (keep-alive across a chunked request, with the +/// next request pipelined in the same packet — exercises the raw-consumed +/// accounting). +#[test] +fn chunked_request_trailers_and_pipelining() { + let port = spawn_server(echo_pipeline()); + let mut s = TcpStream::connect(("127.0.0.1", port)).unwrap(); + s.set_read_timeout(Some(Duration::from_secs(5))).unwrap(); + s.write_all( + b"POST /echo HTTP/1.1\r\nHost: x\r\nTransfer-Encoding: chunked\r\n\r\n\ + 3\r\nabc\r\n0\r\nx-trailer: ignored\r\n\r\n\ + GET / HTTP/1.1\r\nHost: x\r\nConnection: close\r\n\r\n", + ) + .unwrap(); + let mut resp = Vec::new(); + s.read_to_end(&mut resp).unwrap(); + let text = String::from_utf8_lossy(&resp); + // Two responses back to back: the echo, then hello. + assert!(text.contains("abc"), "echo body missing: {text}"); + let second = &text[text.find("abc").unwrap()..]; + assert!(second.contains("hello"), "pipelined second response missing: {text}"); +} + +/// A chunked body whose decoded size exceeds max_body_bytes is rejected +/// with 413 the moment the limit would be crossed. +#[test] +fn chunked_request_over_limit_413() { + let port = free_port(); + let addr: SocketAddr = format!("127.0.0.1:{port}").parse().unwrap(); + let pipe = echo_pipeline(); + std::thread::spawn(move || { + let cfg = Config { + listener_pool: 2, + scheduler_threads: Some(2), + max_body_bytes: 8, // tiny + ..Config::new(addr) + }; + serve_with(cfg, pipe).unwrap(); + }); + for _ in 0..50 { + if TcpStream::connect(addr).is_ok() { break; } + std::thread::sleep(Duration::from_millis(50)); + } + let mut s = TcpStream::connect(addr).unwrap(); + s.set_read_timeout(Some(Duration::from_secs(5))).unwrap(); + s.write_all( + b"POST /echo HTTP/1.1\r\nHost: x\r\nTransfer-Encoding: chunked\r\n\r\n\ + 9\r\n123456789\r\n0\r\n\r\n", + ) + .unwrap(); + let mut resp = Vec::new(); + s.read_to_end(&mut resp).unwrap(); + assert_eq!(http_status(&resp), 413); +} + +/// Garbage where a chunk-size line should be is 400. +#[test] +fn chunked_request_malformed_size_400() { + let port = spawn_server(echo_pipeline()); + let resp = send_request( + port, + b"POST /echo HTTP/1.1\r\nHost: x\r\nTransfer-Encoding: chunked\r\n\r\n\ + zz\r\nhello\r\n0\r\n\r\n", + ); + assert_eq!(http_status(&resp), 400); +} + +/// Content-Length + Transfer-Encoding: chunked together is the smuggling +/// ambiguity: rejected 400 at parse time. +#[test] +fn chunked_plus_content_length_400() { + let port = spawn_server(echo_pipeline()); + let resp = send_request( + port, + b"POST /echo HTTP/1.1\r\nHost: x\r\nContent-Length: 5\r\nTransfer-Encoding: chunked\r\n\r\n\ + 5\r\nhello\r\n0\r\n\r\n", + ); + assert_eq!(http_status(&resp), 400); +} + +/// A chunked body that stalls mid-stream is killed by the request +/// deadline: the connection just closes (no response owed mid-body). +#[test] +fn chunked_request_stall_killed_at_request_timeout() { + let port = spawn_server_with_timeouts( + echo_pipeline(), + Duration::from_secs(30), + Duration::from_millis(400), // request_timeout + ); + let mut s = TcpStream::connect(("127.0.0.1", port)).unwrap(); + s.set_read_timeout(Some(Duration::from_secs(5))).unwrap(); + // Head + first chunk, then stall forever before the 0-chunk. + s.write_all( + b"POST /echo HTTP/1.1\r\nHost: x\r\nTransfer-Encoding: chunked\r\n\r\n5\r\nhello\r\n", + ) + .unwrap(); + let start = std::time::Instant::now(); + let mut resp = Vec::new(); + s.read_to_end(&mut resp).unwrap(); // server closes; EOF + assert!(resp.is_empty(), "expected silent close, got: {:?}", String::from_utf8_lossy(&resp)); + assert!(start.elapsed() < Duration::from_secs(3), "close took too long"); +}