diff --git a/protocols/turnloop-http/README.md b/protocols/turnloop-http/README.md index dcb6bf7..6b6627f 100644 --- a/protocols/turnloop-http/README.md +++ b/protocols/turnloop-http/README.md @@ -8,7 +8,7 @@ Own HTTP/1.1 and HTTP/2 wire engines, with no runtime or transport dependency. - `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. `Pool::contains`/`forget` check or drop a stale `ConnectionId`. `Pool::next_timeout` is one O(1) deadline over idle connections and every request registered with `set_request_deadline`; drain `handle_timeout` and `handle_request_timeout`. `client::Deadlines` is the same heap for host-keyed timers. - `multipart::Form` encodes in-memory `multipart/form-data` text and file parts. The host supplies 16 bytes of entropy; the boundary is regenerated until it occurs in no part. -- `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/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 + ); + } +} 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/compression.rs b/protocols/turnloop-http/src/compression.rs index 59c0e70..9aacb80 100644 --- a/protocols/turnloop-http/src/compression.rs +++ b/protocols/turnloop-http/src/compression.rs @@ -15,19 +15,36 @@ 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()); } } } /// 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: +/// +/// | `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 { @@ -70,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), @@ -129,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 { @@ -186,23 +226,44 @@ impl StreamingDecoder { } } self.total = 0; - self.limit = limit; self.done = false; - self.failed = false; Ok(()) } - /// `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; + 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, limit)?; + 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(), + }); + } } - result } - fn process_inner(&mut self, input: &[u8], output: &mut [u8], end: bool) -> Result { + /// 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, + limit: usize, + ) -> Result { if self.done { if !input.is_empty() { return Err(corrupt()); @@ -211,6 +272,7 @@ impl StreamingDecoder { consumed: 0, written: 0, finished: true, + needs_input: false, }); } if output.is_empty() { @@ -218,6 +280,7 @@ impl StreamingDecoder { consumed: 0, written: 0, finished: false, + needs_input: false, }); } let mut consumed = 0; @@ -278,6 +341,7 @@ impl StreamingDecoder { consumed: 0, written: 0, finished: false, + needs_input: false, }); } let zlib = input[0] & 15 == 8 @@ -338,6 +402,7 @@ impl StreamingDecoder { consumed: 0, written: 0, finished: true, + needs_input: false, }); } let result = decoder @@ -345,10 +410,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()); } } @@ -364,6 +433,7 @@ impl StreamingDecoder { consumed: 0, written: 0, finished: true, + needs_input: false, }); } let mut source = input; @@ -373,6 +443,7 @@ impl StreamingDecoder { consumed: 0, written: 0, finished: false, + needs_input: false, }); } decoder.reset(&mut source).map_err(|_| corrupt())?; @@ -399,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", @@ -411,9 +482,188 @@ impl StreamingDecoder { consumed, written, finished, + needs_input: false, }) } } +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 737ced9..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() @@ -54,9 +64,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 +99,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 @@ -95,7 +138,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, @@ -109,6 +182,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 +277,7 @@ impl Decoder { head_request: false, connect_request: false, keep_alive: false, + upgrade_request: false, } } /// Must precede response parsing. Pipelining is intentionally not provided. @@ -212,6 +288,57 @@ 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) + } + /// 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")); @@ -288,6 +415,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 +563,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; @@ -441,23 +578,45 @@ 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. #[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 { @@ -465,23 +624,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")); @@ -493,13 +691,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()); @@ -511,7 +704,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"); @@ -525,11 +718,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/src/lib.rs b/protocols/turnloop-http/src/lib.rs index 90df02b..d574932 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/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/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..583cc78 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(); @@ -1309,3 +1323,563 @@ 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:?}"); + } +} + +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#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#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() { + 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()); +} + +/// 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); +} + +/// 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}"); + } +} + +/// 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"); +} 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),