From 0d704e50886b0dc3eb665a041c809b627aced4be Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Ralph=20K=C3=BCpper?= Date: Tue, 22 Sep 2026 19:33:49 +0200 Subject: [PATCH 1/8] Raise Event::Upgrade on the request side too `State::Upgrade` was reachable only in `Mode::Response`. A server decoding `GET / HTTP/1.1` with `Connection: upgrade` saw an ordinary head and `End`, so a host that matched on the public `Event::Upgrade` - the obvious thing to match on - served every WebSocket handshake as a normal request. Perry's P5 lane found the asymmetry only by reading the decoder. Of the two fixes the issue offers, this takes the first: the enum now means one thing in both modes. Documenting the asymmetry would have kept a public variant that is unreachable in the mode most servers run in, and every server would still have to re-derive the same RFC 9110 7.8 test from the headers. In `Mode::Request` a request ends with `Event::Upgrade` in place of `End` when it is HTTP/1.1 with an `Upgrade` header and the `upgrade` token in `Connection`, or when it is a CONNECT - the request-side mirror of the 101 and CONNECT-2xx cases the response side already had. It is raised after the request body, if any, and like `End` it reads no input, so the next protocol's bytes stay in the host's buffer. An `Upgrade` on an HTTP/1.0 request is ignored, as 7.8 requires. The decision stays with the caller. Unlike the response side, the decoder does not give up the connection: after a request-side `Upgrade` it is `reusable` on the same terms as after `End`, so a server that declines answers normally and calls `reset`. `Event::Upgrade` and `Mode` now say all of this. The in-crate drivers follow. `asynchronous::Http1` counts a request-side upgrade as the end of the message for `reusable`, and `reset` clears it. `server::http1` treats `Upgrade` as the end of the request; a service that declines keeps the connection, and one that answers 101 (or 2xx to CONNECT) now closes it, because that driver cannot hand the transport off and would otherwise parse the next protocol's bytes as a request. turnloop-websocket's `accept` already accepted `Upgrade | End`. Behaviour change: a request-mode host that matched only `End` must now also handle `Upgrade` for upgrade and CONNECT requests. Tests: `http1_request_side_raises_upgrade` (codecs) checks the event, its zero consumption, the untouched trailing bytes, body-then-upgrade, CONNECT, reuse after a decline, and the three non-upgrade shapes; before the fix every upgrade case ended in `End`. `http1_server_upgrade_request_is_the_services_decision` (asynchronous) drives `server::http1` over TCP: a declined upgrade serves the next request on the same connection, and an accepted one closes without parsing the bytes that follow - on the old code the service saw `End` and the driver failed parsing "not HTTP at all" as a request. The interop CONNECT proxy fixture now stops on `Upgrade` as well as `End`. Closes #46 --- .../turnloop-http/src/asynchronous/mod.rs | 5 +- .../turnloop-http/src/asynchronous/server.rs | 22 +++- protocols/turnloop-http/src/http1.rs | 48 ++++++++- protocols/turnloop-http/tests/asynchronous.rs | 101 ++++++++++++++++++ protocols/turnloop-http/tests/codecs.rs | 75 +++++++++++++ protocols/turnloop-http/tests/interop.rs | 2 +- 6 files changed, 248 insertions(+), 5 deletions(-) diff --git a/protocols/turnloop-http/src/asynchronous/mod.rs b/protocols/turnloop-http/src/asynchronous/mod.rs index 03dada0..4125f22 100644 --- a/protocols/turnloop-http/src/asynchronous/mod.rs +++ b/protocols/turnloop-http/src/asynchronous/mod.rs @@ -28,8 +28,10 @@ impl Http1 { upgraded: false, } } + /// A request that asked to upgrade ends the message too, and a server that + /// declined it may reuse the connection. A response upgrade never can. pub fn reusable(&self) -> bool { - self.stream.is_some() && self.ended && self.decoder.reusable() + self.stream.is_some() && (self.ended || self.upgraded) && self.decoder.reusable() } pub fn abort(&mut self) { self.stream.take(); @@ -37,6 +39,7 @@ impl Http1 { pub fn reset(&mut self) -> io::Result<()> { self.decoder.reset().map_err(io::Error::other)?; self.ended = false; + self.upgraded = false; Ok(()) } pub fn response_to(&mut self, method: &str) { diff --git a/protocols/turnloop-http/src/asynchronous/server.rs b/protocols/turnloop-http/src/asynchronous/server.rs index a2447ae..7f198e9 100644 --- a/protocols/turnloop-http/src/asynchronous/server.rs +++ b/protocols/turnloop-http/src/asynchronous/server.rs @@ -253,6 +253,7 @@ pub struct Response { output: Vec, finished: bool, keep_alive: bool, + status: u16, } impl Default for Response { fn default() -> Self { @@ -261,6 +262,7 @@ impl Default for Response { output: Vec::with_capacity(65536), finished: false, keep_alive: true, + status: 0, } } } @@ -275,6 +277,7 @@ impl Response { } self.encoder = Some(super::encode_head(head, length, &mut self.output)?); self.keep_alive = head.keep_alive && !head.token("connection", "close"); + self.status = head.status; Ok(()) } pub fn body(&mut self, bytes: &[u8]) -> io::Result<()> { @@ -299,6 +302,12 @@ impl Response { /// A connection that ends after a response (shutdown, `connection: close` or a /// non-reusable request) and an idle connection at shutdown close with a lingering /// close bounded by [`Options::linger_timeout`] and [`Shutdown::stop_by`]. +/// +/// A request asking to upgrade (or a CONNECT) ends with +/// [`Event::Upgrade`](crate::http1::Event::Upgrade) in place of `End`. A service +/// that declines answers normally and the connection stays HTTP/1. This driver +/// cannot hand the transport over, so a `101` (or a `2xx` to CONNECT) closes the +/// connection after the response. Serve upgrades from [`super::Http1`] directly. pub async fn http1( stream: S, shutdown: Shutdown, @@ -326,6 +335,7 @@ pub async fn http1( return Ok(()); }; let head = head?; + let connect = head.method == "CONNECT"; service(crate::http1::Event::Head(head), &mut response)?; turnloop_io::write_all( conn.stream.as_mut().ok_or_else(super::closed)?, @@ -333,11 +343,15 @@ pub async fn http1( ) .await?; response.output.clear(); + let mut upgrade = false; loop { let mut ended = false; let received = conn .event(|event| { - ended = matches!(event, crate::http1::Event::End); + // An upgrade request ends like any other; the service + // decides whether to switch by the status it answers. + upgrade = matches!(event, crate::http1::Event::Upgrade); + ended = upgrade || matches!(event, crate::http1::Event::End); service(event, &mut response) }) .await?; @@ -357,7 +371,11 @@ pub async fn http1( if !response.finished { return Err(io::Error::other("service did not finish response")); } - if shutdown.is_stopped() || !conn.reusable() || !response.keep_alive { + // This driver has no handoff, so an accepted upgrade ends the connection + // rather than parsing the next protocol's bytes as HTTP/1. + let switched = + upgrade && (response.status == 101 || connect && (200..300).contains(&response.status)); + if shutdown.is_stopped() || switched || !conn.reusable() || !response.keep_alive { linger(&mut conn.stream, &mut conn.input, &shutdown).await; return Ok(()); } diff --git a/protocols/turnloop-http/src/http1.rs b/protocols/turnloop-http/src/http1.rs index 737ced9..4e9bbb1 100644 --- a/protocols/turnloop-http/src/http1.rs +++ b/protocols/turnloop-http/src/http1.rs @@ -54,9 +54,19 @@ impl Head { .any(|h| tokens(&h.value).any(|t| t.eq_ignore_ascii_case(token.as_bytes()))) } } +/// Which side of the connection a [`Decoder`] reads. +/// +/// The two modes raise the same [`Event`]s, [`Event::Upgrade`] included: a +/// server decoding an upgrade or CONNECT request sees it just as a client +/// decoding the `101` or `2xx` does. What differs is who decides. In +/// [`Mode::Response`] the peer has already switched protocols, so the decoder +/// is finished with the connection. In [`Mode::Request`] the peer has only +/// asked, and the host answers. #[derive(Debug, Clone, Copy, PartialEq, Eq)] pub enum Mode { + /// Server side: decode requests. Request, + /// Client side: decode responses. Call [`Decoder::response_to`] first. Response, } #[derive(Debug, Clone, Copy, PartialEq, Eq)] @@ -79,7 +89,30 @@ pub enum Event<'a> { Informational(Head), Body(&'a [u8]), Trailers(Vec
), + /// The message is complete. Reads no input: see [`Step`]. End, + /// The message is complete, and it leaves HTTP/1 or asks to. Raised + /// *instead of* [`Event::End`], with `consumed == 0`, as the last event of + /// the message. Every byte after it belongs to the next protocol, so the + /// host takes its retained input along. + /// + /// Both [`Mode`]s raise it: + /// + /// * [`Mode::Response`]: a `101 Switching Protocols`, or a `2xx` answer to a + /// request passed to [`Decoder::response_to`] as `CONNECT`. The switch has + /// happened, so the decoder is never [`reusable`](Decoder::reusable) + /// afterwards. + /// * [`Mode::Request`]: an HTTP/1.1 request with an `Upgrade` header and + /// the `upgrade` token in `Connection` (RFC 9110 section 7.8), or any + /// `CONNECT` request. It follows the request body, if there is one. The + /// switch has **not** happened: answering `101` (or `2xx` for CONNECT) is + /// the host's decision. A host that accepts stops feeding this decoder. + /// A host that declines sends an ordinary response and may carry on: the + /// decoder is [`reusable`](Decoder::reusable) on the same terms as after + /// [`Event::End`], and [`Decoder::reset`] resumes HTTP/1. + /// + /// An `Upgrade` header on an HTTP/1.0 request is ignored, as RFC 9110 + /// section 7.8 requires, and that request ends with [`Event::End`]. Upgrade, } /// One decode step: how much of `input` was consumed, and the event it @@ -109,6 +142,8 @@ pub struct Decoder { head_request: bool, connect_request: bool, keep_alive: bool, + /// The request being read asks to leave HTTP/1, so it ends in `Upgrade`. + upgrade_request: bool, } fn invalid(message: &'static str) -> Error { Error::new("HPE_INVALID_HEADER_TOKEN", message) @@ -202,6 +237,7 @@ impl Decoder { head_request: false, connect_request: false, keep_alive: false, + upgrade_request: false, } } /// Must precede response parsing. Pipelining is intentionally not provided. @@ -288,6 +324,12 @@ impl Decoder { head.keep_alive = !head.token("connection", "close") && (head.version == 1 || head.token("connection", "keep-alive")); self.keep_alive = head.keep_alive; + // RFC 9110 7.8: an Upgrade in an HTTP/1.0 request is ignored. + self.upgrade_request = self.mode == Mode::Request + && (head.method == "CONNECT" + || (head.version == 1 + && head.get("upgrade").is_some() + && head.token("connection", "upgrade"))); step.consumed = end; if self.mode == Mode::Response && (100..200).contains(&head.status) @@ -430,7 +472,11 @@ impl Decoder { } State::End => { self.state = State::Done; - step.event = Some(Event::End); + step.event = Some(if self.upgrade_request { + Event::Upgrade + } else { + Event::End + }); } State::Upgrade => { self.state = State::Done; diff --git a/protocols/turnloop-http/tests/asynchronous.rs b/protocols/turnloop-http/tests/asynchronous.rs index 9e47111..ad5d1b2 100644 --- a/protocols/turnloop-http/tests/asynchronous.rs +++ b/protocols/turnloop-http/tests/asynchronous.rs @@ -1832,3 +1832,104 @@ fn http1_idle_keepalive_lingers_at_shutdown() { assert_eq!(gate.discarded.get(), PIPELINED.len()); assert!(gate.released.get()); } + +/// PerryTS/turnloop#46: an upgrade request reaches the service as +/// `Event::Upgrade`, and the service decides. Declined, the connection stays +/// HTTP/1 and serves the next request. Accepted, the driver closes rather than +/// parse the next protocol's bytes as a request. +#[test] +fn http1_server_upgrade_request_is_the_services_decision() { + let mut executor = LocalExecutor::::new(Config::default()).expect("executor"); + let h = executor.handle(); + let listener = Listener::bind(&h, "127.0.0.1:0".parse().expect("address")).expect("listen"); + let address = listener.local_addr().expect("address"); + let mut server = executor + .spawn_local(async move { + let mut events = Vec::new(); + for _ in 0..2 { + let stream = listener.accept().await.expect("accept"); + server::http1(stream, server::Shutdown::default(), |event, out| { + match event { + Event::Head(head) => { + events.push(format!("head {}", head.target)); + if head.target == "/accept" { + let head = Head { + status: 101, + headers: vec![ + Header::new("connection", "upgrade"), + Header::new("upgrade", "test"), + ], + ..response() + }; + out.start(&head, BodyLength::Empty)?; + } else { + let body = head.target.as_bytes(); + out.start(&response(), BodyLength::Known(body.len() as u64))?; + out.body(body)?; + } + } + Event::Upgrade => { + events.push("upgrade".into()); + out.finish(&[])?; + } + Event::End => { + events.push("end".into()); + out.finish(&[])?; + } + _ => {} + } + Ok(()) + }) + .await + .expect("the connection ends cleanly"); + } + events + }) + .expect("spawn"); + let mut client = executor + .spawn_local(async move { + let mut responses = Vec::new(); + for wire in [ + b"GET /decline HTTP/1.1\r\nhost: l\r\nconnection: upgrade\r\nupgrade: test\r\n\r\n\ + GET /next HTTP/1.1\r\nhost: l\r\nconnection: close\r\n\r\n" + .as_slice(), + b"GET /accept HTTP/1.1\r\nhost: l\r\nconnection: upgrade\r\nupgrade: test\r\n\r\n\ + not HTTP at all\r\n\r\n", + ] { + let mut stream = h + .connect(address, Default::default()) + .await + .expect("connect"); + write_all(&mut stream, wire).await.expect("request"); + responses.push(read_until_eof(&mut stream).await); + close(&mut stream).await.expect("close"); + } + responses + }) + .expect("spawn"); + run_until(&mut executor, &mut server, &mut client); + let responses = finish(&mut client); + assert_eq!( + finish(&mut server), + [ + "head /decline", + "upgrade", + "head /next", + "end", + "head /accept", + "upgrade" + ] + ); + let declined = String::from_utf8(responses[0].clone()).expect("ASCII"); + assert!(declined.starts_with("HTTP/1.1 200"), "{declined}"); + assert!( + declined.contains("\r\n\r\n/declineHTTP/1.1 200") && declined.ends_with("\r\n\r\n/next"), + "{declined}" + ); + let accepted = String::from_utf8(responses[1].clone()).expect("ASCII"); + assert!(accepted.starts_with("HTTP/1.1 101"), "{accepted}"); + assert!( + accepted.ends_with("\r\n\r\n"), + "nothing after the switch: {accepted}" + ); +} diff --git a/protocols/turnloop-http/tests/codecs.rs b/protocols/turnloop-http/tests/codecs.rs index d60f719..5230a0a 100644 --- a/protocols/turnloop-http/tests/codecs.rs +++ b/protocols/turnloop-http/tests/codecs.rs @@ -1309,3 +1309,78 @@ fn step_contract_rejects_both_wrong_loop_conditions() { } assert!(ended); } + +/// Drive a request decoder to its stop step with the documented loop, naming +/// each event and its `consumed`. Returns the names and the retained bytes. +fn request_events(decoder: &mut Decoder, wire: &[u8]) -> (Vec, Vec) { + let mut wire = wire.to_vec(); + let mut names = Vec::new(); + loop { + let step = decoder.receive(&wire).unwrap(); + let consumed = step.consumed; + let progressed = consumed > 0 || step.event.is_some(); + match step.event { + Some(Event::Head(h)) => names.push(format!("head {} {consumed}", h.method)), + Some(Event::Body(b)) => names.push(format!("body {} {consumed}", b.len())), + Some(Event::End) => names.push(format!("end {consumed}")), + Some(Event::Upgrade) => names.push(format!("upgrade {consumed}")), + Some(other) => names.push(format!("{other:?}")), + None => {} + } + wire.drain(..consumed); + if !progressed { + return (names, wire); + } + } +} + +/// PerryTS/turnloop#46: a server decoding an upgrade request sees +/// `Event::Upgrade`, as a client decoding the `101` does, and the bytes after it +/// stay with the host. The decision is the host's, so a decline keeps HTTP/1. +#[test] +fn http1_request_side_raises_upgrade() { + let ws = b"GET /chat HTTP/1.1\r\nHost: a\r\nConnection: keep-alive, Upgrade\r\n\ + Upgrade: websocket\r\n\r\n\x81\x05hello"; + let mut decoder = Decoder::new(Mode::Request, Limits::default()); + let (names, rest) = request_events(&mut decoder, ws); + assert_eq!(names, ["head GET 84", "upgrade 0"]); + assert_eq!( + rest, b"\x81\x05hello", + "the next protocol's bytes are untouched" + ); + // Declining is ordinary: the decoder is reusable like after `End`. + assert!(decoder.reusable()); + decoder.reset().unwrap(); + let (names, _) = request_events(&mut decoder, b"GET / HTTP/1.1\r\nHost: a\r\n\r\n"); + assert_eq!(names, ["head GET 27", "end 0"], "the flag does not leak"); + + // A request body is delivered first; the upgrade follows it. + let mut decoder = Decoder::new(Mode::Request, Limits::default()); + let (names, rest) = request_events( + &mut decoder, + b"POST / HTTP/1.1\r\nHost: a\r\nConnection: upgrade\r\nUpgrade: h2c\r\n\ + Content-Length: 3\r\n\r\nabcNEXT", + ); + assert_eq!(names, ["head POST 82", "body 3 3", "upgrade 0"]); + assert_eq!(rest, b"NEXT"); + + // CONNECT asks to leave HTTP/1 as well. + let mut decoder = Decoder::new(Mode::Request, Limits::default()); + let (names, rest) = request_events( + &mut decoder, + b"CONNECT a:443 HTTP/1.1\r\nHost: a:443\r\n\r\n\x16\x03\x01", + ); + assert_eq!(names, ["head CONNECT 39", "upgrade 0"]); + assert_eq!(rest, b"\x16\x03\x01"); + + // Neither half alone is an upgrade, and HTTP/1.0 ignores it (RFC 9110 7.8). + for wire in [ + b"GET / HTTP/1.1\r\nHost: a\r\nUpgrade: websocket\r\n\r\n".as_slice(), + b"GET / HTTP/1.1\r\nHost: a\r\nConnection: upgrade\r\n\r\n", + b"GET / HTTP/1.0\r\nConnection: upgrade\r\nUpgrade: websocket\r\n\r\n", + ] { + let mut decoder = Decoder::new(Mode::Request, Limits::default()); + let (names, _) = request_events(&mut decoder, wire); + assert_eq!(names.last().map(String::as_str), Some("end 0"), "{names:?}"); + } +} diff --git a/protocols/turnloop-http/tests/interop.rs b/protocols/turnloop-http/tests/interop.rs index ea29b00..dd2499f 100644 --- a/protocols/turnloop-http/tests/interop.rs +++ b/protocols/turnloop-http/tests/interop.rs @@ -116,7 +116,7 @@ fn response(stream: &mut impl Read, decoder: &mut http1::Decoder) -> (Head, Vec< loop { let step = decoder.receive(&buffer).unwrap(); let progressed = step.consumed > 0 || step.event.is_some(); - let end = matches!(step.event, Some(http1::Event::End)); + let end = matches!(step.event, Some(http1::Event::End | http1::Event::Upgrade)); match step.event { Some(http1::Event::Head(h)) => head = Some(h), Some(http1::Event::Body(bytes)) => body.extend_from_slice(bytes), From 8b42c026c11225f602ec16e34386741586e61b32 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Ralph=20K=C3=BCpper?= Date: Tue, 22 Sep 2026 19:35:08 +0200 Subject: [PATCH 2/8] Let http1::Encoder emit a reason phrase, EOF framing and HEAD/304 heads Three response shapes Node puts on the wire could not go through the encoder, and Perry's P5 lane wrote the head by hand for each: - `Encoder::start` always wrote the canonical reason phrase, so `res.writeHead(404, "Nope")` meant patching the status line afterwards by re-finding the CRLF in the encoder's own output. - `BodyLength` had no close-delimited framing. - A HEAD response advertises the `Content-Length` it would have sent and sends no body. `Known(0)` rejected that head as a conflict with the header, and `Known(n)` refused to `finish` without n bytes. Now: - `Encoder::start_with_reason(head, reason, body, out)` writes the given phrase. It is validated against RFC 9112's reason-phrase grammar (tab, space, visible ASCII, obs-text) before anything is written, so a phrase cannot inject a header line, and it is refused on a request. `start` is unchanged. A separate constructor rather than a field on `Head` keeps every existing `Head { .. }` literal compiling, and a decoded head has no reason to carry. - `BodyLength::CloseDelimited`: responses only, no framing header, body bytes written raw, `finish` writes nothing and the host's close ends the body. A head with `Content-Length` or `Transfer-Encoding`, a request, and a 1xx/204/304 status are refused, as are trailers. - `BodyLength::Omitted`: responses only; the head's framing headers go out verbatim, none is added, and no body follows. That is a response to HEAD (length or chunked coding advertised) and a 304. A 204 or 1xx may not advertise a length at all (RFC 9110 8.6), so a framing header there is still refused. Of the issue's two shapes for this - a variant or a `head_response` flag on `start` - the variant fits: body framing is exactly what `BodyLength` already chooses, and a flag would be a second argument that changes what the first means. Breaking: `BodyLength` gains two variants, so an exhaustive `match` on it needs two more arms. No in-tree match was exhaustive. Tests: `http1_encoder_writes_a_custom_reason_phrase`, `http1_encoder_writes_a_close_delimited_body` and `http1_encoder_writes_a_body_forbidden_response` assert the exact wire bytes, decode the close-delimited and HEAD responses back with `Decoder` (a body ended by `eof`, and a HEAD response that ends at its head and stays reusable), and check every refusal leaves the output untouched. Closes #47 --- protocols/turnloop-http/src/http1.rs | 89 +++++++++++--- protocols/turnloop-http/tests/codecs.rs | 150 ++++++++++++++++++++++++ 2 files changed, 222 insertions(+), 17 deletions(-) diff --git a/protocols/turnloop-http/src/http1.rs b/protocols/turnloop-http/src/http1.rs index 4e9bbb1..67a5129 100644 --- a/protocols/turnloop-http/src/http1.rs +++ b/protocols/turnloop-http/src/http1.rs @@ -499,11 +499,31 @@ impl Decoder { } } } +/// How the body after an encoded head is framed. #[derive(Debug, Clone, Copy)] pub enum BodyLength { + /// No body. A `Content-Length` in the head, if any, must be `0`. Empty, + /// Exactly this many bytes, framed by `Content-Length`. Known(u64), + /// `Transfer-Encoding: chunked`; the only framing that carries trailers. Chunked, + /// Responses only: no framing, the body ends when the connection closes + /// (RFC 9112 section 6.3, rule 8). The head must carry neither + /// `Content-Length` nor `Transfer-Encoding`, and after + /// [`finish`](Encoder::finish) the host closes the connection - that close + /// *is* the end of the body. Adding `connection: close` is the host's + /// choice; the peer reads to EOF either way. + CloseDelimited, + /// Responses only: the head's framing headers are sent exactly as given, + /// and no body follows them. This is a response to HEAD, which advertises + /// the `Content-Length` (or chunked coding) it *would* have sent, and a + /// `304`, which may do the same. A `204` or `1xx` may not advertise a length + /// at all (RFC 9110 sections 8.6 and 15.2), so either header is refused + /// there. The encoder cannot see the request method, so a HEAD response is + /// the host's word. No framing header is added, and [`body`](Encoder::body) + /// accepts only empty slices. + Omitted, } /// Caller-owned wire buffer, reused across commands. No body buffering. pub struct Encoder { @@ -511,23 +531,62 @@ pub struct Encoder { finished: bool, } impl Encoder { + /// Encode a head. A response's status line uses the canonical reason + /// phrase; [`start_with_reason`](Self::start_with_reason) sets another one. pub fn start(head: &Head, body: BodyLength, out: &mut Vec) -> Result { + Self::encode(head, None, body, out) + } + /// Encode a response head with a custom reason phrase, as Node's + /// `res.writeHead(404, "Nope")` puts `HTTP/1.1 404 Nope` on the wire. The + /// phrase may be empty, and may hold spaces, tabs, visible ASCII and + /// obs-text (RFC 9112 section 4); a request head or a control character is + /// refused before anything is written. + pub fn start_with_reason( + head: &Head, + reason: &str, + body: BodyLength, + out: &mut Vec, + ) -> Result { + if head.status == 0 { + return Err(invalid("reason phrase on a request")); + } + if reason.bytes().any(|b| (b < 32 && b != b'\t') || b == 127) { + return Err(invalid("invalid reason phrase")); + } + Self::encode(head, Some(reason), body, out) + } + fn encode( + head: &Head, + reason: Option<&str>, + body: BodyLength, + out: &mut Vec, + ) -> Result { // Validate everything before mutating the output. for h in &head.headers { validate_field(h)?; } let (cl, te) = lengths(head)?; - let expect_cl = match body { - BodyLength::Known(n) => Some(n), - BodyLength::Empty => Some(0), - BodyLength::Chunked => None, + let conflict = match body { + BodyLength::Known(n) => te || cl.is_some_and(|cl| cl != n), + BodyLength::Empty => te || cl.is_some_and(|cl| cl != 0), + BodyLength::Chunked => cl.is_some(), + BodyLength::CloseDelimited => te || cl.is_some(), + BodyLength::Omitted => { + (cl.is_some() || te) && (head.status < 200 || head.status == 204) + } }; - if (cl.is_some() && cl != expect_cl) - || (te && !matches!(body, BodyLength::Chunked)) - || (cl.is_some() && matches!(body, BodyLength::Chunked)) - { + if conflict { return Err(invalid("body length conflicts with headers")); } + let response_only = matches!(body, BodyLength::CloseDelimited | BodyLength::Omitted); + if response_only && head.status == 0 { + return Err(invalid("body length is only valid for a response")); + } + if matches!(body, BodyLength::CloseDelimited) + && (head.status < 200 || matches!(head.status, 204 | 304)) + { + return Err(invalid("status forbids a body")); + } if head.status == 0 { if !valid_token(head.method.as_bytes()) { return Err(invalid("invalid method")); @@ -539,13 +598,8 @@ impl Encoder { } else { let status = http::StatusCode::from_u16(head.status).map_err(|_| invalid("invalid status"))?; - write!( - out, - "HTTP/1.1 {} {}\r\n", - head.status, - status.canonical_reason().unwrap_or("") - ) - .unwrap(); + let reason = reason.unwrap_or(status.canonical_reason().unwrap_or("")); + write!(out, "HTTP/1.1 {} {reason}\r\n", head.status).unwrap(); } for h in &head.headers { out.extend_from_slice(h.name.as_bytes()); @@ -557,7 +611,7 @@ impl Encoder { match body { BodyLength::Known(n) => write!(out, "content-length: {n}\r\n").unwrap(), BodyLength::Chunked => out.extend_from_slice(b"transfer-encoding: chunked\r\n"), - BodyLength::Empty => {} + BodyLength::Empty | BodyLength::CloseDelimited | BodyLength::Omitted => {} } } out.extend_from_slice(b"\r\n"); @@ -571,11 +625,12 @@ impl Encoder { return Err(invalid("body already finished")); } match self.left { - BodyLength::Empty => { + BodyLength::Empty | BodyLength::Omitted => { if !bytes.is_empty() { return Err(invalid("unexpected body")); } } + BodyLength::CloseDelimited => out.extend_from_slice(bytes), BodyLength::Known(n) => { if bytes.len() as u64 > n { return Err(invalid("body exceeds content-length")); diff --git a/protocols/turnloop-http/tests/codecs.rs b/protocols/turnloop-http/tests/codecs.rs index 5230a0a..589816b 100644 --- a/protocols/turnloop-http/tests/codecs.rs +++ b/protocols/turnloop-http/tests/codecs.rs @@ -1384,3 +1384,153 @@ fn http1_request_side_raises_upgrade() { assert_eq!(names.last().map(String::as_str), Some("end 0"), "{names:?}"); } } + +fn status_head(status: u16, headers: &[(&str, &str)]) -> Head { + Head { + method: String::new(), + target: String::new(), + status, + version: 1, + headers: headers.iter().map(|(n, v)| Header::new(n, v)).collect(), + keep_alive: true, + } +} + +/// PerryTS/turnloop#47, part 1: Node's `res.writeHead(404, "Nope")`. +#[test] +fn http1_encoder_writes_a_custom_reason_phrase() { + let head = status_head(404, &[]); + let mut wire = Vec::new(); + Encoder::start_with_reason(&head, "Nope", BodyLength::Empty, &mut wire).unwrap(); + assert_eq!(wire, b"HTTP/1.1 404 Nope\r\n\r\n"); + wire.clear(); + Encoder::start_with_reason(&head, "", BodyLength::Empty, &mut wire).unwrap(); + assert_eq!(wire, b"HTTP/1.1 404 \r\n\r\n", "an empty phrase is legal"); + wire.clear(); + Encoder::start(&head, BodyLength::Empty, &mut wire).unwrap(); + assert_eq!( + wire, b"HTTP/1.1 404 Not Found\r\n\r\n", + "start keeps the canonical one" + ); + // Refused before anything is written: a phrase that would inject a header, + // and a phrase on a request line, which has no place for one. + wire.clear(); + for reason in ["a\r\nset-cookie: x", "a\nb", "nul\0"] { + assert!( + Encoder::start_with_reason(&head, reason, BodyLength::Empty, &mut wire).is_err(), + "{reason:?}" + ); + } + let request = Request::new("http://localhost/", "GET") + .unwrap() + .head(false); + assert!(Encoder::start_with_reason(&request, "OK", BodyLength::Empty, &mut wire).is_err()); + assert!(wire.is_empty()); +} + +/// PerryTS/turnloop#47, part 2: a body that ends at EOF, as an HTTP/1.0-style +/// response with no framing does. Decoded back, it is one body ended by `eof`. +#[test] +fn http1_encoder_writes_a_close_delimited_body() { + let head = status_head(200, &[("content-type", "text/plain")]); + let mut wire = Vec::new(); + let mut encoder = Encoder::start(&head, BodyLength::CloseDelimited, &mut wire).unwrap(); + encoder.body(b"until ", &mut wire).unwrap(); + encoder.body(b"close", &mut wire).unwrap(); + encoder.finish(&[], &mut wire).unwrap(); + assert_eq!( + wire, b"HTTP/1.1 200 OK\r\ncontent-type: text/plain\r\n\r\nuntil close", + "no framing header, no chunk markup, nothing after the body" + ); + let mut decoder = Decoder::new(Mode::Response, Limits::default()); + decoder.response_to("GET"); + let mut body = Vec::new(); + let mut input = wire.as_slice(); + loop { + let step = decoder.receive(input).unwrap(); + if let Some(Event::Body(b)) = step.event { + body.extend_from_slice(b); + } + input = &input[step.consumed..]; + if input.is_empty() { + break; + } + } + decoder.eof().unwrap(); + assert!(matches!( + decoder.receive(&[]).unwrap().event, + Some(Event::End) + )); + assert_eq!(body, b"until close"); + + // Trailers need chunked framing; framing headers contradict EOF framing; + // a request cannot be close-delimited; 204 and 304 carry no body. + let mut encoder = Encoder::start(&head, BodyLength::CloseDelimited, &mut wire).unwrap(); + assert!(encoder.finish(&[Header::new("x", "y")], &mut wire).is_err()); + wire.clear(); + for bad in [ + status_head(200, &[("content-length", "5")]), + status_head(200, &[("transfer-encoding", "chunked")]), + status_head(204, &[]), + status_head(304, &[]), + Request::new("http://localhost/", "POST") + .unwrap() + .head(false), + ] { + assert!(Encoder::start(&bad, BodyLength::CloseDelimited, &mut wire).is_err()); + } + assert!(wire.is_empty()); +} + +/// PerryTS/turnloop#47, part 3: a response that advertises a length and sends +/// no body. `Known(0)` and `Known(n)` both refuse this head; `Omitted` writes +/// it verbatim, and a client decoding a HEAD response reads it as complete. +#[test] +fn http1_encoder_writes_a_body_forbidden_response() { + let head = status_head(200, &[("content-length", "1234")]); + let mut wire = Vec::new(); + assert!(Encoder::start(&head, BodyLength::Known(0), &mut wire).is_err()); + assert!(Encoder::start(&head, BodyLength::Empty, &mut wire).is_err()); + let mut encoder = Encoder::start(&head, BodyLength::Omitted, &mut wire).unwrap(); + assert!(encoder.body(b"x", &mut wire).is_err()); + encoder.body(b"", &mut wire).unwrap(); + encoder.finish(&[], &mut wire).unwrap(); + assert_eq!(wire, b"HTTP/1.1 200 OK\r\ncontent-length: 1234\r\n\r\n"); + let mut decoder = Decoder::new(Mode::Response, Limits::default()); + decoder.response_to("HEAD"); + let head_step = decoder.receive(&wire).unwrap(); + assert_eq!(head_step.consumed, wire.len()); + assert!(matches!( + decoder.receive(&[]).unwrap().event, + Some(Event::End) + )); + assert!(decoder.reusable()); + + // A HEAD response may advertise chunked coding, and a 304 its length. + for head in [ + status_head(200, &[("transfer-encoding", "chunked")]), + status_head(304, &[("content-length", "1234")]), + status_head(204, &[]), + ] { + wire.clear(); + Encoder::start(&head, BodyLength::Omitted, &mut wire) + .unwrap() + .finish(&[], &mut wire) + .unwrap(); + assert!(wire.ends_with(b"\r\n\r\n")); + assert!(!wire.windows(2).any(|w| w == b"0\r"), "no chunk markup"); + } + // A 204 or 1xx may not advertise one (RFC 9110 8.6), and a request is not + // a response to HEAD. + wire.clear(); + for bad in [ + status_head(204, &[("content-length", "0")]), + status_head(103, &[("content-length", "4")]), + Request::new("http://localhost/", "GET") + .unwrap() + .head(false), + ] { + assert!(Encoder::start(&bad, BodyLength::Omitted, &mut wire).is_err()); + } + assert!(wire.is_empty()); +} From d0b8702fac7681289bf710be54c5548f6f7c3c44 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Ralph=20K=C3=BCpper?= Date: Tue, 22 Sep 2026 19:36:17 +0200 Subject: [PATCH 3/8] Add http1::Decoder::wants_step for the zero-byte End step `Event::End` (and `Event::Upgrade`) come from a step that consumes nothing: the step that consumes the last byte of a message only moves the decoder to `End`, and the next call delivers it. A host that feeds the decoder "while there is input" hands over every byte and stops, so the message sits finished inside the decoder. Two lanes lost a debugging cycle to it: P6's client took ~5 s per fetch with no connection reuse, and P11's first real-server probe timed out on every URL - the peer's idle timeout was the only thing ending the exchange. #85 stated the `Step` contract (loop while `consumed > 0 || event.is_some()`) in the crate root and on both `Step` types. The issue's comment argues that two independent consumers hitting the same trap is the case for a predicate over another note, and a host keyed on input rather than on steps has no reason to read the loop condition anyway. So both: - `Decoder::wants_step()` is true exactly when a zero-input event is pending (`State::End` or `State::Upgrade`): after the last body byte or final chunk, after the head of a bodiless message, and after `eof` ends a close-delimited body. `while !input.is_empty() || decoder.wants_step()` is a correct loop. - `http1::Step`'s documentation now carries a runnable loop, as `http2::Step`'s does, and a paragraph on the zero-consumption trap naming `wants_step`. The crate-root contract names the "while it has input" loop beside the "while consumed" one. Tests: `http1_end_needs_a_step_after_the_last_byte` runs the loop both lanes wrote over a Content-Length, a chunked and a 204 response, and asserts it consumes every byte yet never sees `End` and leaves the decoder unreusable - the P6 symptom - while `wants_step` is true. The same loop keyed on `wants_step` too reaches `End` from a zero-consumption step, leaves the decoder reusable, and ends with `wants_step` false. It also covers `eof` on a close-delimited body and a request-side upgrade. The existing `step_contract_rejects_both_wrong_loop_conditions` still covers the "stop when nothing was consumed" loop. Closes #50 --- protocols/turnloop-http/src/http1.rs | 48 ++++++++++++++++- protocols/turnloop-http/src/lib.rs | 6 ++- protocols/turnloop-http/tests/codecs.rs | 71 +++++++++++++++++++++++++ 3 files changed, 122 insertions(+), 3 deletions(-) diff --git a/protocols/turnloop-http/src/http1.rs b/protocols/turnloop-http/src/http1.rs index 67a5129..3553e8c 100644 --- a/protocols/turnloop-http/src/http1.rs +++ b/protocols/turnloop-http/src/http1.rs @@ -128,7 +128,37 @@ pub enum Event<'a> { /// | `0` | `Some` | an event that reads no input: [`Event::End`] and [`Event::Upgrade`] both arrive this way. | /// /// So the loop condition is `consumed > 0 || event.is_some()`, and a host that -/// stops on either zero alone is wrong in one of the two directions. +/// stops on either zero alone is wrong in one of the two directions: +/// +/// ``` +/// # use turnloop_http::http1::{Decoder, Event, Mode}; +/// # let mut decoder = Decoder::new(Mode::Response, Default::default()); +/// # decoder.response_to("GET"); +/// # let mut input = b"HTTP/1.1 200 OK\r\ncontent-length: 2\r\n\r\nhi".to_vec(); +/// let mut ended = false; +/// loop { +/// let step = decoder.receive(&input)?; +/// let consumed = step.consumed; +/// let progressed = consumed > 0 || step.event.is_some(); +/// // Body borrows `input`, so handle the event before draining. +/// ended |= matches!(step.event, Some(Event::End)); +/// input.drain(..consumed); +/// if !progressed { +/// break; // read more bytes from the transport, then continue +/// } +/// } +/// // `End` came from a step after the last byte was consumed. +/// assert!(ended && input.is_empty() && decoder.reusable()); +/// # Ok::<(), turnloop_http::Error>(()) +/// ``` +/// +/// **The zero-consumption event is the trap.** The step that consumes the last +/// byte of a message does not carry `End`; the *next* call does, and it reads +/// nothing. A host that loops while it has input - or stops when a step +/// consumed nothing - never makes that call, and the finished message sits in +/// the decoder until the peer's idle timeout closes the connection. +/// [`Decoder::wants_step`] names the pending call for a host that keys its +/// loop on input rather than on steps. #[derive(Debug)] pub struct Step<'a> { pub consumed: usize, @@ -248,6 +278,22 @@ impl Decoder { pub fn reusable(&self) -> bool { self.state == State::Done && self.keep_alive } + /// True when the decoder holds an event that reads no input, so the host + /// must call [`receive`](Self::receive) again even with nothing new to feed + /// it - an empty slice will do. That event is [`Event::End`] or + /// [`Event::Upgrade`]. It becomes pending when the last body byte, the + /// final chunk or the head of a bodiless message is consumed, and after + /// [`eof`](Self::eof) ends a close-delimited body. + /// + /// This is the signal a host that drives on "bytes to feed" is missing. + /// Such a host stops once its buffer is empty and never sees `End`: the + /// message sits finished inside the decoder, the connection is not + /// [`reusable`](Self::reusable), and only the peer's idle timeout ends the + /// exchange. Loop while `!input.is_empty() || decoder.wants_step()`, or use + /// the [`Step`] contract's condition, which covers the same case. + pub fn wants_step(&self) -> bool { + matches!(self.state, State::End | State::Upgrade) + } pub fn reset(&mut self) -> Result<()> { if !self.reusable() { return Err(invalid("connection is not reusable")); diff --git a/protocols/turnloop-http/src/lib.rs b/protocols/turnloop-http/src/lib.rs index 9fec9e3..464e997 100644 --- a/protocols/turnloop-http/src/lib.rs +++ b/protocols/turnloop-http/src/lib.rs @@ -31,8 +31,10 @@ //! starts at all. //! * **`consumed == 0` with an event** is an event that reads no input. //! HTTP/1's [`http1::Event::End`] and [`http1::Event::Upgrade`] both arrive -//! this way. A loop that continues only while input was consumed drops the end -//! of every message. HTTP/2 has no step of this shape: there an event always +//! this way. A loop that continues only while input was consumed - or only +//! while it has input to feed - drops the end of every message; +//! [`http1::Decoder::wants_step`] says when such an event is pending. HTTP/2 +//! has no step of this shape: there an event always //! consumes, and `consumed == 0` is always the stop case. //! * **`consumed == 0` with no event** is the only stop condition, and it is //! returned whether the decoder needs more bytes or is finished for good. The diff --git a/protocols/turnloop-http/tests/codecs.rs b/protocols/turnloop-http/tests/codecs.rs index 589816b..02be791 100644 --- a/protocols/turnloop-http/tests/codecs.rs +++ b/protocols/turnloop-http/tests/codecs.rs @@ -1396,6 +1396,77 @@ fn status_head(status: u16, headers: &[(&str, &str)]) -> Head { } } +/// PerryTS/turnloop#50, as P6 and P11 hit it: a host that feeds the decoder +/// "while there is input" hands over every byte of a keep-alive response and +/// never sees `End`, because `End` comes from a later step that reads nothing. +/// `wants_step` is the signal that step is owed. +#[test] +fn http1_end_needs_a_step_after_the_last_byte() { + let responses: [&[u8]; 3] = [ + b"HTTP/1.1 200 OK\r\ncontent-length: 5\r\n\r\nhello", + b"HTTP/1.1 200 OK\r\ntransfer-encoding: chunked\r\n\r\n5\r\nhello\r\n0\r\n\r\n", + b"HTTP/1.1 204 No Content\r\n\r\n", + ]; + for wire in responses { + // The loop both lanes wrote: stop once every byte is handed over. + let mut decoder = Decoder::new(Mode::Response, Limits::default()); + decoder.response_to("GET"); + let mut input = wire.to_vec(); + let mut ended = false; + while !input.is_empty() { + let step = decoder.receive(&input).unwrap(); + ended |= matches!(step.event, Some(Event::End)); + input.drain(..step.consumed); + } + assert!(!ended, "every byte consumed, and still no End"); + assert!(!decoder.reusable(), "so the connection cannot be pooled"); + assert!(decoder.wants_step(), "but the decoder says a step is owed"); + + // The same loop, keyed on wants_step as well, completes the message. + let mut decoder = Decoder::new(Mode::Response, Limits::default()); + decoder.response_to("GET"); + let mut input = wire.to_vec(); + let mut events = Vec::new(); + while !input.is_empty() || decoder.wants_step() { + let step = decoder.receive(&input).unwrap(); + events.push((step.consumed, matches!(step.event, Some(Event::End)))); + input.drain(..step.consumed); + } + assert_eq!(events.last(), Some(&(0, true)), "End reads nothing"); + assert!(decoder.reusable()); + assert!(!decoder.wants_step(), "and nothing more is owed"); + } + + // `eof` ending a close-delimited body also leaves `End` owed. + let mut decoder = Decoder::new(Mode::Response, Limits::default()); + decoder.response_to("GET"); + let head = b"HTTP/1.1 200 OK\r\n\r\n"; + assert_eq!(decoder.receive(head).unwrap().consumed, head.len()); + assert_eq!(decoder.receive(b"body").unwrap().consumed, 4); + assert!( + !decoder.wants_step(), + "a close-delimited body is open until EOF" + ); + decoder.eof().unwrap(); + assert!(decoder.wants_step()); + assert!(matches!( + decoder.receive(&[]).unwrap().event, + Some(Event::End) + )); + assert!(!decoder.wants_step()); + + // And an upgrade, on either side, is the same shape. + let mut decoder = Decoder::new(Mode::Request, Limits::default()); + let wire = b"GET / HTTP/1.1\r\nHost: a\r\nConnection: upgrade\r\nUpgrade: x\r\n\r\n"; + assert_eq!(decoder.receive(wire).unwrap().consumed, wire.len()); + assert!(decoder.wants_step()); + assert!(matches!( + decoder.receive(&[]).unwrap().event, + Some(Event::Upgrade) + )); + assert!(!decoder.wants_step()); +} + /// PerryTS/turnloop#47, part 1: Node's `res.writeHead(404, "Nope")`. #[test] fn http1_encoder_writes_a_custom_reason_phrase() { From d41fe24a1b0dc8cda35537ea1def4d3febfd6abd Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Ralph=20K=C3=BCpper?= Date: Tue, 22 Sep 2026 19:36:59 +0200 Subject: [PATCH 4/8] Let a host ask http1::Decoder whether it is mid-message The only way to learn whether a zero-byte read was a legal end of body or a truncated response was to call `eof()` and see whether it failed - and `eof()` commits: on the wrong answer it has already failed the decoder. Perry's P11 client needs the answer on every connection, before choosing between retrying, failing with its own diagnosis, or logging. The issue offers a `wants_step()`-style predicate or an explicit query. `wants_step` (#50) answers a different question - whether an event is owed - so this adds the explicit pair, both read-only: - `is_mid_message()`: a final head has been decoded and the end of the message has not been reached (fixed-length or close-delimited body, chunk framing, trailers). An informational response is not the message. - `eof_is_clean()`: true exactly where `eof()` would return `Ok` - inside a close-delimited body and once the message is complete. `eof()` is now written in terms of it, so the two cannot drift. Two predicates rather than one because the question P11 asks has three answers: not in a message, in a message where EOF is truncation, and in a close-delimited body where EOF is the end. Test: `http1_mid_message_is_queryable_without_feeding_eof` walks a Content-Length, a close-delimited and a chunked response through each state and asserts both answers, that asking changes nothing (the body continues to decode afterwards), that a 100 Continue is not mid-message, and that `eof_is_clean() == false` is followed by `eof()` failing. Closes #80 --- protocols/turnloop-http/src/http1.rs | 49 +++++++++++++++--- protocols/turnloop-http/tests/codecs.rs | 66 +++++++++++++++++++++++++ 2 files changed, 109 insertions(+), 6 deletions(-) diff --git a/protocols/turnloop-http/src/http1.rs b/protocols/turnloop-http/src/http1.rs index 3553e8c..5156401 100644 --- a/protocols/turnloop-http/src/http1.rs +++ b/protocols/turnloop-http/src/http1.rs @@ -294,6 +294,41 @@ impl Decoder { pub fn wants_step(&self) -> bool { matches!(self.state, State::End | State::Upgrade) } + /// True between a message's [`Event::Head`] and the point its end is + /// reached: the body, chunk framing or trailers are still being read. False + /// before the head (including after an [`Event::Informational`], which is + /// not the message), once `End` or `Upgrade` is owed or delivered, and after + /// a failure. Bytes of a head the host has buffered but not yet completed + /// are the host's to know; the decoder has not accepted them. + /// + /// A close-delimited body is mid-message too, though EOF is its legal end: + /// ask [`eof_is_clean`](Self::eof_is_clean) to tell the two apart. + pub fn is_mid_message(&self) -> bool { + matches!( + self.state, + State::Fixed(_) + | State::ChunkSize + | State::Chunk(_) + | State::ChunkEnd + | State::Trailers + | State::Eof + ) + } + /// Whether the transport ending now would be a clean end rather than a + /// truncation: exactly the cases in which [`eof`](Self::eof) returns `Ok`, + /// answered without changing any state. True inside a close-delimited body + /// and once the message is complete; false inside a framed body, before a + /// head arrives and after a failure. So a host whose read returned zero can + /// choose what to do - retry, fail with its own diagnosis, log - before it + /// commits to `eof`. + /// + /// Before a head is always "not clean" here, because a client that sees EOF + /// there got no response. A server whose connection closes while idle + /// between requests (with no buffered head bytes) treats that as an + /// ordinary close on its own. + pub fn eof_is_clean(&self) -> bool { + matches!(self.state, State::Eof | State::End | State::Done) + } pub fn reset(&mut self) -> Result<()> { if !self.reusable() { return Err(invalid("connection is not reusable")); @@ -533,16 +568,18 @@ impl Decoder { } Ok(step) } + /// Declare the end of input. Ends a close-delimited body (leaving + /// [`Event::End`] owed, see [`wants_step`](Self::wants_step)); anywhere + /// [`eof_is_clean`](Self::eof_is_clean) is false it fails the decoder. pub fn eof(&mut self) -> Result<()> { + if !self.eof_is_clean() { + self.state = State::Failed; + return Err(Error::new("UND_ERR_SOCKET", "unexpected EOF")); + } if self.state == State::Eof { self.state = State::End; - Ok(()) - } else if matches!(self.state, State::End | State::Done) { - Ok(()) - } else { - self.state = State::Failed; - Err(Error::new("UND_ERR_SOCKET", "unexpected EOF")) } + Ok(()) } } /// How the body after an encoded head is framed. diff --git a/protocols/turnloop-http/tests/codecs.rs b/protocols/turnloop-http/tests/codecs.rs index 02be791..2c8df95 100644 --- a/protocols/turnloop-http/tests/codecs.rs +++ b/protocols/turnloop-http/tests/codecs.rs @@ -1467,6 +1467,72 @@ fn http1_end_needs_a_step_after_the_last_byte() { assert!(!decoder.wants_step()); } +/// PerryTS/turnloop#80: when a read returns zero, P11 must decide between a +/// legal end of body and a truncated response before calling `eof`, which +/// commits (and fails the decoder) either way. The queries answer from state. +#[test] +fn http1_mid_message_is_queryable_without_feeding_eof() { + let position = |d: &Decoder| (d.is_mid_message(), d.eof_is_clean()); + // Content-Length: mid-body, and EOF here would be a truncation. + let mut decoder = Decoder::new(Mode::Response, Limits::default()); + decoder.response_to("GET"); + assert_eq!(position(&decoder), (false, false), "no response yet"); + let head = b"HTTP/1.1 200 OK\r\ncontent-length: 6\r\n\r\n"; + decoder.receive(head).unwrap(); + assert_eq!(position(&decoder), (true, false)); + decoder.receive(b"par").unwrap(); + assert_eq!(position(&decoder), (true, false)); + // Asking changed nothing: the rest of the body still decodes. + assert!(matches!( + decoder.receive(b"tial").unwrap().event, + Some(Event::Body(b"tia")) + )); + assert_eq!(position(&decoder), (false, true), "complete, End owed"); + decoder.receive(&[]).unwrap(); + assert_eq!(position(&decoder), (false, true)); + decoder.eof().unwrap(); + + // Close-delimited: mid-body, and EOF is its legal end. + let mut decoder = Decoder::new(Mode::Response, Limits::default()); + decoder.response_to("GET"); + decoder.receive(b"HTTP/1.0 200 OK\r\n\r\n").unwrap(); + decoder.receive(b"abc").unwrap(); + assert_eq!(position(&decoder), (true, true)); + decoder.eof().unwrap(); + assert_eq!(position(&decoder), (false, true)); + + // Chunked framing and trailers are mid-message too. + let mut decoder = Decoder::new(Mode::Response, Limits::default()); + decoder.response_to("GET"); + let mut wire = + b"HTTP/1.1 200 OK\r\ntransfer-encoding: chunked\r\n\r\n2\r\nab\r\n0\r\n".to_vec(); + let mut seen = Vec::new(); + loop { + let step = decoder.receive(&wire).unwrap(); + let (consumed, progressed) = (step.consumed, step.consumed > 0 || step.event.is_some()); + wire.drain(..consumed); + seen.push(position(&decoder)); + if !progressed { + break; + } + } + assert!(seen.iter().all(|p| *p == (true, false)), "{seen:?}"); + + // An informational response is not the message. + let mut decoder = Decoder::new(Mode::Response, Limits::default()); + decoder.response_to("GET"); + decoder.receive(b"HTTP/1.1 100 Continue\r\n\r\n").unwrap(); + assert_eq!(position(&decoder), (false, false)); + + // The answers agree with what `eof` then does. + let mut decoder = Decoder::new(Mode::Response, Limits::default()); + decoder.response_to("GET"); + decoder.receive(head).unwrap(); + assert!(!decoder.eof_is_clean()); + assert!(decoder.eof().is_err()); + assert_eq!(position(&decoder), (false, false), "failed"); +} + /// PerryTS/turnloop#47, part 1: Node's `res.writeHead(404, "Nope")`. #[test] fn http1_encoder_writes_a_custom_reason_phrase() { From 39601d7a5c4dc6ff7220d4ce03f7965dd4c8e05a Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Ralph=20K=C3=BCpper?= Date: Tue, 22 Sep 2026 19:40:22 +0200 Subject: [PATCH 5/8] Say whether a StreamingDecoder step needs input or output space `StreamingDecoder::process` returned after a single engine call, which can stop anywhere: after a gzip header, at a member boundary, when the output filled, or when the input ran out. Every case looked the same - a step that consumed and wrote something - so a host could only loop until a step did neither. Perry's P6 client does exactly that, and so has no signal for "give me a bigger buffer" versus "feed me more bytes". `process` now runs engine steps until one of three reasons holds, and `DecodeStep` gains `needs_input` to say which: - `finished`: the body is complete. - `needs_input`: every input byte the decoder can use is used; read more, or pass `end`. - neither: `output` is full (`written == output.len()`); drain and call again. The issue asks for a flag rather than an enum, and a flag is enough: with `finished` it separates all three cases, and it keeps `DecodeStep` a plain struct of counts. Running to a stop reason inside `process` is what makes the flag true to its name - a single engine call can stop with input left and space free, which is neither - and it means one call per transport read and one per full buffer, with no no-progress call at the end. Existing loops keep working: their extra call now returns `0, 0, needs_input`. `decode` uses the flag for its truncation check instead of the both-zero test. Breaking: `DecodeStep` has a new public field, so a struct literal of it outside the crate needs `needs_input`. Test: `streaming_decoder_says_whether_it_needs_input_or_output_space`, for gzip, deflate, brotli and zstd (both zstd backends: default and `--all-features`). With all input and a 4-byte buffer, every step but the last writes exactly 4 bytes with `needs_input` false, in at most len/4 + 1 calls; fed one byte at a time into a 1000-byte buffer, every step but the last reports `needs_input`, and the last `finished`. On the old code the first half fails: the gzip header step returns with nothing written and space free. Closes #54 --- protocols/turnloop-http/src/compression.rs | 53 +++++++++- protocols/turnloop-http/tests/codecs.rs | 112 +++++++++++++++++---- 2 files changed, 143 insertions(+), 22 deletions(-) diff --git a/protocols/turnloop-http/src/compression.rs b/protocols/turnloop-http/src/compression.rs index 59c0e70..45729b7 100644 --- a/protocols/turnloop-http/src/compression.rs +++ b/protocols/turnloop-http/src/compression.rs @@ -15,7 +15,8 @@ pub fn decode(encoding: &str, input: &[u8], output: &mut Vec, limit: usize) if step.finished { return Ok(()); } - if step.consumed == 0 && step.written == 0 { + // All input was given with `end`, so wanting more means truncation. + if step.needs_input { return Err(corrupt()); } } @@ -23,11 +24,26 @@ pub fn decode(encoding: &str, input: &[u8], output: &mut Vec, limit: usize) /// A single incremental decode step. Retain input after `consumed`, and consume /// output before calling again. No internal input/output staging allocation. +/// +/// A step returns for exactly one of three reasons, and says which: +/// +/// | `finished` | `needs_input` | stopped because | next | +/// |---|---|---|---| +/// | `true` | `false` | the body is complete | nothing; `reset` for another body | +/// | `false` | `true` | every input byte the decoder can use is used | more input, or `end = true` | +/// | `false` | `false` | `output` is full (`written == output.len()`) | drain, then call again | +/// +/// So a host sizes its scratch buffer from the third case - it filled the +/// buffer - and reads from the transport only in the second. The unconsumed +/// tail of the input in the second case is a partial header or trailer the +/// decoder cannot use yet; keep it and append to it. #[derive(Debug)] pub struct DecodeStep { pub consumed: usize, pub written: usize, pub finished: bool, + /// The decoder stopped for input, not for output space. See the table. + pub needs_input: bool, } #[derive(Clone, Copy, PartialEq, Eq)] enum Stage { @@ -191,6 +207,12 @@ impl StreamingDecoder { self.failed = false; Ok(()) } + /// Decode as far as `input` and `output` allow: the call returns only when + /// the body is finished, the decoder needs more input, or `output` is full, + /// and [`DecodeStep`] says which. One call per transport read and one per + /// full output buffer is enough; there is no need to loop until a step + /// makes no progress. + /// /// `end` means no more encoded bytes will arrive, not that output is unbounded. pub fn process(&mut self, input: &[u8], output: &mut [u8], end: bool) -> Result { if self.failed { @@ -203,6 +225,28 @@ impl StreamingDecoder { result } fn process_inner(&mut self, input: &[u8], output: &mut [u8], end: bool) -> Result { + // Run single engine steps until one of the three stop reasons holds, so + // the step that comes back names which one it was. + let mut consumed = 0; + let mut written = 0; + loop { + let step = self.step_once(&input[consumed..], &mut output[written..], end)?; + consumed += step.consumed; + written += step.written; + let stalled = step.consumed == 0 && step.written == 0; + if step.finished || written == output.len() || stalled { + return Ok(DecodeStep { + consumed, + written, + finished: step.finished, + needs_input: !step.finished && written < output.len(), + }); + } + } + } + /// One call into the engine. It may stop short of all three reasons: at a + /// gzip member boundary, after a header, or wherever the engine returns. + fn step_once(&mut self, input: &[u8], output: &mut [u8], end: bool) -> Result { if self.done { if !input.is_empty() { return Err(corrupt()); @@ -211,6 +255,7 @@ impl StreamingDecoder { consumed: 0, written: 0, finished: true, + needs_input: false, }); } if output.is_empty() { @@ -218,6 +263,7 @@ impl StreamingDecoder { consumed: 0, written: 0, finished: false, + needs_input: false, }); } let mut consumed = 0; @@ -278,6 +324,7 @@ impl StreamingDecoder { consumed: 0, written: 0, finished: false, + needs_input: false, }); } let zlib = input[0] & 15 == 8 @@ -338,6 +385,7 @@ impl StreamingDecoder { consumed: 0, written: 0, finished: true, + needs_input: false, }); } let result = decoder @@ -364,6 +412,7 @@ impl StreamingDecoder { consumed: 0, written: 0, finished: true, + needs_input: false, }); } let mut source = input; @@ -373,6 +422,7 @@ impl StreamingDecoder { consumed: 0, written: 0, finished: false, + needs_input: false, }); } decoder.reset(&mut source).map_err(|_| corrupt())?; @@ -411,6 +461,7 @@ impl StreamingDecoder { consumed, written, finished, + needs_input: false, }) } } diff --git a/protocols/turnloop-http/tests/codecs.rs b/protocols/turnloop-http/tests/codecs.rs index 2c8df95..2251df5 100644 --- a/protocols/turnloop-http/tests/codecs.rs +++ b/protocols/turnloop-http/tests/codecs.rs @@ -527,29 +527,43 @@ fn expect_continue_streaming_abort_and_reuse() { assert!(!connection.reusable()); } -#[test] -fn streaming_compression_fragmented_bounded_and_truncated() { +const STREAMED: &[u8] = b"streamed body streamed body streamed body"; +/// Encode `body` with one content coding. zstd has no encoder on every target +/// the suite runs on, so it has a fixture for `STREAMED` only. +fn encode_with(coding: &str, body: &[u8]) -> Vec { use std::io::Write; - let body = b"streamed body streamed body streamed body"; - let mut gzip = flate2::write::GzEncoder::new(Vec::new(), Default::default()); - gzip.write_all(body).unwrap(); - let mut deflate = flate2::write::DeflateEncoder::new(Vec::new(), Default::default()); - deflate.write_all(body).unwrap(); - let mut br = Vec::new(); - { - let mut writer = brotli::CompressorWriter::new(&mut br, 4096, 4, 22); - writer.write_all(body).unwrap(); + match coding { + "gzip" => { + let mut gzip = flate2::write::GzEncoder::new(Vec::new(), Default::default()); + gzip.write_all(body).unwrap(); + gzip.finish().unwrap() + } + "deflate" => { + let mut deflate = flate2::write::DeflateEncoder::new(Vec::new(), Default::default()); + deflate.write_all(body).unwrap(); + deflate.finish().unwrap() + } + "br" => { + let mut br = Vec::new(); + { + let mut writer = brotli::CompressorWriter::new(&mut br, 4096, 4, 22); + writer.write_all(body).unwrap(); + } + br + } + "zstd" => { + assert_eq!(body, STREAMED, "zstd has a fixture for STREAMED only"); + // Generated by Node 26 zlib.zstdCompressSync, also run on WASI. + hex("28b52ffd2029a500007073747265616d656420626f6479200100114e25") + } + _ => panic!("{coding}"), } - // zstd fixture generated by Node 26 zlib.zstdCompressSync, also run on WASI. - let cases = vec![ - ("gzip", gzip.finish().unwrap()), - ("deflate", deflate.finish().unwrap()), - ("br", br), - ( - "zstd", - hex("28b52ffd2029a500007073747265616d656420626f6479200100114e25"), - ), - ]; +} +const CODINGS: [&str; 4] = ["gzip", "deflate", "br", "zstd"]; +#[test] +fn streaming_compression_fragmented_bounded_and_truncated() { + let body = STREAMED; + let cases = CODINGS.map(|coding| (coding, encode_with(coding, body))); for (encoding, wire) in cases { let mut decoder = turnloop_http::compression::StreamingDecoder::new(encoding, 100).unwrap(); let mut input = Vec::new(); @@ -1671,3 +1685,59 @@ fn http1_encoder_writes_a_body_forbidden_response() { } assert!(wire.is_empty()); } + +/// PerryTS/turnloop#54: a step says whether it stopped for input or for output +/// space, so a host reads only when asked to and can size its buffer. +#[test] +fn streaming_decoder_says_whether_it_needs_input_or_output_space() { + use turnloop_http::compression::StreamingDecoder; + for coding in CODINGS { + let wire = encode_with(coding, STREAMED); + + // All input present, a 4-byte output: only output space can stop it, + // and every call but the last fills the buffer. + let mut decoder = StreamingDecoder::new(coding, 1000).unwrap(); + let (mut pos, mut result, mut calls) = (0, Vec::new(), 0); + loop { + let mut out = [0; 4]; + let step = decoder.process(&wire[pos..], &mut out, true).unwrap(); + calls += 1; + pos += step.consumed; + result.extend_from_slice(&out[..step.written]); + if step.finished { + assert!(!step.needs_input, "{coding}"); + break; + } + assert!(!step.needs_input, "{coding}: all input was present"); + assert_eq!(step.written, 4, "{coding}: stopped only when full"); + } + assert_eq!(result, STREAMED, "{coding}"); + assert_eq!(pos, wire.len(), "{coding}"); + assert!(calls <= STREAMED.len() / 4 + 1, "{coding}: {calls} calls"); + + // One byte at a time into a roomy buffer: only input can stop it, and + // it asks for more exactly when it has used what it has. + let mut decoder = StreamingDecoder::new(coding, 1000).unwrap(); + let (mut input, mut result) = (Vec::new(), Vec::new()); + for (i, byte) in wire.iter().enumerate() { + input.push(*byte); + let mut out = [0; 1000]; + let step = decoder + .process(&input, &mut out, i + 1 == wire.len()) + .unwrap(); + input.drain(..step.consumed); + result.extend_from_slice(&out[..step.written]); + if i + 1 == wire.len() { + assert!(step.finished && !step.needs_input, "{coding}"); + } else { + assert!(step.needs_input && !step.finished, "{coding} at {i}"); + } + } + assert_eq!(result, STREAMED, "{coding}"); + assert!(input.is_empty(), "{coding}"); + } + // An empty output buffer is output-bound, never a request for input. + let mut decoder = StreamingDecoder::new("gzip", 1000).unwrap(); + let step = decoder.process(b"\x1f", &mut [], false).unwrap(); + assert!(!step.needs_input && !step.finished); +} From 9e556c7bc2d705191295b2c4e3614acefc65cc0d Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Ralph=20K=C3=BCpper?= Date: Tue, 22 Sep 2026 19:45:00 +0200 Subject: [PATCH 6/8] Keep a zstd frame boundary across an empty decode step The native zstd engine set `boundary` from every `run_on_buffers` result. A call that moves nothing - empty input after the last byte of a frame - reports the header size of the *next* frame as `remaining`, so it cleared a boundary that had just been reached. The following `end` then looked like a truncated frame and failed a complete body. A host hits this whenever it learns the body is over after feeding the last byte, which is the normal HTTP/1 shape: the last body bytes arrive in one step and `End` in the next, zero-byte one (#50). The previous commit's run-to-a-stop-reason loop makes the empty call itself, so it made the bug reachable from a single `process` per read; a host that looped until no progress could already reach it. Only a call that consumed or wrote something now moves the boundary. Test: `streaming_decoder_finishes_when_end_follows_the_last_byte` feeds each coding byte by byte without `end`, makes an idle call, then declares `end` with no input. Native zstd failed it with UND_ERR_SOCKET before this change; the pure-Rust zstd path, gzip, deflate and brotli pass either way. Refs #54 --- protocols/turnloop-http/src/compression.rs | 12 +++++--- protocols/turnloop-http/tests/codecs.rs | 32 ++++++++++++++++++++++ 2 files changed, 40 insertions(+), 4 deletions(-) diff --git a/protocols/turnloop-http/src/compression.rs b/protocols/turnloop-http/src/compression.rs index 45729b7..3b38664 100644 --- a/protocols/turnloop-http/src/compression.rs +++ b/protocols/turnloop-http/src/compression.rs @@ -393,10 +393,14 @@ impl StreamingDecoder { .map_err(|_| corrupt())?; consumed = result.bytes_read; written = result.bytes_written; - *boundary = result.remaining == 0; - finished = result.remaining == 0 && end && consumed == input.len(); - if result.remaining != 0 && end && consumed == input.len() && written < output.len() - { + // A call that moves nothing reports the size of the *next* + // frame's header, which says nothing about the frame that just + // ended: only progress may move the boundary. + if consumed > 0 || written > 0 { + *boundary = result.remaining == 0; + } + finished = *boundary && end && consumed == input.len(); + if !*boundary && end && consumed == input.len() && written < output.len() { return Err(corrupt()); } } diff --git a/protocols/turnloop-http/tests/codecs.rs b/protocols/turnloop-http/tests/codecs.rs index 2251df5..bacc9b0 100644 --- a/protocols/turnloop-http/tests/codecs.rs +++ b/protocols/turnloop-http/tests/codecs.rs @@ -1741,3 +1741,35 @@ fn streaming_decoder_says_whether_it_needs_input_or_output_space() { let step = decoder.process(b"\x1f", &mut [], false).unwrap(); assert!(!step.needs_input && !step.finished); } + +/// A host learns that the body is over after it has fed the last byte - an +/// HTTP/1 `End` arrives in a later step - so `end` often comes with no input. +/// Native zstd used to lose its frame boundary on the empty step between. +#[test] +fn streaming_decoder_finishes_when_end_follows_the_last_byte() { + use turnloop_http::compression::StreamingDecoder; + for coding in CODINGS { + let wire = encode_with(coding, STREAMED); + let mut decoder = StreamingDecoder::new(coding, 1000).unwrap(); + let mut out = [0; 1000]; + let (mut pending, mut result) = (Vec::new(), Vec::new()); + for byte in &wire { + pending.push(*byte); + let step = decoder.process(&pending, &mut out, false).unwrap(); + pending.drain(..step.consumed); + result.extend_from_slice(&out[..step.written]); + } + assert!(pending.is_empty(), "{coding}"); + // An idle step with nothing new, then the end. (Brotli marks its own + // end, so it may already be finished; the others wait for `end`.) + let step = decoder.process(&[], &mut out, false).unwrap(); + assert_eq!(step.written, 0, "{coding}"); + assert!( + step.needs_input || coding == "br" && step.finished, + "{coding}" + ); + let step = decoder.process(&[], &mut out, true).unwrap(); + assert!(step.finished, "{coding}"); + assert_eq!(result, STREAMED, "{coding}"); + } +} From 1b6c69facc30ccd1214754fd1a8e6aff0ff90fbd Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Ralph=20K=C3=BCpper?= Date: Tue, 22 Sep 2026 19:45:34 +0200 Subject: [PATCH 7/8] Decode a Content-Encoding list, innermost last `StreamingDecoder::new` matched its argument as one exact token, so a legal `gzip, gzip` - or two `Content-Encoding` header lines - failed the whole response with UND_ERR_NOT_SUPPORTED. Perry's P11 client is the first consumer to send its own `Accept-Encoding`, and an independent review of it found the gap; it now splits and chains on the caller side. The issue offers an iterator of tokens or split-and-chain internally. This does both, because they are the same thing at two call shapes: - `StreamingDecoder::new(value, limit)` takes one header value, list included. - `StreamingDecoder::from_codings(values, limit)` takes every header line's value in order and merges them as RFC 9110 5.3 merges any list-valued field. `Head::values(name)` yields those lines (`get` returns only the first). Codings are decoded in reverse of the order listed: the last-listed is outermost. Names are case-insensitive (RFC 9110 8.4.1), `x-gzip` is `gzip`, and `identity` and empty list elements are skipped. An unknown coding still fails with UND_ERR_NOT_SUPPORTED, and so does a chain longer than `MAX_CODINGS` (5): the list is attacker-chosen and each coding costs state and a buffer; undici stops at five for the same reason. A single coding is unchanged: it decodes straight into the caller's output, and the chain vector stays empty and unallocated. Each further coding decodes into a fixed 16 KiB staging buffer - over the 8 KiB a gzip header may take, so the next coding always sees a whole header - allocated at construction and kept across `reset`, so steady-state reuse stays allocation-free. A chained step keeps the #54 contract: it returns only when finished, when it needs input, or when the output is full. `limit` bounds every coding's output, not only the last: an outer coding could otherwise expand without bound into an inner one that produces nothing (empty gzip members, for instance), costing CPU with no decoded bytes to trip the limit. Internally the per-coding state moves into a `Codec`, and `StreamingDecoder` owns the innermost one plus the staged outer ones. Tests: `streaming_decoder_applies_a_content_encoding_list_innermost_last` decodes a 80 KB `gzip, br` body whole, byte-by-byte into 7-byte output, and in other input/output splits, before and after `reset`; the same from two header lines through `Head::values`; the reversed order fails; `gzip, gzip` under case, `x-gzip`, identity and empty elements, through the convenience `decode`; every coding under gzip, deflate and brotli (zstd included) in four input/output splits; truncation; the decoded-size limit; unknown codings and a six-coding chain refused, five accepted. `streaming_decompressors_reuse_scratch` gains a `gzip, br` case and still measures zero allocations across 100 reused bodies. All pass with the native and the pure-Rust zstd backends. Closes #79 --- protocols/turnloop-http/README.md | 2 +- protocols/turnloop-http/src/compression.rs | 275 ++++++++++++++++--- protocols/turnloop-http/src/http1.rs | 10 + protocols/turnloop-http/tests/allocations.rs | 20 +- protocols/turnloop-http/tests/codecs.rs | 110 ++++++++ 5 files changed, 370 insertions(+), 47 deletions(-) diff --git a/protocols/turnloop-http/README.md b/protocols/turnloop-http/README.md index ac5f54a..ef35ee3 100644 --- a/protocols/turnloop-http/README.md +++ b/protocols/turnloop-http/README.md @@ -7,7 +7,7 @@ Own HTTP/1.1 and HTTP/2 wire engines, with no runtime or transport dependency. - `http2::Connection` handles both roles. Flush and acknowledge `output`, retain incomplete input, consume one frame/event at a time. `send_data` can accept only a prefix or zero when flow-controlled; retain the remainder. Return DATA capacity with `release_capacity` after the application consumes it. Queue application responses in the host during stalls. On transport loss call `eof`, then drain `poll_failed_stream`. - `hpack` provides the independent bounded RFC 7541 codec. The encoder never indexes credentials/cookies. Decode failures poison a context. - `client::Pool` reserves connections before DNS/connect so parallel commands respect per-origin/proxy limits. `Route` emits resolution/TLS requests and builds proxy CONNECT or absolute-form heads. `Resolver` belongs to the host. `Request::redirect` applies Fetch redirects; use `DEFAULT_MAX_REDIRECTS` (20). JS conversions and promise delivery remain with Perry. -- `compression::StreamingDecoder` accepts input and caller-owned output. Reuse it with `reset` to retain scratch buffers across bodies. Hosts may cache one per content encoding. `decode` is the convenience whole-body path and constructs algorithm state each time. +- `compression::StreamingDecoder` accepts input and caller-owned output. Each step says whether it stopped for input (`needs_input`), for output space, or because the body finished. `new` takes a `Content-Encoding` value, list included (`gzip, br`, decoded innermost-last); `from_codings(head.values("content-encoding"), limit)` merges repeated header lines. Reuse it with `reset` to retain scratch buffers across bodies. Hosts may cache one per content encoding. `decode` is the convenience whole-body path and constructs algorithm state each time. No engine samples a clock. The host passes `Instant` deadlines and invokes timeout handlers. Hold output storage stable until a completion-shaped write finishes: do not mutate the engine while an I/O operation borrows its output. Error codes are transport causes; Perry creates the JS error objects and detailed OS diagnostics. diff --git a/protocols/turnloop-http/src/compression.rs b/protocols/turnloop-http/src/compression.rs index 3b38664..9aacb80 100644 --- a/protocols/turnloop-http/src/compression.rs +++ b/protocols/turnloop-http/src/compression.rs @@ -23,7 +23,8 @@ pub fn decode(encoding: &str, input: &[u8], output: &mut Vec, limit: usize) } /// A single incremental decode step. Retain input after `consumed`, and consume -/// output before calling again. No internal input/output staging allocation. +/// output before calling again. A single coding stages nothing internally; a +/// chain stages between its codings in buffers allocated at construction. /// /// A step returns for exactly one of three reasons, and says which: /// @@ -86,19 +87,51 @@ enum Engine { boundary: bool, }, } +/// Content codings one decoder accepts in a chain. A list is attacker-chosen and +/// every coding adds state and a staging buffer; undici stops at five as well. +pub const MAX_CODINGS: usize = 5; +/// Bytes staged between two codings of a chain, above the 8 KiB a gzip header +/// may take, so the inner coding can always see a whole header. +const STAGE_BYTES: usize = 16 * 1024; +/// Incremental decoder for a `Content-Encoding` value: one coding, or a list +/// of them applied in order (see [`StreamingDecoder::from_codings`]). pub struct StreamingDecoder { + /// The innermost coding - listed first, decoded last - writing the + /// caller's output. The only coding when the list has one. + codec: Codec, + /// The other codings, outermost (listed last) first, each with the bytes it + /// has produced and the next coding has not yet taken. Empty, and so never + /// allocated, for a single coding. + outer: Vec, + limit: usize, + failed: bool, +} +/// One coding's decoder state. +struct Codec { engine: Engine, total: usize, - limit: usize, done: bool, - failed: bool, +} +/// An outer coding of a chain and the output it has staged for the next one. +struct Staged { + codec: Codec, + buffer: Box<[u8]>, + start: usize, + end: usize, +} +fn unsupported() -> Error { + Error::new("UND_ERR_NOT_SUPPORTED", "unsupported content encoding") } fn corrupt() -> Error { Error::new("UND_ERR_SOCKET", "invalid or incomplete compressed body") } -impl StreamingDecoder { - pub fn new(encoding: &str, limit: usize) -> Result { - let engine = match encoding { +impl Codec { + fn new(token: &[u8]) -> Result { + // Content codings are case-insensitive (RFC 9110 section 8.4.1). + let name = std::str::from_utf8(token) + .map_err(|_| unsupported())? + .to_ascii_lowercase(); + let engine = match name.as_str() { "" | "identity" => Engine::Identity, "gzip" | "x-gzip" => Engine::Deflate { decoder: flate2::Decompress::new(false), @@ -145,24 +178,15 @@ impl StreamingDecoder { reset: true, boundary: false, }, - _ => { - return Err(Error::new( - "UND_ERR_NOT_SUPPORTED", - "unsupported content encoding", - )); - } + _ => return Err(unsupported()), }; Ok(Self { engine, total: 0, - limit, done: false, - failed: false, }) } - /// Reuse algorithm and scratch storage for another body of the same encoding. - /// Warm up with the largest expected body shape before measuring allocations. - pub fn reset(&mut self, limit: usize) -> Result<()> { + fn reset(&mut self) -> Result<()> { match &mut self.engine { Engine::Identity => {} Engine::Deflate { @@ -202,35 +226,22 @@ impl StreamingDecoder { } } self.total = 0; - self.limit = limit; self.done = false; - self.failed = false; Ok(()) } - /// Decode as far as `input` and `output` allow: the call returns only when - /// the body is finished, the decoder needs more input, or `output` is full, - /// and [`DecodeStep`] says which. One call per transport read and one per - /// full output buffer is enough; there is no need to loop until a step - /// makes no progress. - /// - /// `end` means no more encoded bytes will arrive, not that output is unbounded. - pub fn process(&mut self, input: &[u8], output: &mut [u8], end: bool) -> Result { - if self.failed { - return Err(corrupt()); - } - let result = self.process_inner(input, output, end); - if result.is_err() { - self.failed = true; - } - result - } - fn process_inner(&mut self, input: &[u8], output: &mut [u8], end: bool) -> Result { + fn run( + &mut self, + input: &[u8], + output: &mut [u8], + end: bool, + limit: usize, + ) -> Result { // Run single engine steps until one of the three stop reasons holds, so // the step that comes back names which one it was. let mut consumed = 0; let mut written = 0; loop { - let step = self.step_once(&input[consumed..], &mut output[written..], end)?; + let step = self.step_once(&input[consumed..], &mut output[written..], end, limit)?; consumed += step.consumed; written += step.written; let stalled = step.consumed == 0 && step.written == 0; @@ -246,7 +257,13 @@ impl StreamingDecoder { } /// One call into the engine. It may stop short of all three reasons: at a /// gzip member boundary, after a header, or wherever the engine returns. - fn step_once(&mut self, input: &[u8], output: &mut [u8], end: bool) -> Result { + fn step_once( + &mut self, + input: &[u8], + output: &mut [u8], + end: bool, + limit: usize, + ) -> Result { if self.done { if !input.is_empty() { return Err(corrupt()); @@ -453,7 +470,7 @@ impl StreamingDecoder { } } } - if written > self.limit.saturating_sub(self.total) { + if written > limit.saturating_sub(self.total) { return Err(Error::new( "UND_ERR_RES_EXCEEDED_MAX_SIZE", "decoded body exceeds limit", @@ -469,6 +486,184 @@ impl StreamingDecoder { }) } } +impl Staged { + /// Move the unread bytes to the front, so the free space is contiguous. + fn compact(&mut self) { + if self.start > 0 { + self.buffer.copy_within(self.start..self.end, 0); + self.end -= self.start; + self.start = 0; + } + } +} +impl StreamingDecoder { + /// A decoder for one `Content-Encoding` header value, which may be a list + /// (`"gzip, br"`). The same as [`from_codings`](Self::from_codings) with + /// that one value. + pub fn new(encoding: &str, limit: usize) -> Result { + Self::from_codings([encoding], limit) + } + /// A decoder for a `Content-Encoding` field: pass every header line's + /// value, in order. Repeated lines are merged as for any list-valued field + /// (RFC 9110 section 5.3), and each value may itself be a comma-separated + /// list. + /// + /// Codings are listed in the order they were applied, so they are decoded + /// in reverse: in `gzip, br` the last-listed `br` is outermost and decoded + /// first. Names are case-insensitive, `x-gzip` is `gzip`, and `identity` + /// and empty list elements are skipped; no coding at all decodes as + /// identity. An unknown coding, or more than [`MAX_CODINGS`], fails with + /// `UND_ERR_NOT_SUPPORTED`. + /// + /// One coding decodes straight into the caller's output as before. Each + /// further coding stages its output in a fixed 16 KiB buffer allocated + /// here and kept across [`reset`](Self::reset); `limit` bounds every + /// coding's output, not only the last, so an outer coding cannot expand + /// without bound into an inner one that produces nothing. + pub fn from_codings(values: I, limit: usize) -> Result + where + I: IntoIterator, + I::Item: AsRef<[u8]>, + { + let mut codecs = Vec::new(); + for value in values { + for token in value.as_ref().split(|b| *b == b',') { + let token = token.trim_ascii(); + if token.is_empty() || token.eq_ignore_ascii_case(b"identity") { + continue; + } + if codecs.len() == MAX_CODINGS { + return Err(Error::new( + "UND_ERR_NOT_SUPPORTED", + "too many content codings", + )); + } + codecs.push(Codec::new(token)?); + } + } + let mut codecs = codecs.into_iter(); + let codec = match codecs.next() { + Some(codec) => codec, + None => Codec::new(b"identity")?, + }; + let outer = codecs + .rev() + .map(|codec| Staged { + codec, + buffer: vec![0; STAGE_BYTES].into_boxed_slice(), + start: 0, + end: 0, + }) + .collect(); + Ok(Self { + codec, + outer, + limit, + failed: false, + }) + } + /// Reuse algorithm and scratch storage for another body of the same encoding. + /// Warm up with the largest expected body shape before measuring allocations. + pub fn reset(&mut self, limit: usize) -> Result<()> { + self.codec.reset()?; + for stage in &mut self.outer { + stage.codec.reset()?; + stage.start = 0; + stage.end = 0; + } + self.limit = limit; + self.failed = false; + Ok(()) + } + /// Decode as far as `input` and `output` allow: the call returns only when + /// the body is finished, the decoder needs more input, or `output` is full, + /// and [`DecodeStep`] says which. One call per transport read and one per + /// full output buffer is enough; there is no need to loop until a step + /// makes no progress. + /// + /// `end` means no more encoded bytes will arrive, not that output is unbounded. + pub fn process(&mut self, input: &[u8], output: &mut [u8], end: bool) -> Result { + if self.failed { + return Err(corrupt()); + } + let result = self.process_inner(input, output, end); + if result.is_err() { + self.failed = true; + } + result + } + fn process_inner(&mut self, input: &[u8], output: &mut [u8], end: bool) -> Result { + let limit = self.limit; + let Some((last, earlier)) = self.outer.split_last_mut() else { + return self.codec.run(input, output, end, limit); + }; + if self.codec.done { + // Finished: the same answer, or error, a single coding gives. + return self.codec.run(input, output, end, limit); + } + let mut consumed = 0; + let mut written = 0; + loop { + let mut progress = false; + // Outermost first: each stage decodes what the one before it staged + // (the first, the caller's input) into its own buffer. A stage's + // input is complete once the stage feeding it is done. + for i in 0..=earlier.len() { + let (before, rest) = earlier.split_at_mut(i); + let stage = match rest.first_mut() { + Some(stage) => stage, + None => &mut *last, + }; + stage.compact(); + if stage.end == stage.buffer.len() { + continue; + } + let step = match before.last_mut() { + None => { + let step = stage.codec.run( + &input[consumed..], + &mut stage.buffer[stage.end..], + end, + limit, + )?; + consumed += step.consumed; + step + } + Some(feed) => { + let step = stage.codec.run( + &feed.buffer[feed.start..feed.end], + &mut stage.buffer[stage.end..], + feed.codec.done, + limit, + )?; + feed.start += step.consumed; + step + } + }; + stage.end += step.written; + progress |= step.consumed > 0 || step.written > 0; + } + // The innermost coding decodes the last stage into the caller's output. + let step = self.codec.run( + &last.buffer[last.start..last.end], + &mut output[written..], + last.codec.done, + limit, + )?; + last.start += step.consumed; + written += step.written; + progress |= step.consumed > 0 || step.written > 0; + if step.finished || written == output.len() || !progress { + return Ok(DecodeStep { + consumed, + written, + finished: step.finished, + needs_input: !step.finished && written < output.len(), + }); + } + } + } +} fn gzip_header(input: &[u8]) -> Result> { if input.len() < 10 { return Ok(None); diff --git a/protocols/turnloop-http/src/http1.rs b/protocols/turnloop-http/src/http1.rs index 5156401..181ac7b 100644 --- a/protocols/turnloop-http/src/http1.rs +++ b/protocols/turnloop-http/src/http1.rs @@ -47,6 +47,16 @@ impl Head { .find(|h| h.name.eq_ignore_ascii_case(name)) .map(|h| h.value.as_slice()) } + /// Every value of a header that may repeat, in wire order. [`get`](Self::get) + /// returns only the first line; a list-valued field such as + /// `Content-Encoding` needs all of them, for example + /// `StreamingDecoder::from_codings(head.values("content-encoding"), limit)`. + pub fn values<'a>(&'a self, name: &'a str) -> impl Iterator + 'a { + self.headers + .iter() + .filter(move |h| h.name.eq_ignore_ascii_case(name)) + .map(|h| h.value.as_slice()) + } pub fn token(&self, name: &str, token: &str) -> bool { self.headers .iter() diff --git a/protocols/turnloop-http/tests/allocations.rs b/protocols/turnloop-http/tests/allocations.rs index 540bfad..c3da2e1 100644 --- a/protocols/turnloop-http/tests/allocations.rs +++ b/protocols/turnloop-http/tests/allocations.rs @@ -149,15 +149,23 @@ fn streaming_decompressors_reuse_scratch() { gzip.write_all(body).unwrap(); let mut deflate = flate2::write::ZlibEncoder::new(Vec::new(), Default::default()); deflate.write_all(body).unwrap(); - let mut brotli = Vec::new(); - { - let mut writer = brotli::CompressorWriter::new(&mut brotli, 4096, 4, 22); - writer.write_all(body).unwrap(); - } + let brotli_of = |bytes: &[u8]| { + let mut out = Vec::new(); + { + let mut writer = brotli::CompressorWriter::new(&mut out, 4096, 4, 22); + writer.write_all(bytes).unwrap(); + } + out + }; + let brotli = brotli_of(body); + let gzip = gzip.finish().unwrap(); + // A chain stages between its codings in buffers kept across `reset`. + let chain = brotli_of(&gzip); let cases = [ - ("gzip", gzip.finish().unwrap()), + ("gzip", gzip), ("deflate", deflate.finish().unwrap()), ("br", brotli), + ("gzip, br", chain), ( "zstd", "28b52ffd2038cd000098616c6c6f636174696f6e2070726f66696c65200100d933c3" diff --git a/protocols/turnloop-http/tests/codecs.rs b/protocols/turnloop-http/tests/codecs.rs index bacc9b0..583cc78 100644 --- a/protocols/turnloop-http/tests/codecs.rs +++ b/protocols/turnloop-http/tests/codecs.rs @@ -1773,3 +1773,113 @@ fn streaming_decoder_finishes_when_end_follows_the_last_byte() { assert_eq!(result, STREAMED, "{coding}"); } } + +/// Decode `wire` with `decoder`, `input` bytes per read and `output` bytes of +/// scratch, honouring `needs_input` exactly: read only when it asks. +fn decode_streamed( + decoder: &mut turnloop_http::compression::StreamingDecoder, + wire: &[u8], + input: usize, + output: usize, +) -> Result, turnloop_http::Error> { + let (mut fed, mut pending, mut result) = (0, Vec::new(), Vec::new()); + let mut out = vec![0; output]; + loop { + let step = decoder.process(&pending, &mut out, fed == wire.len())?; + pending.drain(..step.consumed); + result.extend_from_slice(&out[..step.written]); + if step.finished { + return Ok(result); + } + if step.needs_input { + assert!(fed < wire.len(), "asked for input after `end`"); + let next = (fed + input).min(wire.len()); + pending.extend_from_slice(&wire[fed..next]); + fed = next; + } + } +} + +/// PerryTS/turnloop#79: a `Content-Encoding` list, and repeated header lines, +/// decode as one chain, innermost (first-listed) last. +#[test] +fn streaming_decoder_applies_a_content_encoding_list_innermost_last() { + use turnloop_http::compression::{MAX_CODINGS, StreamingDecoder, decode}; + let body: Vec = (0..20_000u32) + .flat_map(|i| (i % 97).to_le_bytes()) + .collect(); + // `gzip, br`: gzip was applied first, so br is outermost. + let wire = encode_with("br", &encode_with("gzip", &body)); + for (input, output) in [(wire.len(), 1 << 20), (1, 7), (1000, 3), (7, 1 << 16)] { + let mut decoder = StreamingDecoder::new("gzip, br", 1 << 20).unwrap(); + let decoded = decode_streamed(&mut decoder, &wire, input, output).unwrap(); + assert!(decoded == body, "{input}/{output}"); + // Reused for a second body of the same chain. + decoder.reset(1 << 20).unwrap(); + let decoded = decode_streamed(&mut decoder, &wire, input, output).unwrap(); + assert!(decoded == body, "reset {input}/{output}"); + } + // Two header lines are the same list, in order. + let head = Head { + method: String::new(), + target: String::new(), + status: 200, + version: 1, + headers: vec![ + Header::new("content-encoding", "gzip"), + Header::new("x-other", "br"), + Header::new("Content-Encoding", "br"), + ], + keep_alive: true, + }; + let mut decoder = + StreamingDecoder::from_codings(head.values("content-encoding"), 1 << 20).unwrap(); + assert!(decode_streamed(&mut decoder, &wire, 4096, 4096).unwrap() == body); + // The other order is a different coding, and fails. + let mut decoder = StreamingDecoder::new("br, gzip", 1 << 20).unwrap(); + assert!(decode_streamed(&mut decoder, &wire, 4096, 4096).is_err()); + + // The legal `gzip, gzip` that P11 saw fail, with case, `x-gzip`, identity + // and empty elements - and the convenience `decode` takes a list too. + let twice = encode_with("gzip", &encode_with("gzip", STREAMED)); + for list in ["gzip, gzip", "GZIP,x-gzip", " gzip ,, identity, Gzip "] { + let mut out = Vec::new(); + decode(list, &twice, &mut out, 1000).unwrap(); + assert_eq!(out, STREAMED, "{list}"); + } + // Every coding under every other. + for inner in CODINGS { + for outer in ["gzip", "deflate", "br"] { + let wire = encode_with(outer, &encode_with(inner, STREAMED)); + for (input, output) in [(1, 5), (1, 1000), (1000, 5), (wire.len(), 1000)] { + let mut decoder = StreamingDecoder::from_codings([inner, outer], 1000).unwrap(); + let decoded = decode_streamed(&mut decoder, &wire, input, output); + assert!( + decoded.as_deref() == Ok(STREAMED), + "{inner}, {outer} {input}/{output}: {decoded:?}" + ); + } + } + } + + // Truncation anywhere in the chain is still an error. + let mut decoder = StreamingDecoder::new("gzip, br", 1 << 20).unwrap(); + assert!(decode_streamed(&mut decoder, &wire[..wire.len() - 1], 64, 64).is_err()); + // The limit bounds the decoded body. + let mut decoder = StreamingDecoder::new("gzip, br", body.len() - 1).unwrap(); + let error = decode_streamed(&mut decoder, &wire, 4096, 4096).unwrap_err(); + assert_eq!(error.code, "UND_ERR_RES_EXCEEDED_MAX_SIZE"); + // Unknown codings and over-long chains are refused up front. + for list in ["gzip, compress", "gzip, br\u{e9}"] { + let error = StreamingDecoder::new(list, 1000).err().map(|e| e.code); + assert_eq!(error, Some("UND_ERR_NOT_SUPPORTED"), "{list}"); + } + let chain = ["gzip"; MAX_CODINGS].join(", "); + assert!(StreamingDecoder::new(&chain, 1000).is_ok()); + let error = StreamingDecoder::new(&format!("{chain}, gzip"), 1000).err(); + assert_eq!(error.map(|e| e.code), Some("UND_ERR_NOT_SUPPORTED")); + // No coding at all is identity. + let mut out = Vec::new(); + decode(" , identity", b"plain", &mut out, 100).unwrap(); + assert_eq!(out, b"plain"); +} From 996fdcd624c0e2cf855ea2bf0c124c9205bf6a15 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Ralph=20K=C3=BCpper?= Date: Tue, 22 Sep 2026 20:03:41 +0200 Subject: [PATCH 8/8] Merge repeated Content-Encoding lines in the async client asynchronous::client::Decoders::start read only the first Content-Encoding line, so a response carrying `gzip` on two separate lines was decoded once and handed the caller still-gzipped bytes. RFC 9110 5.3 makes repeated lines one list; StreamingDecoder already takes a list, so the lines are joined into the decoder's cache key. The single-line case still borrows the header value. The new unit test decodes a body gzipped twice, once with two header lines and once with one `gzip, gzip` line, and fails without the change. Refs #79 --- .../turnloop-http/src/asynchronous/client.rs | 78 ++++++++++++++++++- 1 file changed, 76 insertions(+), 2 deletions(-) diff --git a/protocols/turnloop-http/src/asynchronous/client.rs b/protocols/turnloop-http/src/asynchronous/client.rs index 1ebe129..31ec64e 100644 --- a/protocols/turnloop-http/src/asynchronous/client.rs +++ b/protocols/turnloop-http/src/asynchronous/client.rs @@ -652,8 +652,24 @@ impl Decoders { if !options.decompress { return Ok(()); } - let encoding = std::str::from_utf8(head.get("content-encoding").unwrap_or(b"identity")) - .map_err(io::Error::other)?; + // Repeated Content-Encoding lines are one list (RFC 9110 section 5.3). + // The common single-line case borrows; only a repeated field pays for + // joining the lines into one cache key. + let joined; + let encoding = if head.values("content-encoding").nth(1).is_some() { + let mut key = Vec::new(); + for value in head.values("content-encoding") { + if !key.is_empty() { + key.extend_from_slice(b", "); + } + key.extend_from_slice(value); + } + joined = String::from_utf8(key).map_err(io::Error::other)?; + joined.as_str() + } else { + std::str::from_utf8(head.get("content-encoding").unwrap_or(b"identity")) + .map_err(io::Error::other)? + }; if encoding == "identity" { return Ok(()); } @@ -714,3 +730,61 @@ impl Decoders { Ok(()) } } +#[cfg(test)] +mod tests { + use super::*; + use std::io::Write; + + fn gzip(bytes: &[u8]) -> Vec { + let mut encoder = flate2::write::GzEncoder::new(Vec::new(), flate2::Compression::fast()); + encoder.write_all(bytes).expect("gzip into a Vec"); + encoder.finish().expect("gzip into a Vec") + } + fn head(encodings: &[&str]) -> Head { + Head { + method: String::new(), + target: String::new(), + status: 200, + version: 1, + headers: encodings + .iter() + .map(|value| Header::new("Content-Encoding", value)) + .collect(), + keep_alive: true, + } + } + fn decode(decoders: &mut Decoders, head: &Head, wire: &[u8]) -> io::Result> { + decoders.start(head, &Options::default())?; + let mut out = Vec::new(); + decoders.feed(wire, true, &mut |bytes| { + out.extend_from_slice(bytes); + Ok(()) + })?; + Ok(out) + } + + /// PerryTS/turnloop#79: repeated Content-Encoding lines are one list, so + /// two `gzip` lines decode twice rather than once. + #[test] + fn repeated_content_encoding_lines_decode_as_one_list() { + let body = b"a body compressed twice, one coding per header line".repeat(50); + let wire = gzip(&gzip(&body)); + let mut decoders = Decoders::default(); + assert_eq!( + decode(&mut decoders, &head(&["gzip", "gzip"]), &wire).unwrap(), + body + ); + // The same chain on one line shares the joined cache entry. + assert_eq!( + decode(&mut decoders, &head(&["gzip, gzip"]), &wire).unwrap(), + body + ); + assert_eq!(decoders.cache.len(), 1); + // A single line still decodes once. + let once = gzip(&body); + assert_eq!( + decode(&mut decoders, &head(&["gzip"]), &once).unwrap(), + body + ); + } +}