From 2977f304f5c6523ccc7eb4c56869dbdf7807550e Mon Sep 17 00:00:00 2001 From: Kun Lai Date: Wed, 20 Aug 2025 17:00:04 +0000 Subject: [PATCH 1/7] refact(bhttp): move framing indicator codec to a dedicated struct Signed-off-by: Kun Lai --- bhttp/src/lib.rs | 70 +++++++++++++++++++++++++++++++++++++++--------- 1 file changed, 58 insertions(+), 12 deletions(-) diff --git a/bhttp/src/lib.rs b/bhttp/src/lib.rs index caa6b37..08004ee 100644 --- a/bhttp/src/lib.rs +++ b/bhttp/src/lib.rs @@ -30,6 +30,54 @@ const TRANSFER_ENCODING: &[u8] = b"transfer-encoding"; const CHUNKED: &[u8] = b"chunked"; /// An HTTP status code. +#[derive(Clone, Copy, Debug)] +pub enum FramingIndicator { + Request(Mode), + Response(Mode), +} + +impl FramingIndicator { + pub fn is_request(&self) -> bool { + matches!(self, Self::Request(_)) + } + + pub fn mode(&self) -> Mode { + match self { + Self::Request(mode) => *mode, + Self::Response(mode) => *mode, + } + } + + pub fn write_bhttp(&self, w: &mut impl io::Write) -> Res<()> { + let code: u64 = match self { + Self::Request(Mode::KnownLength) => 0, + Self::Response(Mode::KnownLength) => 1, + Self::Request(Mode::IndeterminateLength) => 2, + Self::Response(Mode::IndeterminateLength) => 3, + }; + + write_varint(code, w)?; + Ok(()) + } + + pub fn read_bhttp(r: &mut T) -> Res + where + T: BorrowMut + ?Sized, + R: ReadSeek + ?Sized, + { + let t = read_varint(r)?.ok_or(Error::Truncated)?; + let request = t == 0 || t == 2; + let mode = Mode::try_from(t)?; + + let framing_indicator = if request { + FramingIndicator::Request(mode) + } else { + FramingIndicator::Response(mode) + }; + Ok(framing_indicator) + } +} + #[derive(Clone, Copy, Debug)] pub struct StatusCode(u16); @@ -552,12 +600,10 @@ impl ControlData { } #[must_use] - fn code(&self, mode: Mode) -> u64 { - match (self, mode) { - (Self::Request { .. }, Mode::KnownLength) => 0, - (Self::Response(_), Mode::KnownLength) => 1, - (Self::Request { .. }, Mode::IndeterminateLength) => 2, - (Self::Response(_), Mode::IndeterminateLength) => 3, + fn framing_indicator(&self, mode: Mode) -> FramingIndicator { + match self { + Self::Request { .. } => FramingIndicator::Request(mode), + Self::Response { .. } => FramingIndicator::Response(mode), } } @@ -957,16 +1003,16 @@ impl Message { T: BorrowMut + ?Sized, R: ReadSeek + ?Sized, { - let t = read_varint(r)?.ok_or(Error::Truncated)?; - let request = t == 0 || t == 2; - let mode = Mode::try_from(t)?; + let framing_indicator = FramingIndicator::read_bhttp(r)?; + let is_request = framing_indicator.is_request(); + let mode = framing_indicator.mode(); - let mut control = ControlData::read_bhttp(request, r)?; + let mut control = ControlData::read_bhttp(is_request, r)?; let mut informational = Vec::new(); while let Some(status) = control.informational() { let fields = FieldSection::read_bhttp(mode, r)?; informational.push(InformationalResponse::new(status, fields)); - control = ControlData::read_bhttp(request, r)?; + control = ControlData::read_bhttp(is_request, r)?; } let hfields = FieldSection::read_bhttp(mode, r)?; @@ -993,7 +1039,7 @@ impl Message { /// Write a BHTTP message. pub fn write_bhttp(&self, mode: Mode, w: &mut impl io::Write) -> Res<()> { - write_varint(self.header.control.code(mode), w)?; + self.header.control.framing_indicator(mode).write_bhttp(w)?; for info in &self.informational { info.write_bhttp(mode, w)?; } From 4accd17862a976d786598d97c6855ef77ba9628b Mon Sep 17 00:00:00 2001 From: Kun Lai Date: Wed, 20 Aug 2025 17:05:01 +0000 Subject: [PATCH 2/7] feat(http-compat): add support for convert between bhttp binary and http crate Signed-off-by: Kun Lai --- bhttp/Cargo.toml | 9 + bhttp/src/err.rs | 14 + bhttp/src/http_compat/decode.rs | 512 ++++++++++++++++++++++++++++++++ bhttp/src/http_compat/encode.rs | 443 +++++++++++++++++++++++++++ bhttp/src/http_compat/mod.rs | 13 + bhttp/src/lib.rs | 3 + 6 files changed, 994 insertions(+) create mode 100644 bhttp/src/http_compat/decode.rs create mode 100644 bhttp/src/http_compat/encode.rs create mode 100644 bhttp/src/http_compat/mod.rs diff --git a/bhttp/Cargo.toml b/bhttp/Cargo.toml index 214b116..664a502 100644 --- a/bhttp/Cargo.toml +++ b/bhttp/Cargo.toml @@ -16,15 +16,24 @@ readme.workspace = true default = [] http = ["dep:url"] stream = ["dep:futures", "dep:pin-project"] +http-compat = ["http", "stream", "dep:http", "dep:futures", "dep:http-body", "dep:futures-core", "dep:bytes", "dep:tokio-util", "tokio-util/compat", "tokio-util/io", "dep:tokio", "dep:async-stream"] [dependencies] futures = {version = "0.3", optional = true} pin-project = {version = "1.1", optional = true} thiserror = "2" url = {version = "2", optional = true} +bytes = { version = "1.0", optional = true } +futures-core = { version = "0.3.24", optional = true } +http = { version = "1.0", optional = true } +http-body = { version = "1.0", optional = true } +tokio-util = { version = "0.7.16", optional = true } +tokio = { version = "1.47.1", optional = true, default-features = false } +async-stream = { version = "0.3.6", optional = true } [dev-dependencies] hex = "0.4" +http-body-util = "0.1" [dev-dependencies.sync-async] path= "../sync-async" diff --git a/bhttp/src/err.rs b/bhttp/src/err.rs index 45d6fab..5a01a1b 100644 --- a/bhttp/src/err.rs +++ b/bhttp/src/err.rs @@ -4,6 +4,8 @@ pub enum Error { ConnectUnsupported, #[error("a field contained invalid Unicode: {0}")] CharacterEncoding(#[from] std::string::FromUtf8Error), + #[error("a field contained invalid Unicode: {0}")] + CharacterEncoding2(#[from] std::str::Utf8Error), #[error("read a response when expecting a request")] ExpectedRequest, #[error("read a request when expecting a response")] @@ -37,6 +39,18 @@ pub enum Error { #[error("a URL could not be parsed into components: {0}")] #[cfg(feature = "http")] UrlParse(#[from] url::ParseError), + #[error("the status code was not supported {0}")] + #[cfg(feature = "http-compat")] + UnsportedStatusCode(u16), + #[error("got an unknown HTTP body frame type, only data frame and trailer frame are supported")] + #[cfg(feature = "http-compat")] + UnknownHttpBodyFrameType, + #[error("http error")] + #[cfg(feature = "http-compat")] + HttpError(#[from] http::Error), + #[error("http header value contains invalid bytes")] + #[cfg(feature = "http-compat")] + HttpHeaderValueNotString(#[from] http::header::ToStrError), } pub type Res = Result; diff --git a/bhttp/src/http_compat/decode.rs b/bhttp/src/http_compat/decode.rs new file mode 100644 index 0000000..5a12c5a --- /dev/null +++ b/bhttp/src/http_compat/decode.rs @@ -0,0 +1,512 @@ +use std::{ + pin::Pin, + task::{Context, Poll}, +}; + +use bytes::Bytes; +use futures::AsyncRead as FuturesAsyncRead; +use futures::{stream::BoxStream, StreamExt}; +use http::{request, response, HeaderMap, Request, Response, Uri}; +use http_body::{Body as HttpBody, Frame}; +use tokio_util::{compat::FuturesAsyncReadCompatExt as _, io::ReaderStream}; + +use crate::{ + stream::{AsyncMessage, AsyncReadMessage as _}, + Error, Message, Res, +}; + +/// An enum representing either an HTTP request parts or response parts +/// +/// This is used internally during the decoding process to handle both +/// request and response types uniformly before constructing the final +/// HTTP structures. +#[derive(Debug)] +enum HttpMessageParts { + /// An HTTP request parts + Request(request::Parts), + /// An HTTP response parts + Response(response::Parts), +} + +/// An enum representing either an HTTP request or response +/// +/// This is the final result of the decoding process, containing either +/// a fully constructed HTTP request or response with a streaming body. +pub enum HttpMessage { + /// An HTTP request with a streaming BHTTP body + Request(Request>), + /// An HTTP response with a streaming BHTTP body + Response(Response>), +} + +/// A body type that wraps an [`futures::AsyncRead`] and progressively decodes BHTTP body data +/// +/// This struct implements the [`http_body::Body`] trait, providing a streaming +/// interface for reading the body data from a BHTTP message. It handles both +/// regular body data and trailers. +pub struct BhttpBody { + stream: BoxStream<'static, Result, Error>>, + _phantom: std::marker::PhantomData, +} + +impl BhttpBody { + /// Create a new BHTTP body decoder + /// + /// # Arguments + /// + /// * `async_message` - The async message reader to decode body data from + fn new(mut async_message: AsyncMessage) -> Self { + let stream = async_stream::try_stream! { + let body = async_message.body()?; + let body = body.compat(); + let mut stream = ReaderStream::new(body); + + // Read body data progressively + while let Some(frame_data) = stream.next().await { + yield Frame::data(frame_data?); + } + + // Read trailer data + let trailers = async_message.trailer().await?; + let trailers = HeaderMap::try_from(&trailers)?; + yield Frame::trailers(trailers); + }; + + Self { + stream: stream.boxed(), + _phantom: std::marker::PhantomData, + } + } +} + +impl HttpBody for BhttpBody +where + R: FuturesAsyncRead + Unpin + 'static, +{ + type Data = Bytes; + type Error = Error; + + /// Poll for the next frame of body data + /// + /// This method implements the streaming body reading process, returning + /// either data frames or trailer frames as they become available. + /// + /// # Arguments + /// + /// * `cx` - The task context for polling + /// + /// # Returns + /// + /// * `Poll::Ready(Some(Ok(Frame)))` - Next frame of body data or trailers + /// * `Poll::Ready(Some(Err(Error)))` - An error occurred during decoding + /// * `Poll::Ready(None)` - All body data has been read + /// * `Poll::Pending` - No data is ready yet, but may be in the future + fn poll_frame( + mut self: Pin<&mut Self>, + cx: &mut Context<'_>, + ) -> Poll, Self::Error>>> { + self.stream.as_mut().poll_next(cx) + } +} + +/// Asynchronous streaming decoding from BHTTP format to HTTP types. +/// +/// This type provides functionality to decode BHTTP format data into +/// `http::Request`/`http::Response` structures through asynchronous streaming. +/// Both **Known-Length Message** and **Indeterminate-Length Messages** are supported. +/// +/// This decoder works with any type that implements [`futures::AsyncRead`], +/// making it suitable for various input sources like network streams or files. +/// +/// # Important Notes +/// +/// - Due to limitations in the http crate, any Informational Response (1xx) status +/// codes in HTTP responses will be ignored during encoding. +/// - In our support for Indeterminate-Length Binary Encoding, the body is treated +/// as a continuous binary stream, which means HTTP frames might be reorganized +/// or split. Applications should implement their own framing methods and should +/// not assume that Frame encodings from the http crate will remain the same +/// after encoding and decoding. +pub struct BhttpDecoder { + reader: R, +} + +impl BhttpDecoder { + /// Create a new BHTTP decoder + /// + /// # Arguments + /// + /// * `reader` - The async reader to decode BHTTP data from + pub fn new(reader: R) -> Self { + Self { reader } + } +} + +impl BhttpDecoder +where + R: FuturesAsyncRead + Unpin + Send + 'static, +{ + /// Decode an HTTP message from BHTTP format, returning either a request or response + /// + /// This method performs the complete decoding process: + /// 1. Reads the message header + /// 2. Determines if it's a request or response + /// 3. Constructs the appropriate HTTP type with a streaming body + /// + /// # Returns + /// + /// A decode result which can be convert to either an HTTP request or response with a streaming body, + /// or an error if decoding fails. + pub async fn decode_message(self) -> Res> { + let mut async_message = Message::async_read(self.reader); + + let header = async_message.header().await?; + let parts = HttpMessageParts::try_from(header)?; + + match parts { + HttpMessageParts::Request(parts) => Ok(HttpMessage::Request(Request::from_parts( + parts, + BhttpBody::new(async_message), + ))), + HttpMessageParts::Response(parts) => Ok(HttpMessage::Response(Response::from_parts( + parts, + BhttpBody::new(async_message), + ))), + } + } +} + +impl TryFrom for HttpMessageParts { + type Error = Error; + + /// Convert a BHTTP header into either HTTP request or response parts + /// + /// This method parses the control data and field section from a BHTTP header + /// and constructs the corresponding HTTP request or response parts. + /// + /// # Arguments + /// + /// * `header` - The BHTTP header to convert + /// + /// # Returns + /// + /// Either HTTP request parts or response parts, depending on the control data, + /// or an error if parsing fails. + fn try_from(header: crate::Header) -> Res { + if header.control().is_request() { + Ok(HttpMessageParts::Request(request::Parts::try_from(header)?)) + } else { + Ok(HttpMessageParts::Response(response::Parts::try_from( + header, + )?)) + } + } +} + +impl TryFrom for request::Parts { + type Error = Error; + + fn try_from(header: crate::Header) -> Res { + let control_data = header.control(); + + // Build the HTTP request parts + let mut builder = request::Builder::new(); + + // Set method + if let Some(method) = control_data.method() { + builder = builder.method(method); + } + + // Convert field section to headers + let headers = HeaderMap::try_from(&header.fields)?; + + // Set URI + let uri = { + let mut uri_builder = Uri::builder(); + + if let Some(scheme) = control_data.scheme() { + uri_builder = uri_builder.scheme(scheme); + } + + // Handle authority field + match control_data.authority() { + Some(authority_bytes) if !authority_bytes.is_empty() => { + uri_builder = uri_builder.authority(std::str::from_utf8(authority_bytes)?); + } + _ => { + // The http::Uri does not allow an absent authority when the scheme is present, but + // this is permitted in bhttp and report an InvalidUri(InvalidFormat) error. To + // reconcile this difference, we will attempt to retrieve the authority from the + // request headers. + if control_data.scheme().is_some() { + if let Some(host_value) = headers.get("host") { + uri_builder = uri_builder.authority(host_value.to_str()?); + } + } + } + } + + if let Some(path) = control_data.path() { + uri_builder = uri_builder.path_and_query(std::str::from_utf8(path)?); + } + + uri_builder.build()? + }; + builder = builder.uri(uri); + + // Set headers + for (name, value) in headers { + if let Some(name) = name { + builder = builder.header(name, value); + } + } + + Ok(builder.body(())?.into_parts().0) + } +} + +impl TryFrom for response::Parts { + type Error = Error; + + fn try_from(header: crate::Header) -> Res { + let control_data = header.control(); + let mut builder = response::Builder::new(); + + // Set status code + if let Some(status) = control_data.status() { + builder = builder.status(status.code()); + } + + // Set headers + let headers = HeaderMap::try_from(&header.fields)?; + for (name, value) in headers { + if let Some(name) = name { + builder = builder.header(name, value); + } + } + + Ok(builder.body(())?.into_parts().0) + } +} + +#[cfg(test)] +mod tests { + use super::*; + use futures::executor::block_on; + use futures::io::Cursor; + use http_body_util::BodyExt; + + #[test] + fn decode_rfc9292_request_example_known_length() { + // Example from Section 5.1 of RFC 9292 - Known-Length Binary Encoding of Request + const REQUEST_EXAMPLE: &[u8] = &[ + 0x00, 0x03, 0x47, 0x45, 0x54, 0x05, 0x68, 0x74, 0x74, 0x70, 0x73, 0x00, 0x0a, 0x2f, + 0x68, 0x65, 0x6c, 0x6c, 0x6f, 0x2e, 0x74, 0x78, 0x74, 0x40, 0x6c, 0x0a, 0x75, 0x73, + 0x65, 0x72, 0x2d, 0x61, 0x67, 0x65, 0x6e, 0x74, 0x34, 0x63, 0x75, 0x72, 0x6c, 0x2f, + 0x37, 0x2e, 0x31, 0x36, 0x2e, 0x33, 0x20, 0x6c, 0x69, 0x62, 0x63, 0x75, 0x72, 0x6c, + 0x2f, 0x37, 0x2e, 0x31, 0x36, 0x2e, 0x33, 0x20, 0x4f, 0x70, 0x65, 0x6e, 0x53, 0x53, + 0x4c, 0x2f, 0x30, 0x2e, 0x39, 0x2e, 0x37, 0x6c, 0x20, 0x7a, 0x6c, 0x69, 0x62, 0x2f, + 0x31, 0x2e, 0x32, 0x2e, 0x33, 0x04, 0x68, 0x6f, 0x73, 0x74, 0x0f, 0x77, 0x77, 0x77, + 0x2e, 0x65, 0x78, 0x61, 0x6d, 0x70, 0x6c, 0x65, 0x2e, 0x63, 0x6f, 0x6d, 0x0f, 0x61, + 0x63, 0x63, 0x65, 0x70, 0x74, 0x2d, 0x6c, 0x61, 0x6e, 0x67, 0x75, 0x61, 0x67, 0x65, + 0x06, 0x65, 0x6e, 0x2c, 0x20, 0x6d, 0x69, 0x00, 0x00, + ]; + + let cursor = Cursor::new(REQUEST_EXAMPLE); + let decoder = BhttpDecoder::new(cursor); + let result = block_on(decoder.decode_message()).expect("Failed to decode message"); + + match result { + HttpMessage::Request(request) => { + // Check method + assert_eq!(request.method(), http::Method::GET); + + // Check URI + assert_eq!(request.uri(), "https://www.example.com/hello.txt"); + + // Check headers + let headers = request.headers(); + assert_eq!( + headers.get("user-agent").unwrap(), + "curl/7.16.3 libcurl/7.16.3 OpenSSL/0.9.7l zlib/1.2.3" + ); + assert_eq!(headers.get("host").unwrap(), "www.example.com"); + assert_eq!(headers.get("accept-language").unwrap(), "en, mi"); + + // Check body - should be empty + let body_data = block_on(request.into_body().collect()).unwrap().to_bytes(); + assert!(body_data.is_empty()); + } + HttpMessage::Response(_) => { + panic!("Expected a request, but got a response"); + } + } + } + + #[test] + fn decode_rfc9292_request_example_indeterminate_length() { + // Example from Section 5.1 of RFC 9292 - Indeterminate-Length Binary Encoding of Request + const REQUEST_EXAMPLE: &[u8] = &[ + 0x02, 0x03, 0x47, 0x45, 0x54, 0x05, 0x68, 0x74, 0x74, 0x70, 0x73, 0x00, 0x0a, 0x2f, + 0x68, 0x65, 0x6c, 0x6c, 0x6f, 0x2e, 0x74, 0x78, 0x74, 0x0a, 0x75, 0x73, 0x65, 0x72, + 0x2d, 0x61, 0x67, 0x65, 0x6e, 0x74, 0x34, 0x63, 0x75, 0x72, 0x6c, 0x2f, 0x37, 0x2e, + 0x31, 0x36, 0x2e, 0x33, 0x20, 0x6c, 0x69, 0x62, 0x63, 0x75, 0x72, 0x6c, 0x2f, 0x37, + 0x2e, 0x31, 0x36, 0x2e, 0x33, 0x20, 0x4f, 0x70, 0x65, 0x6e, 0x53, 0x53, 0x4c, 0x2f, + 0x30, 0x2e, 0x39, 0x2e, 0x37, 0x6c, 0x20, 0x7a, 0x6c, 0x69, 0x62, 0x2f, 0x31, 0x2e, + 0x32, 0x2e, 0x33, 0x04, 0x68, 0x6f, 0x73, 0x74, 0x0f, 0x77, 0x77, 0x77, 0x2e, 0x65, + 0x78, 0x61, 0x6d, 0x70, 0x6c, 0x65, 0x2e, 0x63, 0x6f, 0x6d, 0x0f, 0x61, 0x63, 0x63, + 0x65, 0x70, 0x74, 0x2d, 0x6c, 0x61, 0x6e, 0x67, 0x75, 0x61, 0x67, 0x65, 0x06, 0x65, + 0x6e, 0x2c, 0x20, 0x6d, 0x69, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, + 0x00, 0x00, 0x00, 0x00, + ]; + + let cursor = Cursor::new(REQUEST_EXAMPLE); + let decoder = BhttpDecoder::new(cursor); + let result = block_on(decoder.decode_message()).expect("Failed to decode message"); + + match result { + HttpMessage::Request(request) => { + // Check method + assert_eq!(request.method(), http::Method::GET); + + // Check URI + assert_eq!(request.uri(), "https://www.example.com/hello.txt"); + + // Check headers + let headers = request.headers(); + assert_eq!( + headers.get("user-agent").unwrap(), + "curl/7.16.3 libcurl/7.16.3 OpenSSL/0.9.7l zlib/1.2.3" + ); + assert_eq!(headers.get("host").unwrap(), "www.example.com"); + assert_eq!(headers.get("accept-language").unwrap(), "en, mi"); + + // Check body - should be empty + let body_data = block_on(request.into_body().collect()).unwrap().to_bytes(); + assert!(body_data.is_empty()); + } + HttpMessage::Response(_) => { + panic!("Expected a request, but got a response"); + } + } + } + + #[test] + fn decode_rfc9292_response_example_inculding_informational_responses() { + // Example from Section 5.2 of RFC 9292 - Response including Informational Responses + const RESPONSE_EXAMPLE: &[u8] = &[ + 0x03, 0x40, 0x66, 0x07, 0x72, 0x75, 0x6e, 0x6e, 0x69, 0x6e, 0x67, 0x0a, 0x22, 0x73, + 0x6c, 0x65, 0x65, 0x70, 0x20, 0x31, 0x35, 0x22, 0x00, 0x40, 0x67, 0x04, 0x6c, 0x69, + 0x6e, 0x6b, 0x23, 0x3c, 0x2f, 0x73, 0x74, 0x79, 0x6c, 0x65, 0x2e, 0x63, 0x73, 0x73, + 0x3e, 0x3b, 0x20, 0x72, 0x65, 0x6c, 0x3d, 0x70, 0x72, 0x65, 0x6c, 0x6f, 0x61, 0x64, + 0x3b, 0x20, 0x61, 0x73, 0x3d, 0x73, 0x74, 0x79, 0x6c, 0x65, 0x04, 0x6c, 0x69, 0x6e, + 0x6b, 0x24, 0x3c, 0x2f, 0x73, 0x63, 0x72, 0x69, 0x70, 0x74, 0x2e, 0x6a, 0x73, 0x3e, + 0x3b, 0x20, 0x72, 0x65, 0x6c, 0x3d, 0x70, 0x72, 0x65, 0x6c, 0x6f, 0x61, 0x64, 0x3b, + 0x20, 0x61, 0x73, 0x3d, 0x73, 0x63, 0x72, 0x69, 0x70, 0x74, 0x00, 0x40, 0xc8, 0x04, + 0x64, 0x61, 0x74, 0x65, 0x1d, 0x4d, 0x6f, 0x6e, 0x2c, 0x20, 0x32, 0x37, 0x20, 0x4a, + 0x75, 0x6c, 0x20, 0x32, 0x30, 0x30, 0x39, 0x20, 0x31, 0x32, 0x3a, 0x32, 0x38, 0x3a, + 0x35, 0x33, 0x20, 0x47, 0x4d, 0x54, 0x06, 0x73, 0x65, 0x72, 0x76, 0x65, 0x72, 0x06, + 0x41, 0x70, 0x61, 0x63, 0x68, 0x65, 0x0d, 0x6c, 0x61, 0x73, 0x74, 0x2d, 0x6d, 0x6f, + 0x64, 0x69, 0x66, 0x69, 0x65, 0x64, 0x1d, 0x57, 0x65, 0x64, 0x2c, 0x20, 0x32, 0x32, + 0x20, 0x4a, 0x75, 0x6c, 0x20, 0x32, 0x30, 0x30, 0x39, 0x20, 0x31, 0x39, 0x3a, 0x31, + 0x35, 0x3a, 0x35, 0x36, 0x20, 0x47, 0x4d, 0x54, 0x04, 0x65, 0x74, 0x61, 0x67, 0x14, + 0x22, 0x33, 0x34, 0x61, 0x61, 0x33, 0x38, 0x37, 0x2d, 0x64, 0x2d, 0x31, 0x35, 0x36, + 0x38, 0x65, 0x62, 0x30, 0x30, 0x22, 0x0d, 0x61, 0x63, 0x63, 0x65, 0x70, 0x74, 0x2d, + 0x72, 0x61, 0x6e, 0x67, 0x65, 0x73, 0x05, 0x62, 0x79, 0x74, 0x65, 0x73, 0x0e, 0x63, + 0x6f, 0x6e, 0x74, 0x65, 0x6e, 0x74, 0x2d, 0x6c, 0x65, 0x6e, 0x67, 0x74, 0x68, 0x02, + 0x35, 0x31, 0x04, 0x76, 0x61, 0x72, 0x79, 0x0f, 0x41, 0x63, 0x63, 0x65, 0x70, 0x74, + 0x2d, 0x45, 0x6e, 0x63, 0x6f, 0x64, 0x69, 0x6e, 0x67, 0x0c, 0x63, 0x6f, 0x6e, 0x74, + 0x65, 0x6e, 0x74, 0x2d, 0x74, 0x79, 0x70, 0x65, 0x0a, 0x74, 0x65, 0x78, 0x74, 0x2f, + 0x70, 0x6c, 0x61, 0x69, 0x6e, 0x00, 0x33, 0x48, 0x65, 0x6c, 0x6c, 0x6f, 0x20, 0x57, + 0x6f, 0x72, 0x6c, 0x64, 0x21, 0x20, 0x4d, 0x79, 0x20, 0x63, 0x6f, 0x6e, 0x74, 0x65, + 0x6e, 0x74, 0x20, 0x69, 0x6e, 0x63, 0x6c, 0x75, 0x64, 0x65, 0x73, 0x20, 0x61, 0x20, + 0x74, 0x72, 0x61, 0x69, 0x6c, 0x69, 0x6e, 0x67, 0x20, 0x43, 0x52, 0x4c, 0x46, 0x2e, + 0x0d, 0x0a, 0x00, 0x00, + ]; + + let cursor = Cursor::new(RESPONSE_EXAMPLE); + let decoder = BhttpDecoder::new(cursor); + let result = block_on(decoder.decode_message()).expect("Failed to decode message"); + + match result { + HttpMessage::Response(response) => { + // Check status + assert_eq!(response.status(), http::StatusCode::OK); + + // Check headers + let headers = response.headers(); + assert_eq!(headers.len(), 8); // Note that informational status codes is dropped during decoding + assert_eq!( + headers.get("date").unwrap(), + "Mon, 27 Jul 2009 12:28:53 GMT" + ); + assert_eq!(headers.get("server").unwrap(), "Apache"); + assert_eq!( + headers.get("last-modified").unwrap(), + "Wed, 22 Jul 2009 19:15:56 GMT" + ); + assert_eq!(headers.get("etag").unwrap(), "\"34aa387-d-1568eb00\""); + assert_eq!(headers.get("accept-ranges").unwrap(), "bytes"); + assert_eq!(headers.get("content-length").unwrap(), "51"); + assert_eq!(headers.get("vary").unwrap(), "Accept-Encoding"); + assert_eq!(headers.get("content-type").unwrap(), "text/plain"); + + // Check body + let body_data = block_on(response.into_body().collect()).unwrap().to_bytes(); + assert_eq!( + std::str::from_utf8(&body_data).unwrap(), + "Hello World! My content includes a trailing CRLF.\r\n" + ); + } + HttpMessage::Request(_) => { + panic!("Expected a response, but got a request"); + } + } + } + + #[test] + fn decode_rfc9292_response_example_known_length_with_trailer() { + // Example from Section 5.2 of RFC 9292 - Known-Length Encoding of Response + const RESPONSE_EXAMPLE: &[u8] = &[ + 0x01, 0x40, 0xc8, 0x00, 0x1d, 0x54, 0x68, 0x69, 0x73, 0x20, 0x63, 0x6f, 0x6e, 0x74, + 0x65, 0x6e, 0x74, 0x20, 0x63, 0x6f, 0x6e, 0x74, 0x61, 0x69, 0x6e, 0x73, 0x20, 0x43, + 0x52, 0x4c, 0x46, 0x2e, 0x0d, 0x0a, 0x0d, 0x07, 0x74, 0x72, 0x61, 0x69, 0x6c, 0x65, + 0x72, 0x04, 0x74, 0x65, 0x78, 0x74, + ]; + + let cursor = Cursor::new(RESPONSE_EXAMPLE); + let decoder = BhttpDecoder::new(cursor); + let result = block_on(decoder.decode_message()).expect("Failed to decode message"); + + match result { + HttpMessage::Response(response) => { + // Check status + assert_eq!(response.status(), http::StatusCode::OK); + + // Check headers + let headers = response.headers(); + assert_eq!(headers.len(), 0); + + // Read all body + let collected = block_on(response.into_body().collect()).unwrap(); + + // Check trailers + let trailers = collected.trailers(); + assert!(trailers.is_some()); + let trailers = trailers.unwrap(); + assert_eq!(trailers.len(), 1); + assert_eq!(trailers.get("trailer").unwrap(), "text"); + + // Check body + let body_data = collected.to_bytes(); + assert_eq!( + std::str::from_utf8(&body_data).unwrap(), + "This content contains CRLF.\r\n" + ); + } + HttpMessage::Request(_) => { + panic!("Expected a response, but got a request"); + } + } + } +} diff --git a/bhttp/src/http_compat/encode.rs b/bhttp/src/http_compat/encode.rs new file mode 100644 index 0000000..cce2d35 --- /dev/null +++ b/bhttp/src/http_compat/encode.rs @@ -0,0 +1,443 @@ +use std::{ + pin::Pin, + task::{Context, Poll}, +}; + +use crate::{ + rw::{write_len, write_vec}, + ControlData, Error, Field, FieldSection, Mode, Res, StatusCode, +}; +use bytes::Bytes; +use futures_core::stream::Stream as FuturesStream; +use http::{HeaderMap, Request, Response}; +use http_body::Body as HttpBody; + +/// Helper function to build a [`FieldSection`] from HTTP headers +impl TryFrom<&HeaderMap> for FieldSection { + type Error = Error; + + fn try_from(headers: &HeaderMap) -> Res { + let fields = headers + .iter() + .map(|(name, value)| { + Field::new(name.as_str().as_bytes().to_vec(), value.as_bytes().to_vec()) + }) + .collect::>(); + + Ok(FieldSection(fields)) + } +} + +impl TryFrom<&FieldSection> for HeaderMap { + type Error = Error; + + fn try_from(field_section: &FieldSection) -> Res { + let headers = field_section + .iter() + .map(|field| { + Ok(( + http::header::HeaderName::from_bytes(field.name()) + .map_err(http::Error::from)?, + http::header::HeaderValue::from_bytes(field.value()) + .map_err(http::Error::from)?, + )) + }) + .collect::>>()?; + + Ok(headers) + } +} + +/// Enum to wrap either an HTTP Request or Response for unified BHTTP encoding +#[derive(Debug)] +enum HttpMessage { + Request(Request), + Response(Response), +} + +impl HttpMessage { + /// Generate BHTTP headers field section from the wrapped request or response + fn bhttp_headers(&self) -> Res { + let http_headers = match self { + Self::Request(req) => req.headers(), + Self::Response(res) => res.headers(), + }; + + FieldSection::try_from(http_headers) + } + + /// Generate control data from the wrapped request or response for BHTTP encoding + fn bhttp_control_data(&self) -> Res { + match self { + Self::Request(req) => { + let method = req.method().as_str().as_bytes().to_vec(); + let uri = req.uri(); + + let scheme = uri.scheme_str().unwrap_or("https").as_bytes().to_vec(); + let authority = uri + .authority() + .map(|a| a.as_str().as_bytes().to_vec()) + .unwrap_or_default(); + let path = uri + .path_and_query() + .map_or_else(|| b"/".to_vec(), |p| p.as_str().as_bytes().to_vec()); + + Ok(ControlData::Request { + method, + scheme, + authority, + path, + }) + } + Self::Response(res) => { + let status_code = StatusCode::try_from(res.status().as_u16()) + .map_err(|_| Error::InvalidStatus)?; + Ok(ControlData::Response(status_code)) + } + } + } + + /// Get a mutable reference to the body of the wrapped request or response + fn body_mut(&mut self) -> &mut T { + match self { + Self::Request(req) => req.body_mut(), + Self::Response(res) => res.body_mut(), + } + } +} + +/// This type provides functionality to encode `http::Request`/`http::Response` +/// structures into BHTTP format through asynchronous streaming. +/// +/// The encoder works with any type that implements [`http_body::Body`], allowing +/// for efficient streaming encoding of HTTP data without needing to load the entire +/// message into memory. This struct implements the [`futures_core::stream::Stream`] +/// trait, producing chunks of BHTTP-encoded binary data. It supports streaming +/// bodies and trailers. +/// +/// # Important Notes +/// +/// - Currently, we always use **Indeterminate-Length Messages** since we have no +/// way to know the length of the HTTP body in advance. +/// - Due to limitations in the http crate, any Informational Response (1xx) status +/// codes in HTTP responses will be ignored during encoding. +/// - In our support for Indeterminate-Length Binary Encoding, the body is treated +/// as a continuous binary stream, which means HTTP frames might be reorganized +/// or split. Applications should implement their own framing methods and should +/// not assume that Frame encodings from the http crate will remain the same +/// after encoding and decoding. +pub struct BhttpEncoder { + state: EncodeState, + message: HttpMessage, +} + +/// The internal state of the BHTTP encoder +enum EncodeState { + /// Initial state, ready to encode control data and headers + ControlAndHeaders, + /// Streaming body chunks and collecting trailers + BodyChunk { + collected_trailers: Option, + }, + /// All data has been encoded + Done, +} + +impl BhttpEncoder { + /// Create a new converter for an HTTP request + /// + /// # Arguments + /// + /// * `request` - The HTTP request to convert to BHTTP format + pub fn from_request(request: Request) -> Self { + Self { + state: EncodeState::ControlAndHeaders, + message: HttpMessage::Request(request), + } + } + + /// Create a new converter for an HTTP response + /// + /// # Arguments + /// + /// * `response` - The HTTP response to convert to BHTTP format + pub fn from_response(response: Response) -> Self { + Self { + state: EncodeState::ControlAndHeaders, + message: HttpMessage::Response(response), + } + } +} + +impl FuturesStream for BhttpEncoder +where + T: HttpBody + Unpin, + T::Error: std::error::Error + Send + Sync + 'static, +{ + type Item = Res>; + + /// Poll for the next chunk of BHTTP-encoded data + /// + /// This method implements the streaming encoding process: + /// 1. First, it encodes the control data and headers + /// 2. Then, it streams the body data in chunks + /// 3. Finally, it handles trailers if present + /// + /// For those who needs a [`futures::prelude::AsyncRead`] or + /// [`futures::stream::IntoAsyncRead`], see + /// [`into_async_read`](futures::stream::TryStreamExt::into_async_read). + /// + /// # Arguments + /// + /// * `cx` - The task context for polling + /// + /// # Returns + /// + /// * `Poll::Ready(Some(Ok(Vec)))` - Next chunk of BHTTP data + /// * `Poll::Ready(Some(Err(Error)))` - An error occurred during encoding + /// * `Poll::Ready(None)` - All data has been encoded + /// * `Poll::Pending` - No data is ready yet, but may be in the future + fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll> { + let this = self.get_mut(); + + match &mut this.state { + EncodeState::ControlAndHeaders => { + let mut buf = Vec::new(); + + let control_data = this.message.bhttp_control_data()?; + + // 1. Write the framing indicator. + control_data + .framing_indicator(Mode::IndeterminateLength) + .write_bhttp(&mut buf)?; + + // 2. Response with with informational code (100 - 199) is not supported so we just skip encoding Informational Response. + if let Some(code) = control_data.status() { + if code.informational() { + return Poll::Ready(Some(Err(Error::UnsportedStatusCode(code.code())))); + } + } + + // 3. Write (request / the final response) control data + control_data.write_bhttp(&mut buf)?; + + // 4. Write headers field section + let field_section = this.message.bhttp_headers()?; + field_section.write_bhttp(Mode::IndeterminateLength, &mut buf)?; + + // 5. Change to next state + this.state = EncodeState::BodyChunk { + collected_trailers: None, + }; + Poll::Ready(Some(Ok(buf))) + } + EncodeState::BodyChunk { collected_trailers } => { + // Poll the body for more data + let body = this.message.body_mut(); + match Pin::new(body).poll_frame(cx) { + Poll::Ready(Some(Ok(frame))) => { + match frame.into_data() { + Ok(data) => { + // Is a data frame + let mut chunk_data = Vec::new(); + write_vec(&data, &mut chunk_data)?; + Poll::Ready(Some(Ok(chunk_data))) + } + Err(frame) => { + if let Ok(trailers) = frame.into_trailers() { + // Is a trailers frame + if let Some(current) = collected_trailers { + current.extend(trailers); + } else { + *collected_trailers = Some(trailers); + } + Poll::Pending + } else { + // For unknown frame types, report an invalid data error. + this.state = EncodeState::Done; + Poll::Ready(Some(Err(Error::UnknownHttpBodyFrameType))) + } + } + } + } + Poll::Ready(Some(Err(e))) => Poll::Ready(Some(Err(Error::Io( + std::io::Error::new(std::io::ErrorKind::Other, e), + )))), + Poll::Ready(None) => { + // End of body, write zero-length chunk to indicate end, since we always using Mode::IndeterminateLength + let mut buf = Vec::new(); + write_len(0, &mut buf)?; + + // Then, write the trailer field section with indeterminate length mode + let field_section = + if let Some(collected_trailers) = collected_trailers.as_ref() { + FieldSection::try_from(collected_trailers)? + } else { + FieldSection(vec![]) + }; + field_section.write_bhttp(Mode::IndeterminateLength, &mut buf)?; + + // Switch to Done state since we have written the trailers + this.state = EncodeState::Done; + + Poll::Ready(Some(Ok(buf))) + } + Poll::Pending => Poll::Pending, + } + } + EncodeState::Done => Poll::Ready(None), + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + use bytes::Bytes; + use futures::TryStreamExt as _; + use http::{Request, Response}; + use http_body::Frame; + use http_body_util::Full; + use http_body_util::StreamBody; + use sync_async::SyncRead as _; + + #[test] + fn request_to_bhttp_conversion() { + let request = Request::builder() + .method("GET") + .uri("https://example.com/path") + .header("User-Agent", "test-agent") + .header("Accept", "application/json") + .body(Full::new(Bytes::from("test body"))) + .unwrap(); + + let mut reader = BhttpEncoder::from_request(request) + .map_err(std::io::Error::other) + .into_async_read(); + let bhttp_data = reader.sync_read_to_end(); + + // Verify that we got some data + assert!(!bhttp_data.is_empty()); + + // The data should be longer than just the body since it includes headers and metadata + assert!(bhttp_data.len() > 9); + } + + #[test] + fn response_to_bhttp_conversion() { + let response = Response::builder() + .status(200) + .header("Content-Type", "text/plain") + .header("Server", "test-server") + .body(Full::new(Bytes::from("test response body"))) + .unwrap(); + + let mut reader = BhttpEncoder::from_response(response) + .map_err(std::io::Error::other) + .into_async_read(); + + let bhttp_data = reader.sync_read_to_end(); + + // Verify that we got some data + assert!(!bhttp_data.is_empty()); + + // The data should be longer than just the body since it includes headers and metadata + assert!(bhttp_data.len() > 18); + } + + #[test] + fn trailers_conversion() { + // Create a body with trailers + let data_frame = Frame::data(Bytes::from("test body")); + + // Create trailers + let mut trailers = HeaderMap::new(); + trailers.insert("Trailer-Key", "Trailer-Value".parse().unwrap()); + + let trailer_frame = Frame::trailers(trailers); + + let body = StreamBody::new(futures::stream::iter([ + Ok::<_, std::convert::Infallible>(data_frame), + Ok::<_, std::convert::Infallible>(trailer_frame), + ])); + + let response = Response::builder() + .status(200) + .header("Content-Type", "text/plain") + .body(body) + .unwrap(); + + let mut reader = BhttpEncoder::from_response(response) + .map_err(std::io::Error::other) + .into_async_read(); + + let bhttp_data = reader.sync_read_to_end(); + + // Verify that we got some data + assert!(!bhttp_data.is_empty()); + } + + #[test] + fn encode_rfc9292_request_example_indeterminate_length() { + // This example is modified from Section 5.1 of RFC 9292 - Indeterminate-Length Binary Encoding of Request + // + // In the original bhttp example, the scheme is https, the authority is empty, and the path + // is `/hello.txt``. However, this is not allowed by the http crate, which requires that if + // a scheme is present, the authority must also be present. To make the test meaningful, the + // bhttp example is modified, to ensure it conforms to the HTTP URI specification: + // - set the authority to `"example.com"` + const REQUEST_EXAMPLE: &[u8] = &[ + 0x02, 0x03, 0x47, 0x45, 0x54, 0x05, 0x68, 0x74, 0x74, 0x70, 0x73, 0x0f, 0x77, 0x77, + 0x77, 0x2e, 0x65, 0x78, 0x61, 0x6d, 0x70, 0x6c, 0x65, 0x2e, 0x63, 0x6f, 0x6d, 0x0a, + 0x2f, 0x68, 0x65, 0x6c, 0x6c, 0x6f, 0x2e, 0x74, 0x78, 0x74, 0x0a, 0x75, 0x73, 0x65, + 0x72, 0x2d, 0x61, 0x67, 0x65, 0x6e, 0x74, 0x34, 0x63, 0x75, 0x72, 0x6c, 0x2f, 0x37, + 0x2e, 0x31, 0x36, 0x2e, 0x33, 0x20, 0x6c, 0x69, 0x62, 0x63, 0x75, 0x72, 0x6c, 0x2f, + 0x37, 0x2e, 0x31, 0x36, 0x2e, 0x33, 0x20, 0x4f, 0x70, 0x65, 0x6e, 0x53, 0x53, 0x4c, + 0x2f, 0x30, 0x2e, 0x39, 0x2e, 0x37, 0x6c, 0x20, 0x7a, 0x6c, 0x69, 0x62, 0x2f, 0x31, + 0x2e, 0x32, 0x2e, 0x33, 0x04, 0x68, 0x6f, 0x73, 0x74, 0x0f, 0x77, 0x77, 0x77, 0x2e, + 0x65, 0x78, 0x61, 0x6d, 0x70, 0x6c, 0x65, 0x2e, 0x63, 0x6f, 0x6d, 0x0f, 0x61, 0x63, + 0x63, 0x65, 0x70, 0x74, 0x2d, 0x6c, 0x61, 0x6e, 0x67, 0x75, 0x61, 0x67, 0x65, 0x06, + 0x65, 0x6e, 0x2c, 0x20, 0x6d, 0x69, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, + 0x00, 0x00, 0x00, 0x00, 0x00, + ]; + + // Create the HTTP request as specified in the RFC + let request = Request::builder() + .method("GET") + .uri("https://www.example.com/hello.txt") + .header( + "user-agent", + "curl/7.16.3 libcurl/7.16.3 OpenSSL/0.9.7l zlib/1.2.3", + ) + .header("host", "www.example.com") + .header("accept-language", "en, mi") + .body(Full::new(Bytes::new())) // No body for this request + .unwrap(); + + // Encode the request to BHTTP using our streaming encoder + let mut reader = BhttpEncoder::from_request(request) + .map_err(std::io::Error::other) + .into_async_read(); + + let bhttp_data = reader.sync_read_to_end(); + + compare_encoded_data_with_the_expected_example(&bhttp_data, REQUEST_EXAMPLE); + } + + fn compare_encoded_data_with_the_expected_example(generated: &[u8], expected: &[u8]) { + // Compare the encoded data with the expected example + // BHTTP allows trailing padding, so we need to check if our data matches + // the expected data, ignoring any trailing zeros in the expected data + + // Find where the actual data ends (ignoring trailing zeros) + let expected_end = expected.iter().rposition(|&x| x != 0).map_or(0, |i| i + 1); + + let generated_end = generated.iter().rposition(|&x| x != 0).map_or(0, |i| i + 1); + + // Check that the first part matches exactly + assert_eq!( + hex::encode(&generated[..generated_end]), + hex::encode(&expected[..expected_end]) + ); + } +} diff --git a/bhttp/src/http_compat/mod.rs b/bhttp/src/http_compat/mod.rs new file mode 100644 index 0000000..0b516cb --- /dev/null +++ b/bhttp/src/http_compat/mod.rs @@ -0,0 +1,13 @@ +//! Asynchronous streaming conversion between HTTP types and BHTTP format. +//! +//! This module provides asynchronous streaming encoders and decoders for converting between +//! `http::Request`/`http::Response` and BHTTP format using streams. +//! +//! The encoding functionality allows you to convert HTTP requests and responses into +//! BHTTP binary format through an asynchronous stream. The decoding functionality allows +//! you to convert BHTTP binary data back into HTTP requests or responses. +//! +//! Both encoding and decoding operations are performed asynchronously, making them +//! suitable for use in async contexts such as with the axum web framework. +pub mod decode; +pub mod encode; diff --git a/bhttp/src/lib.rs b/bhttp/src/lib.rs index 08004ee..df6b1f2 100644 --- a/bhttp/src/lib.rs +++ b/bhttp/src/lib.rs @@ -16,6 +16,9 @@ mod rw; #[cfg(feature = "stream")] pub mod stream; +#[cfg(feature = "http-compat")] +pub mod http_compat; + pub use err::Error; use err::Res; #[cfg(feature = "http")] From 46443f1d0471cab55b4b56f02a53d9f642878875 Mon Sep 17 00:00:00 2001 From: Kun Lai Date: Sat, 6 Sep 2025 15:51:32 +0000 Subject: [PATCH 3/7] clippy: add deny for clippy::redundant_test_prefix and make clippy happy Signed-off-by: Kun Lai --- bhttp/src/lib.rs | 7 ++++--- ohttp/src/rh/aead.rs | 2 ++ 2 files changed, 6 insertions(+), 3 deletions(-) diff --git a/bhttp/src/lib.rs b/bhttp/src/lib.rs index df6b1f2..6eb5138 100644 --- a/bhttp/src/lib.rs +++ b/bhttp/src/lib.rs @@ -1,4 +1,4 @@ -#![deny(warnings, clippy::pedantic)] +#![deny(warnings, clippy::pedantic, clippy::redundant_test_prefix)] #![allow(clippy::missing_errors_doc)] // Too lazy to document these. use std::{ @@ -40,14 +40,15 @@ pub enum FramingIndicator { } impl FramingIndicator { + #[must_use] pub fn is_request(&self) -> bool { matches!(self, Self::Request(_)) } + #[must_use] pub fn mode(&self) -> Mode { match self { - Self::Request(mode) => *mode, - Self::Response(mode) => *mode, + Self::Request(mode) | Self::Response(mode) => *mode, } } diff --git a/ohttp/src/rh/aead.rs b/ohttp/src/rh/aead.rs index 62b6bbb..8185a7e 100644 --- a/ohttp/src/rh/aead.rs +++ b/ohttp/src/rh/aead.rs @@ -121,6 +121,7 @@ impl Decrypt for Aead { res } + #[allow(unused)] fn alg(&self) -> AeadId { self.algorithm } @@ -136,6 +137,7 @@ impl Encrypt for Aead { Ok(ct) } + #[allow(unused)] fn alg(&self) -> AeadId { self.algorithm } From 107993e20b1583f7074a872016034d80da1b9383 Mon Sep 17 00:00:00 2001 From: Kun Lai Date: Wed, 20 Aug 2025 16:52:17 +0000 Subject: [PATCH 4/7] test(http-compat): remove padding bytes from bhttp example in encode.rs Signed-off-by: Kun Lai --- bhttp/src/http_compat/encode.rs | 30 +++++++----------------------- 1 file changed, 7 insertions(+), 23 deletions(-) diff --git a/bhttp/src/http_compat/encode.rs b/bhttp/src/http_compat/encode.rs index cce2d35..80fb70c 100644 --- a/bhttp/src/http_compat/encode.rs +++ b/bhttp/src/http_compat/encode.rs @@ -383,9 +383,11 @@ mod tests { // // In the original bhttp example, the scheme is https, the authority is empty, and the path // is `/hello.txt``. However, this is not allowed by the http crate, which requires that if - // a scheme is present, the authority must also be present. To make the test meaningful, the - // bhttp example is modified, to ensure it conforms to the HTTP URI specification: - // - set the authority to `"example.com"` + // a scheme is present, the authority must also be present. + // + // The bhttp example has been modified as follows: + // - set the authority to `"example.com"`, to ensure it conforms to the HTTP URI specification + // - removed 10 bytes of padding, as described in Section 5.1 of RFC 9292 const REQUEST_EXAMPLE: &[u8] = &[ 0x02, 0x03, 0x47, 0x45, 0x54, 0x05, 0x68, 0x74, 0x74, 0x70, 0x73, 0x0f, 0x77, 0x77, 0x77, 0x2e, 0x65, 0x78, 0x61, 0x6d, 0x70, 0x6c, 0x65, 0x2e, 0x63, 0x6f, 0x6d, 0x0a, @@ -397,8 +399,7 @@ mod tests { 0x2e, 0x32, 0x2e, 0x33, 0x04, 0x68, 0x6f, 0x73, 0x74, 0x0f, 0x77, 0x77, 0x77, 0x2e, 0x65, 0x78, 0x61, 0x6d, 0x70, 0x6c, 0x65, 0x2e, 0x63, 0x6f, 0x6d, 0x0f, 0x61, 0x63, 0x63, 0x65, 0x70, 0x74, 0x2d, 0x6c, 0x61, 0x6e, 0x67, 0x75, 0x61, 0x67, 0x65, 0x06, - 0x65, 0x6e, 0x2c, 0x20, 0x6d, 0x69, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, - 0x00, 0x00, 0x00, 0x00, 0x00, + 0x65, 0x6e, 0x2c, 0x20, 0x6d, 0x69, 0x00, 0x00, 0x00, ]; // Create the HTTP request as specified in the RFC @@ -421,23 +422,6 @@ mod tests { let bhttp_data = reader.sync_read_to_end(); - compare_encoded_data_with_the_expected_example(&bhttp_data, REQUEST_EXAMPLE); - } - - fn compare_encoded_data_with_the_expected_example(generated: &[u8], expected: &[u8]) { - // Compare the encoded data with the expected example - // BHTTP allows trailing padding, so we need to check if our data matches - // the expected data, ignoring any trailing zeros in the expected data - - // Find where the actual data ends (ignoring trailing zeros) - let expected_end = expected.iter().rposition(|&x| x != 0).map_or(0, |i| i + 1); - - let generated_end = generated.iter().rposition(|&x| x != 0).map_or(0, |i| i + 1); - - // Check that the first part matches exactly - assert_eq!( - hex::encode(&generated[..generated_end]), - hex::encode(&expected[..expected_end]) - ); + assert_eq!(hex::encode(&bhttp_data), hex::encode(&REQUEST_EXAMPLE)); } } From 0106d0d456462ccc9d83603fe2c0afc213c0194d Mon Sep 17 00:00:00 2001 From: Kun Lai Date: Sat, 6 Sep 2025 15:37:00 +0000 Subject: [PATCH 5/7] feat(http-compat): allow creating server response while client request is consuming Signed-off-by: Kun Lai --- bhttp/src/http_compat/decode.rs | 36 ++++++++++---- bhttp/src/stream/mod.rs | 4 ++ ohttp/src/lib.rs | 4 +- ohttp/src/stream.rs | 84 +++++++++++++++++++++++++++++++++ 4 files changed, 118 insertions(+), 10 deletions(-) diff --git a/bhttp/src/http_compat/decode.rs b/bhttp/src/http_compat/decode.rs index 5a12c5a..bb96bf1 100644 --- a/bhttp/src/http_compat/decode.rs +++ b/bhttp/src/http_compat/decode.rs @@ -157,20 +157,40 @@ where /// /// A decode result which can be convert to either an HTTP request or response with a streaming body, /// or an error if decoding fails. - pub async fn decode_message(self) -> Res> { + pub async fn decode_message(self) -> Res> { let mut async_message = Message::async_read(self.reader); let header = async_message.header().await?; let parts = HttpMessageParts::try_from(header)?; - match parts { + Ok(DecodeResult { + async_message, + parts, + }) + } +} + +pub struct DecodeResult { + async_message: AsyncMessage, + parts: HttpMessageParts, +} + +impl DecodeResult { + pub fn reader_ref(&self) -> &R { + self.async_message.reader_ref() + } +} + +impl DecodeResult { + pub fn into_full_message(self) -> Res> { + match self.parts { HttpMessageParts::Request(parts) => Ok(HttpMessage::Request(Request::from_parts( parts, - BhttpBody::new(async_message), + BhttpBody::new(self.async_message), ))), HttpMessageParts::Response(parts) => Ok(HttpMessage::Response(Response::from_parts( parts, - BhttpBody::new(async_message), + BhttpBody::new(self.async_message), ))), } } @@ -316,7 +336,7 @@ mod tests { let decoder = BhttpDecoder::new(cursor); let result = block_on(decoder.decode_message()).expect("Failed to decode message"); - match result { + match result.into_full_message().unwrap() { HttpMessage::Request(request) => { // Check method assert_eq!(request.method(), http::Method::GET); @@ -364,7 +384,7 @@ mod tests { let decoder = BhttpDecoder::new(cursor); let result = block_on(decoder.decode_message()).expect("Failed to decode message"); - match result { + match result.into_full_message().unwrap() { HttpMessage::Request(request) => { // Check method assert_eq!(request.method(), http::Method::GET); @@ -428,7 +448,7 @@ mod tests { let decoder = BhttpDecoder::new(cursor); let result = block_on(decoder.decode_message()).expect("Failed to decode message"); - match result { + match result.into_full_message().unwrap() { HttpMessage::Response(response) => { // Check status assert_eq!(response.status(), http::StatusCode::OK); @@ -478,7 +498,7 @@ mod tests { let decoder = BhttpDecoder::new(cursor); let result = block_on(decoder.decode_message()).expect("Failed to decode message"); - match result { + match result.into_full_message().unwrap() { HttpMessage::Response(response) => { // Check status assert_eq!(response.status(), http::StatusCode::OK); diff --git a/bhttp/src/stream/mod.rs b/bhttp/src/stream/mod.rs index bfba2c1..a018388 100644 --- a/bhttp/src/stream/mod.rs +++ b/bhttp/src/stream/mod.rs @@ -157,6 +157,10 @@ pub struct AsyncMessage { unsafe impl Send for AsyncMessage {} impl AsyncMessage { + pub fn reader_ref(&self) -> &S { + &self.src + } + async fn next_info(&mut self) -> Res> { let request = if matches!(self.state, AsyncMessageState::Init) { // Read control data ... diff --git a/ohttp/src/lib.rs b/ohttp/src/lib.rs index f0a7ef2..9fad975 100644 --- a/ohttp/src/lib.rs +++ b/ohttp/src/lib.rs @@ -219,8 +219,8 @@ impl Server { /// Remove encapsulation on a streamed request. #[cfg(feature = "stream")] - pub fn decapsulate_stream(self, src: S) -> ServerRequestStream { - ServerRequestStream::new(self.config, src) + pub fn decapsulate_stream(&self, src: S) -> ServerRequestStream { + ServerRequestStream::new(self.config.clone(), src) } } diff --git a/ohttp/src/stream.rs b/ohttp/src/stream.rs index 0c402ac..3d027b6 100644 --- a/ohttp/src/stream.rs +++ b/ohttp/src/stream.rs @@ -246,6 +246,46 @@ impl ClientRequest { }, }) } + + /// Get an decapsulator that can be used to process the response. + /// + /// While this can be used while sending the request, + /// doing so creates a risk of revealing unwanted information to the gateway. + /// That includes the round trip time between client and gateway, + /// which might reveal information about the location of the client. + pub fn response_decapsulator(&self) -> Res { + let enc = self.writer.cipher.enc()?; + let secret = export_secret( + &self.writer.cipher, + LABEL_RESPONSE, + self.writer.cipher.config(), + )?; + Ok(ClientResponseDecapsulator { + config: self.writer.cipher.config(), + state: ClientResponseState::Header { + enc, + secret, + nonce: [0; 16], + read: 0, + }, + }) + } +} + +pub struct ClientResponseDecapsulator { + config: HpkeConfig, + state: ClientResponseState, +} + +impl ClientResponseDecapsulator { + /// Get an object that can be used to process the response. + pub fn decapsulate_response(self, src: S) -> Res> { + Ok(ClientResponse { + src, + config: self.config, + state: self.state, + }) + } } impl AsyncWrite for ClientRequest { @@ -620,6 +660,50 @@ impl ServerRequest { }, }) } + + /// Get a response encapsulator that creates a response to the given request. + /// + /// This fails with an error if the request header hasn't been processed. + /// This condition is not exposed through a future anywhere, + /// but you can wait for the first byte of data. + pub fn response_encapsulator(&self) -> Res { + let ServerRequestState::Body { hpke, state: _ } = &self.state else { + return Err(Error::NotReady); + }; + + let response_nonce = random(entropy(hpke.config())); + let aead = make_aead( + Mode::Encrypt, + hpke.config(), + &export_secret(hpke, LABEL_RESPONSE, hpke.config())?, + &self.enc, + &response_nonce, + )?; + + Ok(ServerResponseEncapsulator { + aead, + response_nonce, + }) + } +} + +pub struct ServerResponseEncapsulator { + aead: Aead, + response_nonce: Vec, +} + +impl ServerResponseEncapsulator { + /// Get a response that wraps the given async write instance. + pub fn encapsulate_response(self, dst: D) -> Res> { + Ok(ServerResponse { + writer: ChunkWriter { + dst, + cipher: self.aead, + buf: self.response_nonce, + closed: false, + }, + }) + } } impl ServerRequest { From 68907110a706fd19d4ac07b78192130408f9c169 Mon Sep 17 00:00:00 2001 From: Kun Lai Date: Sat, 6 Sep 2025 17:22:31 +0000 Subject: [PATCH 6/7] fix(ohttp): change nonce array length from 16 to 32 to fix panic Signed-off-by: Kun Lai --- ohttp/src/stream.rs | 7 ++++--- 1 file changed, 4 insertions(+), 3 deletions(-) diff --git a/ohttp/src/stream.rs b/ohttp/src/stream.rs index 3d027b6..7c651f0 100644 --- a/ohttp/src/stream.rs +++ b/ohttp/src/stream.rs @@ -28,6 +28,7 @@ pub(crate) const LABEL_RESPONSE: &[u8] = b"message/bhttp chunked response"; const MAX_CHUNK_PLAINTEXT: usize = 1 << 14; const CHUNK_AAD: &[u8] = b""; const FINAL_CHUNK_AAD: &[u8] = b"final"; +const MAX_ENTROPY_LEN: usize = 32; #[allow(clippy::unnecessary_wraps)] fn ioerror(e: E) -> Poll> @@ -241,7 +242,7 @@ impl ClientRequest { state: ClientResponseState::Header { enc, secret, - nonce: [0; 16], + nonce: Default::default(), read: 0, }, }) @@ -265,7 +266,7 @@ impl ClientRequest { state: ClientResponseState::Header { enc, secret, - nonce: [0; 16], + nonce: Default::default(), read: 0, }, }) @@ -824,7 +825,7 @@ enum ClientResponseState { Header { enc: Vec, secret: SymKey, - nonce: [u8; 16], + nonce: [u8; MAX_ENTROPY_LEN], read: usize, }, Body { From 214b757e894f9c6f58c23d39a7f1aad3b98c82ab Mon Sep 17 00:00:00 2001 From: Kun Lai Date: Sun, 7 Sep 2025 18:07:23 +0000 Subject: [PATCH 7/7] fix(http-compat): remove default value for uri when convert between bhttp and http Signed-off-by: Kun Lai --- bhttp/src/http_compat/decode.rs | 16 +++++++++++----- bhttp/src/http_compat/encode.rs | 8 ++++++-- 2 files changed, 17 insertions(+), 7 deletions(-) diff --git a/bhttp/src/http_compat/decode.rs b/bhttp/src/http_compat/decode.rs index bb96bf1..63eea40 100644 --- a/bhttp/src/http_compat/decode.rs +++ b/bhttp/src/http_compat/decode.rs @@ -245,7 +245,9 @@ impl TryFrom for request::Parts { let mut uri_builder = Uri::builder(); if let Some(scheme) = control_data.scheme() { - uri_builder = uri_builder.scheme(scheme); + if !scheme.is_empty() { + uri_builder = uri_builder.scheme(scheme); + } } // Handle authority field @@ -258,16 +260,20 @@ impl TryFrom for request::Parts { // this is permitted in bhttp and report an InvalidUri(InvalidFormat) error. To // reconcile this difference, we will attempt to retrieve the authority from the // request headers. - if control_data.scheme().is_some() { - if let Some(host_value) = headers.get("host") { - uri_builder = uri_builder.authority(host_value.to_str()?); + if let Some(scheme) = control_data.scheme() { + if !scheme.is_empty() { + if let Some(host_value) = headers.get("host") { + uri_builder = uri_builder.authority(host_value.to_str()?); + } } } } } if let Some(path) = control_data.path() { - uri_builder = uri_builder.path_and_query(std::str::from_utf8(path)?); + if !path.is_empty() { + uri_builder = uri_builder.path_and_query(std::str::from_utf8(path)?); + } } uri_builder.build()? diff --git a/bhttp/src/http_compat/encode.rs b/bhttp/src/http_compat/encode.rs index 80fb70c..fa8fe76 100644 --- a/bhttp/src/http_compat/encode.rs +++ b/bhttp/src/http_compat/encode.rs @@ -73,14 +73,18 @@ impl HttpMessage { let method = req.method().as_str().as_bytes().to_vec(); let uri = req.uri(); - let scheme = uri.scheme_str().unwrap_or("https").as_bytes().to_vec(); + let scheme = uri + .scheme_str() + .map(|scheme| scheme.as_bytes().to_vec()) + .unwrap_or_default(); let authority = uri .authority() .map(|a| a.as_str().as_bytes().to_vec()) .unwrap_or_default(); let path = uri .path_and_query() - .map_or_else(|| b"/".to_vec(), |p| p.as_str().as_bytes().to_vec()); + .map(|p| p.as_str().as_bytes().to_vec()) + .unwrap_or_default(); Ok(ControlData::Request { method,