diff --git a/crates/rttp-client/README.md b/crates/rttp-client/README.md index 3fb57853..2cbc6163 100644 --- a/crates/rttp-client/README.md +++ b/crates/rttp-client/README.md @@ -43,8 +43,18 @@ unsupported, `identity`, or parse-invalid stacks leave the original headers and body unchanged. Decoding is atomic: a malformed layer fails the response without exposing partial plaintext. Empty bodies are not decoded. `max_buffered_response_body_bytes` bounds each decoded layer. Raw -non-zlib `deflate` is not supported. Streaming bodies and async HTTP/2 -stay out of this buffered path. +non-zlib `deflate` is not supported. + +Streaming HTTP/1.1 body readers (`StreamingResponse` / +`AsyncStreamingResponse`) apply the same supported gzip and zlib-wrapped +deflate stacks incrementally without buffering the full body. Successful +streaming decode strips `Content-Encoding` and `Content-Length` from +`headers()` while `head()` keeps the raw capture; unsupported or +parse-invalid stacks leave headers and body bytes unchanged. Chunked +trailers remain visible after compressed EOF. Direct `body_mut()` reads +are not bounded by `max_buffered_response_body_bytes`; bounded helpers such +as `read_to_response` still enforce the decoded-size limit. Async HTTP/2 +streaming decode stays out of scope. ## Bounded Max-Forwards diagnostics @@ -1787,7 +1797,8 @@ header-block model. | area | tested coverage | limits | |------|-----------------|--------| | HTTP/1.1 response parsing | `Content-Length`, chunked transfer coding, chunk extensions, informational responses, bodyless `204`/`304`, duplicate `Set-Cookie`, and framing ambiguity rejection | Not a complete RFC conformance suite | -| Buffered content decoding | Automatic gzip and zlib-wrapped deflate stacks in reverse header order on buffered HTTP/1.1 and supported h2c paths; successful decoding drops stale `Content-Encoding`/`Content-Length`; unsupported or invalid stacks preserve headers and body; malformed layers fail atomically; size bounds apply per decoded layer; `Response::binary()` retains the original capture | No extra compression formats, raw deflate, streaming decode, or async HTTP/2 | +| Buffered content decoding | Automatic gzip and zlib-wrapped deflate stacks in reverse header order on buffered HTTP/1.1 and supported h2c paths; successful decoding drops stale `Content-Encoding`/`Content-Length`; unsupported or invalid stacks preserve headers and body; malformed layers fail atomically; size bounds apply per decoded layer; `Response::binary()` retains the original capture | No extra compression formats, raw deflate, or async HTTP/2 | +| Streaming content decoding | Incremental gzip and zlib-wrapped deflate stacks on blocking and async HTTP/1.1 streaming body readers; reverse order; empty bodies skipped; trailers remain after decoded chunked EOF; `headers()` strips CE/CL only on success while `head()` stays raw; unsupported stacks preserve raw bytes; malformed/raw-deflate map to decode errors; decoded bounds apply on `read_to_response` / buffered helpers, not direct `body_mut()` | No extra compression formats, raw deflate, or async HTTP/2 streaming decode | | HTTP/1.1 request emission | Origin-form requests, absolute-form proxy requests, `CONNECT`, `HEAD`, fixed bodies, streaming chunked uploads, and explicit `Expect: 100-continue` metadata through the shared protocol type | Expect metadata does not gate body transmission; raw `header(("Expect", value))` remains an escape hatch; SOCKS handshakes are delegated to the `socks` crate | | Fetch Metadata | `sec_fetch_site`, `sec_fetch_mode`, `sec_fetch_dest`, `sec_fetch_user`, and `sec_purpose` emit bounded `Sec-Fetch-*`/`Sec-Purpose` request metadata | No browser security policy, automatic header generation, origin validation, navigation policy, request blocking, prefetch execution, or cache behavior | | Save-Data | `save_data` emits bounded `Save-Data: on` request metadata | No reduced-data serving, content adaptation, compression, Client Hints advertisement, retries, or browser data-saver policy | diff --git a/crates/rttp-client/src/connection/async_connection.rs b/crates/rttp-client/src/connection/async_connection.rs index f33746ef..c1720da3 100644 --- a/crates/rttp-client/src/connection/async_connection.rs +++ b/crates/rttp-client/src/connection/async_connection.rs @@ -30,7 +30,10 @@ use crate::connection::connection_reader::{ }; use crate::error; use crate::request::RawRequest; -use crate::response::Response; +use crate::response::{ + content_decoders, strip_content_encoding_and_length, ContentDecoder, Response, + StreamingContentDecoder, +}; use crate::types::{Header, Proxy, ProxyType}; const CRLF: &[u8] = b"\r\n"; @@ -133,7 +136,11 @@ impl<'a, S: AsyncRead + Unpin + ?Sized> AsyncStreamingResponse<'a, S> { } pub fn headers(&self) -> error::Result> { - response_headers(&self.head) + let mut headers = response_headers(&self.head)?; + if self.body.content_decode_succeeded() { + strip_content_encoding_and_length(&mut headers); + } + Ok(headers) } pub fn head(&self) -> &[u8] { @@ -162,7 +169,10 @@ impl<'a, S: AsyncRead + Unpin + ?Sized> AsyncStreamingResponse<'a, S> { async fn read_to_parts(mut self, max_body_bytes: usize) -> error::Result { let close_connection = response_connection_should_close(&self.head)?; let connection_reusable = response_connection_reusable(&self.head, &self.body.kind)?; + let content_length = content_length_from_response_body_kind(&self.body.kind); let mut binary = self.head; + // Buffered responses still decode in RawResponse; keep wire bytes here. + self.body.disable_content_decode(); self .body .read_to_end_bounded(&mut binary, max_body_bytes) @@ -171,7 +181,7 @@ impl<'a, S: AsyncRead + Unpin + ?Sized> AsyncStreamingResponse<'a, S> { binary, trailers: self.body.trailers().clone(), informational_responses: Vec::new(), - content_length: content_length_from_response_body_kind(&self.body.kind), + content_length, connection_reusable, close_connection, }) @@ -186,15 +196,24 @@ pub struct AsyncResponseBodyReader<'a, S: AsyncRead + Unpin + ?Sized> { chunk_needs_crlf: bool, trailers: Vec
, eof: bool, + content_decode: Option, + framed_eof: bool, } impl<'a, S: AsyncRead + Unpin + ?Sized> AsyncResponseBodyReader<'a, S> { - fn new(stream: &'a mut S, kind: ResponseBodyKind) -> Self { + fn new_with_content_decode( + stream: &'a mut S, + kind: ResponseBodyKind, + content_decoders: Option>, + ) -> Self { let remaining = match kind { ResponseBodyKind::ContentLength(length) => length, _ => 0, }; let eof = matches!(kind, ResponseBodyKind::NoBody); + let content_decode = content_decoders + .filter(|_| should_attempt_async_content_decode(&kind)) + .map(|decoders| StreamingContentDecoder::new(&decoders, None)); Self { stream, kind, @@ -203,6 +222,8 @@ impl<'a, S: AsyncRead + Unpin + ?Sized> AsyncResponseBodyReader<'a, S> { chunk_needs_crlf: false, trailers: Vec::new(), eof, + content_decode, + framed_eof: eof, } } @@ -210,19 +231,19 @@ impl<'a, S: AsyncRead + Unpin + ?Sized> AsyncResponseBodyReader<'a, S> { &self.trailers } + fn content_decode_succeeded(&self) -> bool { + self + .content_decode + .as_ref() + .is_some_and(StreamingContentDecoder::succeeded) + } + + fn disable_content_decode(&mut self) { + self.content_decode = None; + } + pub async fn read(&mut self, buf: &mut [u8]) -> error::Result { - match self.kind { - ResponseBodyKind::NoBody => Ok(0), - ResponseBodyKind::ContentLength(_) => self.read_fixed_length(buf).await, - ResponseBodyKind::Chunked => self.read_chunked(buf).await, - ResponseBodyKind::UntilEof => { - let read = self.stream.read(buf).await.map_err(error::request)?; - if read == 0 { - self.eof = true; - } - Ok(read) - } - } + self.read_decoded(buf).await } pub async fn read_to_end(&mut self, body: &mut Vec) -> error::Result { @@ -259,9 +280,70 @@ impl<'a, S: AsyncRead + Unpin + ?Sized> AsyncResponseBodyReader<'a, S> { } } + async fn read_decoded(&mut self, buf: &mut [u8]) -> error::Result { + if buf.is_empty() { + return Ok(0); + } + + loop { + if let Some(decoder) = self.content_decode.as_mut() { + let filled = decoder.fill(buf); + if filled > 0 { + return Ok(filled); + } + if decoder.finished() { + self.eof = true; + return Ok(0); + } + } else { + return self.read_framed(buf).await; + } + + if self.framed_eof { + let decoder = self + .content_decode + .as_mut() + .expect("content decoder present when decoding"); + decoder.finish()?; + let filled = decoder.fill(buf); + self.eof = !decoder.has_pending(); + return Ok(filled); + } + + let mut compressed = [0u8; 8 * 1024]; + let read = self.read_framed(&mut compressed).await?; + if read == 0 { + self.framed_eof = true; + continue; + } + self + .content_decode + .as_mut() + .expect("content decoder present when decoding") + .feed(&compressed[..read])?; + } + } + + async fn read_framed(&mut self, buf: &mut [u8]) -> error::Result { + match self.kind { + ResponseBodyKind::NoBody => Ok(0), + ResponseBodyKind::ContentLength(_) => self.read_fixed_length(buf).await, + ResponseBodyKind::Chunked => self.read_chunked(buf).await, + ResponseBodyKind::UntilEof => { + let read = self.stream.read(buf).await.map_err(error::request)?; + if read == 0 { + self.eof = true; + self.framed_eof = true; + } + Ok(read) + } + } + } + async fn read_fixed_length(&mut self, buf: &mut [u8]) -> error::Result { if self.remaining == 0 || buf.is_empty() { self.eof = self.remaining == 0; + self.framed_eof = self.remaining == 0; return Ok(0); } @@ -280,12 +362,13 @@ impl<'a, S: AsyncRead + Unpin + ?Sized> AsyncResponseBodyReader<'a, S> { self.remaining -= read; if self.remaining == 0 { self.eof = true; + self.framed_eof = true; } Ok(read) } async fn read_chunked(&mut self, buf: &mut [u8]) -> error::Result { - if self.eof || buf.is_empty() { + if self.framed_eof || buf.is_empty() { return Ok(0); } @@ -300,6 +383,7 @@ impl<'a, S: AsyncRead + Unpin + ?Sized> AsyncResponseBodyReader<'a, S> { if chunk_size == 0 { self.trailers = async_read_trailers(self.stream).await?; self.eof = true; + self.framed_eof = true; return Ok(0); } self.chunk_remaining = chunk_size; @@ -322,6 +406,13 @@ impl<'a, S: AsyncRead + Unpin + ?Sized> AsyncResponseBodyReader<'a, S> { } } +fn should_attempt_async_content_decode(kind: &ResponseBodyKind) -> bool { + !matches!( + kind, + ResponseBodyKind::NoBody | ResponseBodyKind::ContentLength(0) + ) +} + pub struct AsyncConnection<'a> { conn: Connection<'a>, } @@ -694,9 +785,12 @@ where S: AsyncRead + Unpin + ?Sized, { let kind = response_body_kind(&head, expect_no_body)?; + let decoders = response_headers(&head) + .ok() + .and_then(|headers| content_decoders(&headers)); Ok(AsyncStreamingResponse { head, - body: AsyncResponseBodyReader::new(stream, kind), + body: AsyncResponseBodyReader::new_with_content_decode(stream, kind, decoders), }) } diff --git a/crates/rttp-client/src/connection/connection_reader.rs b/crates/rttp-client/src/connection/connection_reader.rs index 614df6f2..dde953ed 100644 --- a/crates/rttp-client/src/connection/connection_reader.rs +++ b/crates/rttp-client/src/connection/connection_reader.rs @@ -10,7 +10,10 @@ use url::Url; use crate::config::DEFAULT_MAX_BUFFERED_RESPONSE_BODY_BYTES; use crate::error; -use crate::response::{InformationalResponse, Response}; +use crate::response::{ + content_decoders, strip_content_encoding_and_length, ContentDecoder, InformationalResponse, + Response, StreamingContentDecoder, +}; use crate::types::{Header, RoUrl}; const HEADER_END: &[u8] = b"\r\n\r\n"; @@ -39,6 +42,7 @@ pub struct StreamingResponse<'a, R: Read + ?Sized> { url: RoUrl, head: Vec, body: ResponseBodyReader<'a, R>, + max_buffered_response_body_bytes: usize, } impl<'a, R: Read + ?Sized> StreamingResponse<'a, R> { @@ -47,7 +51,11 @@ impl<'a, R: Read + ?Sized> StreamingResponse<'a, R> { } pub fn headers(&self) -> error::Result> { - response_headers(&self.head) + let mut headers = response_headers(&self.head)?; + if self.body.content_decode_succeeded() { + strip_content_encoding_and_length(&mut headers); + } + Ok(headers) } pub fn head(&self) -> &[u8] { @@ -74,20 +82,28 @@ impl<'a, R: Read + ?Sized> StreamingResponse<'a, R> { } pub fn read_to_response(mut self) -> error::Result { - let mut binary = self.head.clone(); - let content_length = content_length_from_response_body_kind(&self.body.kind); - read_response_body_to_end( - &mut self.body, - &mut binary, - DEFAULT_MAX_BUFFERED_RESPONSE_BODY_BYTES, - )?; + let max_body_bytes = self.max_buffered_response_body_bytes; + self.body.set_max_decoded_bytes(Some(max_body_bytes)); + let mut decoded_body = Vec::new(); + read_response_body_to_end(&mut self.body, &mut decoded_body, max_body_bytes)?; + let content_length = if self.body.content_decode_succeeded() { + None + } else { + content_length_from_response_body_kind(&self.body.kind) + }; + let mut binary = if self.body.content_decode_succeeded() { + strip_content_encoding_and_length_from_head(&self.head)? + } else { + self.head.clone() + }; + binary.extend_from_slice(&decoded_body); Response::with_trailers_and_informational_and_limit( self.url, binary, self.body.trailers().clone(), Vec::new(), content_length, - DEFAULT_MAX_BUFFERED_RESPONSE_BODY_BYTES, + max_body_bytes, ) } } @@ -100,15 +116,28 @@ pub struct ResponseBodyReader<'a, R: Read + ?Sized> { chunk_needs_crlf: bool, trailers: Vec
, eof: bool, + content_decode: Option, + framed_eof: bool, } impl<'a, R: Read + ?Sized> ResponseBodyReader<'a, R> { fn new(reader: &'a mut R, kind: ResponseBodyKind) -> Self { + Self::new_with_content_decode(reader, kind, None) + } + + fn new_with_content_decode( + reader: &'a mut R, + kind: ResponseBodyKind, + content_decoders: Option>, + ) -> Self { let remaining = match kind { ResponseBodyKind::ContentLength(length) => length, _ => 0, }; let eof = matches!(kind, ResponseBodyKind::NoBody); + let content_decode = content_decoders + .filter(|_| should_attempt_content_decode(&kind)) + .map(|decoders| StreamingContentDecoder::new(&decoders, None)); Self { reader, kind, @@ -117,6 +146,8 @@ impl<'a, R: Read + ?Sized> ResponseBodyReader<'a, R> { chunk_needs_crlf: false, trailers: Vec::new(), eof, + content_decode, + framed_eof: eof, } } @@ -124,9 +155,84 @@ impl<'a, R: Read + ?Sized> ResponseBodyReader<'a, R> { &self.trailers } + fn content_decode_succeeded(&self) -> bool { + self + .content_decode + .as_ref() + .is_some_and(StreamingContentDecoder::succeeded) + } + + fn set_max_decoded_bytes(&mut self, max_decoded_bytes: Option) { + if let Some(decoder) = self.content_decode.as_mut() { + decoder.set_max_decoded_bytes(max_decoded_bytes); + } + } + + fn read_framed(&mut self, buf: &mut [u8]) -> io::Result { + match self.kind { + ResponseBodyKind::NoBody => Ok(0), + ResponseBodyKind::ContentLength(_) => self.read_fixed_length(buf), + ResponseBodyKind::Chunked => self.read_chunked(buf), + ResponseBodyKind::UntilEof => { + let read = self.reader.read(buf)?; + if read == 0 { + self.eof = true; + self.framed_eof = true; + } + Ok(read) + } + } + } + + fn read_decoded(&mut self, buf: &mut [u8]) -> io::Result { + if buf.is_empty() { + return Ok(0); + } + + loop { + if let Some(decoder) = self.content_decode.as_mut() { + let filled = decoder.fill(buf); + if filled > 0 { + return Ok(filled); + } + if decoder.finished() { + self.eof = true; + return Ok(0); + } + } else { + return self.read_framed(buf); + } + + if self.framed_eof { + let decoder = self + .content_decode + .as_mut() + .expect("content decoder present when decoding"); + decoder.finish().map_err(error_to_io_error)?; + let filled = decoder.fill(buf); + self.eof = !decoder.has_pending(); + return Ok(filled); + } + + let mut compressed = [0u8; 8 * 1024]; + let read = self.read_framed(&mut compressed)?; + if read == 0 { + self.framed_eof = true; + continue; + } + self + .content_decode + .as_mut() + .expect("content decoder present when decoding") + .feed(&compressed[..read]) + .map_err(error_to_io_error)?; + } + } + fn read_fixed_length(&mut self, buf: &mut [u8]) -> io::Result { if self.remaining == 0 || buf.is_empty() { self.eof = self.remaining == 0; + self.framed_eof = self.remaining == 0; return Ok(0); } @@ -141,12 +247,13 @@ impl<'a, R: Read + ?Sized> ResponseBodyReader<'a, R> { self.remaining -= read; if self.remaining == 0 { self.eof = true; + self.framed_eof = true; } Ok(read) } fn read_chunked(&mut self, buf: &mut [u8]) -> io::Result { - if self.eof || buf.is_empty() { + if self.framed_eof || buf.is_empty() { return Ok(0); } @@ -162,6 +269,7 @@ impl<'a, R: Read + ?Sized> ResponseBodyReader<'a, R> { if chunk_size == 0 { self.trailers = read_trailers(self.reader).map_err(to_io_error)?; self.eof = true; + self.framed_eof = true; return Ok(0); } self.chunk_remaining = chunk_size; @@ -185,18 +293,7 @@ impl<'a, R: Read + ?Sized> ResponseBodyReader<'a, R> { impl Read for ResponseBodyReader<'_, R> { fn read(&mut self, buf: &mut [u8]) -> io::Result { - match self.kind { - ResponseBodyKind::NoBody => Ok(0), - ResponseBodyKind::ContentLength(_) => self.read_fixed_length(buf), - ResponseBodyKind::Chunked => self.read_chunked(buf), - ResponseBodyKind::UntilEof => { - let read = self.reader.read(buf)?; - if read == 0 { - self.eof = true; - } - Ok(read) - } - } + self.read_decoded(buf) } } @@ -259,10 +356,14 @@ impl<'a> ConnectionReader<'a> { pub fn streaming_response(&mut self) -> error::Result> { let head = read_response_head(self.reader)?; let kind = response_body_kind(&head, self.expect_no_body)?; + let decoders = response_headers(&head) + .ok() + .and_then(|headers| content_decoders(&headers)); Ok(StreamingResponse { url: RoUrl::from(self.url.clone()), head, - body: ResponseBodyReader::new(self.reader, kind), + body: ResponseBodyReader::new_with_content_decode(self.reader, kind, decoders), + max_buffered_response_body_bytes: self.max_buffered_response_body_bytes, }) } @@ -847,15 +948,75 @@ fn to_io_error(err: error::Error) -> io::Error { } } +fn error_to_io_error(err: error::Error) -> io::Error { + io::Error::other(err) +} + fn response_body_read_error(err: io::Error) -> error::Error { - match err.kind() { - io::ErrorKind::InvalidData | io::ErrorKind::UnexpectedEof => { - error::bad_response(err.to_string()) + let kind = err.kind(); + let display = err.to_string(); + if let Some(inner) = err.into_inner() { + match inner.downcast::() { + Ok(error) => return *error, + Err(inner) => { + let err = io::Error::new(kind, inner); + return match kind { + io::ErrorKind::InvalidData | io::ErrorKind::UnexpectedEof => { + error::bad_response(err.to_string()) + } + _ => error::request(err), + }; + } } - _ => error::request(err), + } + match kind { + io::ErrorKind::InvalidData | io::ErrorKind::UnexpectedEof => error::bad_response(display), + _ => error::request(io::Error::new(kind, display)), } } +fn should_attempt_content_decode(kind: &ResponseBodyKind) -> bool { + !matches!( + kind, + ResponseBodyKind::NoBody | ResponseBodyKind::ContentLength(0) + ) +} + +fn strip_content_encoding_and_length_from_head(head: &[u8]) -> error::Result> { + let header_end = head + .windows(HEADER_END.len()) + .position(|window| window == HEADER_END) + .ok_or_else(|| error::bad_response("Invalid response header"))?; + let header_block = &head[..header_end]; + let mut lines = header_block + .split(|byte| *byte == b'\n') + .map(|line| line.strip_suffix(b"\r").unwrap_or(line)); + let status_line = lines + .next() + .ok_or_else(|| error::bad_response("Response not have status line"))?; + let mut out = Vec::with_capacity(head.len()); + out.extend_from_slice(status_line); + out.extend_from_slice(CRLF); + for line in lines { + if line.is_empty() { + continue; + } + let Some(colon) = line.iter().position(|byte| *byte == b':') else { + return Err(error::bad_response("Invalid response header")); + }; + let name = &line[..colon]; + if name.eq_ignore_ascii_case(b"Content-Encoding") + || name.eq_ignore_ascii_case(b"Content-Length") + { + continue; + } + out.extend_from_slice(line); + out.extend_from_slice(CRLF); + } + out.extend_from_slice(CRLF); + Ok(out) +} + pub(crate) fn validate_response_trailer_header(name: &str, value: &str) -> error::Result<()> { if !is_http_token(name) || !value.bytes().all(is_header_value_byte) { return Err(error::bad_response("Invalid trailer header")); @@ -1417,6 +1578,63 @@ mod tests { assert_eq!(2, content_length.len()); } + #[test] + fn streaming_gzip_read_to_response_enforces_decoded_limit() { + use flate2::write::GzEncoder; + use flate2::Compression; + use std::io::Write; + + let decoded = vec![b'a'; 256]; + let mut encoder = GzEncoder::new(Vec::new(), Compression::default()); + encoder.write_all(&decoded).unwrap(); + let compressed = encoder.finish().unwrap(); + assert!(compressed.len() <= 64); + + let mut raw = format!( + "HTTP/1.1 200 OK\r\nContent-Encoding: gzip\r\nContent-Length: {}\r\n\r\n", + compressed.len() + ) + .into_bytes(); + raw.extend_from_slice(&compressed); + let url = url::Url::parse("http://localhost").unwrap(); + let mut cursor = Cursor::new(raw); + let mut reader = ConnectionReader::new_with_limit(&url, &mut cursor, false, 64); + let error = reader + .streaming_response() + .unwrap() + .read_to_response() + .expect_err("oversized decoded streaming body should fail"); + assert!(error.is_body_too_large(), "unexpected error: {error}"); + assert_eq!(Some(64), error.body_limit()); + } + + #[test] + fn streaming_gzip_chunked_exposes_trailers_after_decoded_eof() { + use flate2::write::GzEncoder; + use flate2::Compression; + use std::io::Write; + + let mut encoder = GzEncoder::new(Vec::new(), Compression::default()); + encoder.write_all(b"hello").unwrap(); + let compressed = encoder.finish().unwrap(); + let mut raw = + b"HTTP/1.1 200 OK\r\nContent-Encoding: gzip\r\nTransfer-Encoding: chunked\r\n\r\n".to_vec(); + raw.extend_from_slice(format!("{:x}\r\n", compressed.len()).as_bytes()); + raw.extend_from_slice(&compressed); + raw.extend_from_slice(b"\r\n0\r\nX-Trace: abc\r\n\r\n"); + let url = url::Url::parse("http://localhost").unwrap(); + let mut cursor = Cursor::new(raw); + let mut reader = ConnectionReader::new(&url, &mut cursor, false); + let mut response = reader.streaming_response().unwrap(); + let mut body = Vec::new(); + response.body_mut().read_to_end(&mut body).unwrap(); + assert_eq!(b"hello", body.as_slice()); + assert_eq!( + Some("abc"), + response.trailer_value("x-trace").map(String::as_str) + ); + } + #[test] fn test_duplicate_content_length_with_different_values_is_rejected() { let raw = concat!( diff --git a/crates/rttp-client/src/response/content_decode.rs b/crates/rttp-client/src/response/content_decode.rs new file mode 100644 index 00000000..c08e9041 --- /dev/null +++ b/crates/rttp-client/src/response/content_decode.rs @@ -0,0 +1,313 @@ +use std::io::{self, Write}; + +use rttp_protocol::content_encoding::ContentEncoding; + +use crate::error; +use crate::types::Header; + +#[derive(Clone, Copy, Debug, Eq, PartialEq)] +pub(crate) enum ContentDecoder { + Gzip, + Deflate, +} + +/// Select a fully supported gzip / zlib-deflate stack for decoding. +/// +/// Returns `None` for missing, empty, identity, unknown, mixed-unsupported, or +/// parse-invalid `Content-Encoding` values so callers preserve raw headers and +/// body bytes. +pub(crate) fn content_decoders(headers: &[Header]) -> Option> { + let parsed = ContentEncoding::parse_values( + headers + .iter() + .filter(|header| header.name().eq_ignore_ascii_case("Content-Encoding")) + .map(|header| header.value().as_str()), + ) + .ok()?; + if parsed.codings().is_empty() { + return None; + } + let mut decoders = Vec::with_capacity(parsed.len()); + for coding in parsed.codings() { + if coding.eq_ignore_ascii_case("gzip") { + decoders.push(ContentDecoder::Gzip); + } else if coding.eq_ignore_ascii_case("deflate") { + decoders.push(ContentDecoder::Deflate); + } else { + return None; + } + } + Some(decoders) +} + +pub(crate) fn strip_content_encoding_and_length(headers: &mut Vec
) { + headers.retain(|header| { + !header.name().eq_ignore_ascii_case("Content-Encoding") + && !header.name().eq_ignore_ascii_case("Content-Length") + }); +} + +enum LayerWriter { + Gzip(flate2::write::GzDecoder>), + Deflate(flate2::write::ZlibDecoder>), +} + +impl LayerWriter { + fn new(decoder: ContentDecoder) -> Self { + match decoder { + ContentDecoder::Gzip => Self::Gzip(flate2::write::GzDecoder::new(Vec::new())), + ContentDecoder::Deflate => Self::Deflate(flate2::write::ZlibDecoder::new(Vec::new())), + } + } + + fn write_all(&mut self, input: &[u8]) -> io::Result<()> { + match self { + Self::Gzip(writer) => writer.write_all(input), + Self::Deflate(writer) => writer.write_all(input), + } + } + + fn take_output(&mut self) -> io::Result> { + match self { + Self::Gzip(writer) => { + writer.flush()?; + Ok(std::mem::take(writer.get_mut())) + } + Self::Deflate(writer) => { + writer.flush()?; + Ok(std::mem::take(writer.get_mut())) + } + } + } + + fn finish(self) -> io::Result> { + match self { + Self::Gzip(writer) => writer.finish(), + Self::Deflate(writer) => writer.finish(), + } + } +} + +/// Incremental reverse-order gzip / zlib-deflate decoder for streaming bodies. +pub(crate) struct StreamingContentDecoder { + layers: Vec, + layer_decoded_bytes: Vec, + pending: Vec, + pending_pos: usize, + max_decoded_bytes: Option, + saw_input: bool, + finished: bool, + succeeded: bool, +} + +impl StreamingContentDecoder { + pub(crate) fn new(decoders: &[ContentDecoder], max_decoded_bytes: Option) -> Self { + // Header order is applied outer-last on the wire; decode in reverse. + let layers: Vec = decoders + .iter() + .rev() + .copied() + .map(LayerWriter::new) + .collect(); + let layer_count = layers.len(); + Self { + layers, + layer_decoded_bytes: vec![0; layer_count], + pending: Vec::new(), + pending_pos: 0, + max_decoded_bytes, + saw_input: false, + finished: false, + succeeded: false, + } + } + + pub(crate) fn succeeded(&self) -> bool { + self.succeeded + } + + pub(crate) fn finished(&self) -> bool { + self.finished + } + + pub(crate) fn set_max_decoded_bytes(&mut self, max_decoded_bytes: Option) { + self.max_decoded_bytes = max_decoded_bytes; + } + + fn record_layer_bytes(&mut self, layer_index: usize, produced: usize) -> error::Result<()> { + if produced == 0 { + return Ok(()); + } + let total = self.layer_decoded_bytes[layer_index] + .checked_add(produced) + .ok_or_else(|| { + self + .max_decoded_bytes + .map(error::body_too_large) + .unwrap_or_else(|| error::decode("decoded response body is too large")) + })?; + if let Some(max) = self.max_decoded_bytes { + if total > max { + return Err(error::body_too_large(max)); + } + } + self.layer_decoded_bytes[layer_index] = total; + Ok(()) + } + + fn push_pending(&mut self, bytes: Vec) { + if bytes.is_empty() { + return; + } + if self.pending_pos > 0 { + self.pending.drain(..self.pending_pos); + self.pending_pos = 0; + } + self.pending.extend_from_slice(&bytes); + } + + fn pass_through_layers(&mut self, mut current: Vec) -> error::Result<()> { + for index in 0..self.layers.len() { + if current.is_empty() { + break; + } + self.layers[index] + .write_all(¤t) + .map_err(error::decode)?; + current = self.layers[index].take_output().map_err(error::decode)?; + self.record_layer_bytes(index, current.len())?; + } + self.push_pending(current); + Ok(()) + } + + pub(crate) fn feed(&mut self, compressed: &[u8]) -> error::Result<()> { + if compressed.is_empty() { + return Ok(()); + } + if self.finished { + return Err(error::decode("content decoder already finished")); + } + self.saw_input = true; + self.pass_through_layers(compressed.to_vec()) + } + + pub(crate) fn finish(&mut self) -> error::Result<()> { + if self.finished { + return Ok(()); + } + self.finished = true; + if !self.saw_input { + // Empty wire bodies are not decoded; leave headers untouched. + self.layers.clear(); + return Ok(()); + } + + let layers = std::mem::take(&mut self.layers); + let mut carry = Vec::new(); + for (index, layer) in layers.into_iter().enumerate() { + let mut layer = layer; + if !carry.is_empty() { + layer.write_all(&carry).map_err(error::decode)?; + } + let finished = layer.finish().map_err(error::decode)?; + self.record_layer_bytes(index, finished.len())?; + carry = finished; + } + self.push_pending(carry); + self.succeeded = true; + Ok(()) + } + + pub(crate) fn fill(&mut self, buf: &mut [u8]) -> usize { + if self.pending_pos >= self.pending.len() || buf.is_empty() { + return 0; + } + let available = &self.pending[self.pending_pos..]; + let n = available.len().min(buf.len()); + buf[..n].copy_from_slice(&available[..n]); + self.pending_pos += n; + if self.pending_pos >= self.pending.len() { + self.pending.clear(); + self.pending_pos = 0; + } + n + } + + pub(crate) fn has_pending(&self) -> bool { + self.pending_pos < self.pending.len() + } +} + +#[cfg(test)] +mod tests { + use super::*; + use flate2::write::{GzEncoder, ZlibEncoder}; + use flate2::Compression; + use std::io::Write; + + fn gzip(bytes: &[u8]) -> Vec { + let mut encoder = GzEncoder::new(Vec::new(), Compression::default()); + encoder.write_all(bytes).unwrap(); + encoder.finish().unwrap() + } + + fn zlib(bytes: &[u8]) -> Vec { + let mut encoder = ZlibEncoder::new(Vec::new(), Compression::default()); + encoder.write_all(bytes).unwrap(); + encoder.finish().unwrap() + } + + #[test] + fn streams_single_gzip() { + let compressed = gzip(b"hello"); + let mut decoder = StreamingContentDecoder::new(&[ContentDecoder::Gzip], None); + decoder.feed(&compressed[..3]).unwrap(); + decoder.feed(&compressed[3..]).unwrap(); + decoder.finish().unwrap(); + let mut out = vec![0; 16]; + let n = decoder.fill(&mut out); + assert_eq!(b"hello", &out[..n]); + assert!(decoder.succeeded()); + } + + #[test] + fn streams_gzip_then_deflate_stack() { + // Header order gzip, deflate => wire is zlib(gzip(data)) + let compressed = zlib(&gzip(b"OK")); + let mut decoder = + StreamingContentDecoder::new(&[ContentDecoder::Gzip, ContentDecoder::Deflate], None); + for chunk in compressed.chunks(2) { + decoder.feed(chunk).unwrap(); + } + decoder.finish().unwrap(); + let mut out = Vec::new(); + let mut buf = [0u8; 8]; + loop { + let n = decoder.fill(&mut buf); + if n == 0 { + break; + } + out.extend_from_slice(&buf[..n]); + } + assert_eq!(b"OK", out.as_slice()); + assert!(decoder.succeeded()); + } + + #[test] + fn empty_input_skips_success() { + let mut decoder = StreamingContentDecoder::new(&[ContentDecoder::Gzip], None); + decoder.finish().unwrap(); + assert!(!decoder.succeeded()); + } + + #[test] + fn rejects_raw_deflate() { + let mut encoder = flate2::write::DeflateEncoder::new(Vec::new(), Compression::default()); + encoder.write_all(b"OK").unwrap(); + let raw = encoder.finish().unwrap(); + let mut decoder = StreamingContentDecoder::new(&[ContentDecoder::Deflate], None); + let err = decoder.feed(&raw).err().or_else(|| decoder.finish().err()); + assert!(err.is_some()); + } +} diff --git a/crates/rttp-client/src/response/mod.rs b/crates/rttp-client/src/response/mod.rs index 75672832..e3a22a3c 100644 --- a/crates/rttp-client/src/response/mod.rs +++ b/crates/rttp-client/src/response/mod.rs @@ -101,9 +101,14 @@ pub use rttp_protocol::cross_origin_resource_policy::{ }; pub use rttp_protocol::nel::{Nel, NelParseError, NelUnknownMember}; +pub(crate) mod content_decode; mod raw_response; mod response; +pub(crate) use content_decode::{ + content_decoders, strip_content_encoding_and_length, ContentDecoder, StreamingContentDecoder, +}; + pub use rttp_protocol::alt_svc::{AltSvc, AltSvcAlternative, AltSvcParameter, AltSvcParseError}; pub use rttp_protocol::alt_used::{AltUsed, AltUsedParseError}; pub use rttp_protocol::alternates::{ diff --git a/crates/rttp-client/src/response/raw_response.rs b/crates/rttp-client/src/response/raw_response.rs index ec84f80e..24ca4126 100644 --- a/crates/rttp-client/src/response/raw_response.rs +++ b/crates/rttp-client/src/response/raw_response.rs @@ -3,9 +3,9 @@ use std::io::Read; use crate::config::DEFAULT_MAX_BUFFERED_RESPONSE_BODY_BYTES; use crate::error; +use crate::response::content_decode::{content_decoders, ContentDecoder}; use crate::response::ResponseBody; use crate::types::{is_sensitive_debug_header, Cookie, Header, RoUrl, ToUrl}; -use rttp_protocol::content_encoding::ContentEncoding; use rttp_protocol::cookie::HttpSetCookie; use url::Url; @@ -358,29 +358,3 @@ fn decode_http1_text(bytes: &[u8]) -> String { fn response_status_has_no_body(status_code: u32) -> bool { (100..200).contains(&status_code) || status_code == 204 || status_code == 304 } - -enum ContentDecoder { - Gzip, - Deflate, -} - -fn content_decoders(headers: &[Header]) -> Option> { - let parsed = ContentEncoding::parse_values( - headers - .iter() - .filter(|header| header.name().eq_ignore_ascii_case("Content-Encoding")) - .map(|header| header.value().as_str()), - ) - .ok()?; - let mut decoders = Vec::with_capacity(parsed.len()); - for coding in parsed.codings() { - if coding.eq_ignore_ascii_case("gzip") { - decoders.push(ContentDecoder::Gzip); - } else if coding.eq_ignore_ascii_case("deflate") { - decoders.push(ContentDecoder::Deflate); - } else { - return None; - } - } - Some(decoders) -} diff --git a/crates/rttp-client/tests/test_http_async.rs b/crates/rttp-client/tests/test_http_async.rs index 14af9b6f..cac1445f 100644 --- a/crates/rttp-client/tests/test_http_async.rs +++ b/crates/rttp-client/tests/test_http_async.rs @@ -365,6 +365,134 @@ fn test_async_streaming_response_can_read_body_larger_than_buffered_limit() { }); } +#[test] +#[cfg(feature = "async")] +fn test_async_streaming_gzip_fixed_length_decodes_body_and_strips_headers() { + block_on(async { + let compressed = gzip_bytes(b"decoded"); + let head = format!( + "HTTP/1.1 200 OK\r\nContent-Encoding: gzip\r\nContent-Length: {}\r\nX-Trace: keep\r\n\r\n", + compressed.len() + ) + .into_bytes(); + let mut stream = AllowStdIo::new(Cursor::new(compressed)); + let mut response = async_streaming_response_after_header(&mut stream, false, head) + .await + .unwrap(); + let mut body = Vec::new(); + response.body_mut().read_to_end(&mut body).await.unwrap(); + + assert_eq!(b"decoded", body.as_slice()); + let headers = response.headers().unwrap(); + assert!(headers + .iter() + .all(|header| !header.name().eq_ignore_ascii_case("Content-Encoding"))); + assert!(headers + .iter() + .all(|header| !header.name().eq_ignore_ascii_case("Content-Length"))); + assert!(String::from_utf8_lossy(response.head()).contains("Content-Encoding: gzip")); + }); +} + +#[test] +#[cfg(feature = "async")] +fn test_async_streaming_gzip_chunked_decodes_body_and_exposes_trailers() { + block_on(async { + let compressed = gzip_bytes(b"hello"); + let head = + b"HTTP/1.1 200 OK\r\nContent-Encoding: gzip\r\nTransfer-Encoding: chunked\r\n\r\n".to_vec(); + let mut body = format!("{:x}\r\n", compressed.len()).into_bytes(); + body.extend_from_slice(&compressed); + body.extend_from_slice(b"\r\n0\r\nX-Trace: trailer\r\n\r\n"); + let mut stream = AllowStdIo::new(Cursor::new(body)); + let mut response = async_streaming_response_after_header(&mut stream, false, head) + .await + .unwrap(); + let mut out = Vec::new(); + response.body_mut().read_to_end(&mut out).await.unwrap(); + + assert_eq!(b"hello", out.as_slice()); + assert_eq!( + Some("trailer"), + response.trailer_value("x-trace").map(String::as_str) + ); + }); +} + +#[test] +#[cfg(feature = "async")] +fn test_async_streaming_stacked_gzip_deflate_decodes_body() { + block_on(async { + let compressed = zlib_bytes(&gzip_bytes(b"stacked")); + let head = format!( + "HTTP/1.1 200 OK\r\nContent-Encoding: gzip, deflate\r\nContent-Length: {}\r\n\r\n", + compressed.len() + ) + .into_bytes(); + let mut stream = AllowStdIo::new(Cursor::new(compressed)); + let mut response = async_streaming_response_after_header(&mut stream, false, head) + .await + .unwrap(); + let mut body = Vec::new(); + response.body_mut().read_to_end(&mut body).await.unwrap(); + assert_eq!(b"stacked", body.as_slice()); + }); +} + +#[test] +#[cfg(feature = "async")] +fn test_async_streaming_malformed_deflate_returns_decode_error() { + block_on(async { + let body = b"not-zlib".to_vec(); + let head = format!( + "HTTP/1.1 200 OK\r\nContent-Encoding: deflate\r\nContent-Length: {}\r\n\r\n", + body.len() + ) + .into_bytes(); + let mut stream = AllowStdIo::new(Cursor::new(body)); + let mut response = async_streaming_response_after_header(&mut stream, false, head) + .await + .unwrap(); + let mut out = Vec::new(); + let error = response + .body_mut() + .read_to_end(&mut out) + .await + .expect_err("malformed deflate should fail"); + assert_decode_error(error); + }); +} + +#[test] +#[cfg(feature = "async")] +fn test_async_streaming_unknown_content_encoding_preserves_raw() { + block_on(async { + let body = gzip_bytes(b"OK"); + let head = format!( + "HTTP/1.1 200 OK\r\nContent-Encoding: gzip, br\r\nContent-Length: {}\r\n\r\n", + body.len() + ) + .into_bytes(); + let expected = body.clone(); + let mut stream = AllowStdIo::new(Cursor::new(body)); + let mut response = async_streaming_response_after_header(&mut stream, false, head) + .await + .unwrap(); + let mut out = Vec::new(); + response.body_mut().read_to_end(&mut out).await.unwrap(); + assert_eq!(expected, out); + assert_eq!( + Some("gzip, br"), + response + .headers() + .unwrap() + .iter() + .find(|header| header.name().eq_ignore_ascii_case("Content-Encoding")) + .map(|header| header.value().as_str()) + ); + }); +} + #[test] #[cfg(feature = "async")] fn test_async_buffered_gzip_response_exposes_decoded_body_headers() { diff --git a/crates/rttp-client/tests/test_http_basic.rs b/crates/rttp-client/tests/test_http_basic.rs index 885901b2..c28bc3dd 100644 --- a/crates/rttp-client/tests/test_http_basic.rs +++ b/crates/rttp-client/tests/test_http_basic.rs @@ -366,6 +366,185 @@ fn test_streaming_response_can_read_body_larger_than_buffered_limit() { assert_eq!(body_len, body.len()); } +#[test] +fn test_streaming_gzip_fixed_length_decodes_body_and_strips_headers() { + let compressed = gzip_bytes(b"decoded"); + let mut raw = format!( + "HTTP/1.1 200 OK\r\nContent-Encoding: gzip\r\nContent-Length: {}\r\nX-Trace: keep\r\n\r\n", + compressed.len() + ) + .into_bytes(); + raw.extend_from_slice(&compressed); + let mut cursor = Cursor::new(raw); + let url = url::Url::parse("http://localhost/gzip").unwrap(); + let mut reader = ConnectionReader::new(&url, &mut cursor, false); + let mut response = reader.streaming_response().unwrap(); + let mut body = Vec::new(); + response.body_mut().read_to_end(&mut body).unwrap(); + + assert_eq!(b"decoded", body.as_slice()); + let headers = response.headers().unwrap(); + assert!(headers + .iter() + .all(|header| !header.name().eq_ignore_ascii_case("Content-Encoding"))); + assert!(headers + .iter() + .all(|header| !header.name().eq_ignore_ascii_case("Content-Length"))); + assert!(headers + .iter() + .any(|header| header.name().eq_ignore_ascii_case("X-Trace"))); + assert!(String::from_utf8_lossy(response.head()).contains("Content-Encoding: gzip")); +} + +#[test] +fn test_streaming_gzip_chunked_decodes_body_and_exposes_trailers() { + let compressed = gzip_bytes(b"hello"); + let mut raw = b"HTTP/1.1 200 OK\r\nContent-Encoding: gzip\r\nTransfer-Encoding: chunked\r\n\r\n" + .to_vec(); + raw.extend_from_slice(format!("{:x}\r\n", compressed.len()).as_bytes()); + raw.extend_from_slice(&compressed); + raw.extend_from_slice(b"\r\n0\r\nX-Trace: trailer\r\n\r\n"); + let mut cursor = Cursor::new(raw); + let url = url::Url::parse("http://localhost/gzip").unwrap(); + let mut reader = ConnectionReader::new(&url, &mut cursor, false); + let mut response = reader.streaming_response().unwrap(); + let mut body = Vec::new(); + response.body_mut().read_to_end(&mut body).unwrap(); + + assert_eq!(b"hello", body.as_slice()); + assert_eq!( + Some("trailer"), + response.trailer_value("x-trace").map(String::as_str) + ); + assert!(response + .headers() + .unwrap() + .iter() + .all(|header| !header.name().eq_ignore_ascii_case("Content-Encoding"))); +} + +#[test] +fn test_streaming_stacked_gzip_deflate_decodes_body() { + let compressed = zlib_bytes(&gzip_bytes(b"stacked")); + let mut raw = format!( + "HTTP/1.1 200 OK\r\nContent-Encoding: gzip, deflate\r\nContent-Length: {}\r\n\r\n", + compressed.len() + ) + .into_bytes(); + raw.extend_from_slice(&compressed); + let mut cursor = Cursor::new(raw); + let url = url::Url::parse("http://localhost/stack").unwrap(); + let mut reader = ConnectionReader::new(&url, &mut cursor, false); + let mut response = reader.streaming_response().unwrap(); + let mut body = Vec::new(); + response.body_mut().read_to_end(&mut body).unwrap(); + + assert_eq!(b"stacked", body.as_slice()); + assert!(response + .headers() + .unwrap() + .iter() + .all(|header| !header.name().eq_ignore_ascii_case("Content-Encoding"))); +} + +#[test] +fn test_streaming_malformed_gzip_returns_decode_error() { + let body = b"not-gzip"; + let mut raw = format!( + "HTTP/1.1 200 OK\r\nContent-Encoding: gzip\r\nContent-Length: {}\r\n\r\n", + body.len() + ) + .into_bytes(); + raw.extend_from_slice(body); + let mut cursor = Cursor::new(raw); + let url = url::Url::parse("http://localhost/gzip").unwrap(); + let mut reader = ConnectionReader::new(&url, &mut cursor, false); + let error = reader + .streaming_response() + .unwrap() + .read_to_response() + .expect_err("malformed gzip should fail"); + assert_decode_error(error); +} + +#[test] +fn test_streaming_raw_deflate_returns_decode_error() { + let mut encoder = flate2::write::DeflateEncoder::new(Vec::new(), Compression::default()); + encoder.write_all(b"OK").unwrap(); + let body = encoder.finish().unwrap(); + let mut raw = format!( + "HTTP/1.1 200 OK\r\nContent-Encoding: deflate\r\nContent-Length: {}\r\n\r\n", + body.len() + ) + .into_bytes(); + raw.extend_from_slice(&body); + let mut cursor = Cursor::new(raw); + let url = url::Url::parse("http://localhost/deflate").unwrap(); + let mut reader = ConnectionReader::new(&url, &mut cursor, false); + let error = reader + .streaming_response() + .unwrap() + .read_to_response() + .expect_err("raw deflate should fail"); + assert_decode_error(error); +} + +#[test] +fn test_streaming_unknown_content_encoding_preserves_raw_body_and_headers() { + let body = gzip_bytes(b"OK"); + let mut raw = format!( + "HTTP/1.1 200 OK\r\nContent-Encoding: gzip, br\r\nContent-Length: {}\r\n\r\n", + body.len() + ) + .into_bytes(); + raw.extend_from_slice(&body); + let mut cursor = Cursor::new(raw); + let url = url::Url::parse("http://localhost/br").unwrap(); + let mut reader = ConnectionReader::new(&url, &mut cursor, false); + let mut response = reader.streaming_response().unwrap(); + let mut out = Vec::new(); + response.body_mut().read_to_end(&mut out).unwrap(); + + assert_eq!(body, out); + assert_eq!( + Some("gzip, br"), + response + .headers() + .unwrap() + .iter() + .find(|header| header.name().eq_ignore_ascii_case("Content-Encoding")) + .map(|header| header.value().as_str()) + ); +} + +#[test] +fn test_streaming_identity_content_encoding_preserves_raw_body_and_headers() { + let body = b"plain"; + let mut raw = format!( + "HTTP/1.1 200 OK\r\nContent-Encoding: identity\r\nContent-Length: {}\r\n\r\n", + body.len() + ) + .into_bytes(); + raw.extend_from_slice(body); + let mut cursor = Cursor::new(raw); + let url = url::Url::parse("http://localhost/identity").unwrap(); + let mut reader = ConnectionReader::new(&url, &mut cursor, false); + let mut response = reader.streaming_response().unwrap(); + let mut out = Vec::new(); + response.body_mut().read_to_end(&mut out).unwrap(); + + assert_eq!(b"plain", out.as_slice()); + assert_eq!( + Some("identity"), + response + .headers() + .unwrap() + .iter() + .find(|header| header.name().eq_ignore_ascii_case("Content-Encoding")) + .map(|header| header.value().as_str()) + ); +} + #[test] fn test_buffered_gzip_response_exposes_decoded_body_headers() { let body = gzip_bytes(b"decoded");