CHANGELOG.md

@@ -1,3 +1,9 @@
+# 0.3.8 (December 8, 2021)
+
+* Add "extended CONNECT support". Adds `h2::ext::Protocol`, which is used for request and response extensions to connect new protocols over an HTTP/2 stream.
+* Add `max_send_buffer_size` options to client and server builders, and a default of ~400MB. This acts like a high-water mark for the `poll_capacity()` method.
+* Fix panic if receiving malformed HEADERS with stream ID of 0.
+
 # 0.3.7 (October 22, 2021)
 
 * Fix panic if server sends a malformed frame on a stream client was about to open.

Cargo.toml

@@ -5,7 +5,7 @@ name = "h2"
 #   - html_root_url.
 # - Update CHANGELOG.md.
 # - Create git tag
-version = "0.3.7"
+version = "0.3.8"
 license = "MIT"
 authors = [
   "Carl Lerche <me@carllerche.com>",

src/client.rs

@@ -136,6 +136,7 @@
 //! [`Error`]: ../struct.Error.html
 
 use crate::codec::{Codec, SendError, UserError};
+use crate::ext::Protocol;
 use crate::frame::{Headers, Pseudo, Reason, Settings, StreamId};
 use crate::proto::{self, Error};
 use crate::{FlowControl, PingPong, RecvStream, SendStream};
@@ -319,6 +320,9 @@ pub struct Builder {
     /// Initial target window size for new connections.
     initial_target_connection_window_size: Option<u32>,
 
+    /// Maximum amount of bytes to "buffer" for writing per stream.
+    max_send_buffer_size: usize,
+
     /// Maximum number of locally reset streams to keep at a time.
     reset_stream_max: usize,
 
@@ -517,6 +521,19 @@ where
                 (response, stream)
             })
     }
+
+    /// Returns whether the [extended CONNECT protocol][1] is enabled or not.
+    ///
+    /// This setting is configured by the server peer by sending the
+    /// [`SETTINGS_ENABLE_CONNECT_PROTOCOL` parameter][2] in a `SETTINGS` frame.
+    /// This method returns the currently acknowledged value recieved from the
+    /// remote.
+    ///
+    /// [1]: https://datatracker.ietf.org/doc/html/rfc8441#section-4
+    /// [2]: https://datatracker.ietf.org/doc/html/rfc8441#section-3
+    pub fn is_extended_connect_protocol_enabled(&self) -> bool {
+        self.inner.is_extended_connect_protocol_enabled()
+    }
 }
 
 impl<B> fmt::Debug for SendRequest<B>
@@ -614,6 +631,7 @@ impl Builder {
     /// ```
     pub fn new() -> Builder {
         Builder {
+            max_send_buffer_size: proto::DEFAULT_MAX_SEND_BUFFER_SIZE,
             reset_stream_duration: Duration::from_secs(proto::DEFAULT_RESET_STREAM_SECS),
             reset_stream_max: proto::DEFAULT_RESET_STREAM_MAX,
             initial_target_connection_window_size: None,
@@ -948,6 +966,24 @@ impl Builder {
         self
     }
 
+    /// Sets the maximum send buffer size per stream.
+    ///
+    /// Once a stream has buffered up to (or over) the maximum, the stream's
+    /// flow control will not "poll" additional capacity. Once bytes for the
+    /// stream have been written to the connection, the send buffer capacity
+    /// will be freed up again.
+    ///
+    /// The default is currently ~400MB, but may change.
+    ///
+    /// # Panics
+    ///
+    /// This function panics if `max` is larger than `u32::MAX`.
+    pub fn max_send_buffer_size(&mut self, max: usize) -> &mut Self {
+        assert!(max <= std::u32::MAX as usize);
+        self.max_send_buffer_size = max;
+        self
+    }
+
     /// Enables or disables server push promises.
     ///
     /// This value is included in the initial SETTINGS handshake. When set, the
@@ -1170,6 +1206,7 @@ where
             proto::Config {
                 next_stream_id: builder.stream_id,
                 initial_max_send_streams: builder.initial_max_send_streams,
+                max_send_buffer_size: builder.max_send_buffer_size,
                 reset_stream_duration: builder.reset_stream_duration,
                 reset_stream_max: builder.reset_stream_max,
                 settings: builder.settings.clone(),
@@ -1246,11 +1283,10 @@ where
     /// This method returns the currently acknowledged value recieved from the
     /// remote.
     ///
-    /// [settings]: https://tools.ietf.org/html/rfc7540#section-5.1.2
+    /// [1]: https://tools.ietf.org/html/rfc7540#section-5.1.2
     pub fn max_concurrent_send_streams(&self) -> usize {
         self.inner.max_send_streams()
     }
-
     /// Returns the maximum number of concurrent streams that may be initiated
     /// by the server on this connection.
     ///
@@ -1416,6 +1452,7 @@ impl Peer {
     pub fn convert_send_message(
         id: StreamId,
         request: Request<()>,
+        protocol: Option<Protocol>,
         end_of_stream: bool,
     ) -> Result<Headers, SendError> {
         use http::request::Parts;
@@ -1435,7 +1472,7 @@ impl Peer {
 
         // Build the set pseudo header set. All requests will include `method`
         // and `path`.
-        let mut pseudo = Pseudo::request(method, uri);
+        let mut pseudo = Pseudo::request(method, uri, protocol);
 
         if pseudo.scheme.is_none() {
             // If the scheme is not set, then there are a two options.

src/ext.rs

@@ -0,0 +1,55 @@
+//! Extensions specific to the HTTP/2 protocol.
+
+use crate::hpack::BytesStr;
+
+use bytes::Bytes;
+use std::fmt;
+
+/// Represents the `:protocol` pseudo-header used by
+/// the [Extended CONNECT Protocol].
+///
+/// [Extended CONNECT Protocol]: https://datatracker.ietf.org/doc/html/rfc8441#section-4
+#[derive(Clone, Eq, PartialEq)]
+pub struct Protocol {
+    value: BytesStr,
+}
+
+impl Protocol {
+    /// Converts a static string to a protocol name.
+    pub const fn from_static(value: &'static str) -> Self {
+        Self {
+            value: BytesStr::from_static(value),
+        }
+    }
+
+    /// Returns a str representation of the header.
+    pub fn as_str(&self) -> &str {
+        self.value.as_str()
+    }
+
+    pub(crate) fn try_from(bytes: Bytes) -> Result<Self, std::str::Utf8Error> {
+        Ok(Self {
+            value: BytesStr::try_from(bytes)?,
+        })
+    }
+}
+
+impl<'a> From<&'a str> for Protocol {
+    fn from(value: &'a str) -> Self {
+        Self {
+            value: BytesStr::from(value),
+        }
+    }
+}
+
+impl AsRef<[u8]> for Protocol {
+    fn as_ref(&self) -> &[u8] {
+        self.value.as_ref()
+    }
+}
+
+impl fmt::Debug for Protocol {
+    fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
+        self.value.fmt(f)
+    }
+}

src/frame/headers.rs

@@ -1,4 +1,5 @@
 use super::{util, StreamDependency, StreamId};
+use crate::ext::Protocol;
 use crate::frame::{Error, Frame, Head, Kind};
 use crate::hpack::{self, BytesStr};
 
@@ -66,6 +67,7 @@ pub struct Pseudo {
     pub scheme: Option<BytesStr>,
     pub authority: Option<BytesStr>,
     pub path: Option<BytesStr>,
+    pub protocol: Option<Protocol>,
 
     // Response
     pub status: Option<StatusCode>,
@@ -146,6 +148,10 @@ impl Headers {
 
         tracing::trace!("loading headers; flags={:?}", flags);
 
+        if head.stream_id().is_zero() {
+            return Err(Error::InvalidStreamId);
+        }
+
         // Read the padding length
         if flags.is_padded() {
             if src.is_empty() {
@@ -288,6 +294,10 @@ impl fmt::Debug for Headers {
             .field("stream_id", &self.stream_id)
             .field("flags", &self.flags);
 
+        if let Some(ref protocol) = self.header_block.pseudo.protocol {
+            builder.field("protocol", protocol);
+        }
+
         if let Some(ref dep) = self.stream_dep {
             builder.field("stream_dep", dep);
         }
@@ -525,7 +535,7 @@ impl Continuation {
 // ===== impl Pseudo =====
 
 impl Pseudo {
-    pub fn request(method: Method, uri: Uri) -> Self {
+    pub fn request(method: Method, uri: Uri, protocol: Option<Protocol>) -> Self {
         let parts = uri::Parts::from(uri);
 
         let mut path = parts
@@ -546,6 +556,7 @@ impl Pseudo {
             scheme: None,
             authority: None,
             path: Some(path).filter(|p| !p.is_empty()),
+            protocol,
             status: None,
         };
 
@@ -571,6 +582,7 @@ impl Pseudo {
             scheme: None,
             authority: None,
             path: None,
+            protocol: None,
             status: Some(status),
         }
     }
@@ -589,6 +601,11 @@ impl Pseudo {
         self.scheme = Some(bytes_str);
     }
 
+    #[cfg(feature = "unstable")]
+    pub fn set_protocol(&mut self, protocol: Protocol) {
+        self.protocol = Some(protocol);
+    }
+
     pub fn set_authority(&mut self, authority: BytesStr) {
         self.authority = Some(authority);
     }
@@ -677,6 +694,10 @@ impl Iterator for Iter {
                 return Some(Path(path));
             }
 
+            if let Some(protocol) = pseudo.protocol.take() {
+                return Some(Protocol(protocol));
+            }
+
             if let Some(status) = pseudo.status.take() {
                 return Some(Status(status));
             }
@@ -875,6 +896,7 @@ impl HeaderBlock {
                 Method(v) => set_pseudo!(method, v),
                 Scheme(v) => set_pseudo!(scheme, v),
                 Path(v) => set_pseudo!(path, v),
+                Protocol(v) => set_pseudo!(protocol, v),
                 Status(v) => set_pseudo!(status, v),
             }
         });

src/frame/settings.rs

@@ -13,6 +13,7 @@ pub struct Settings {
     initial_window_size: Option<u32>,
     max_frame_size: Option<u32>,
     max_header_list_size: Option<u32>,
+    enable_connect_protocol: Option<u32>,
 }
 
 /// An enum that lists all valid settings that can be sent in a SETTINGS
@@ -27,6 +28,7 @@ pub enum Setting {
     InitialWindowSize(u32),
     MaxFrameSize(u32),
     MaxHeaderListSize(u32),
+    EnableConnectProtocol(u32),
 }
 
 #[derive(Copy, Clone, Eq, PartialEq, Default)]
@@ -107,6 +109,14 @@ impl Settings {
         self.enable_push = Some(enable as u32);
     }
 
+    pub fn is_extended_connect_protocol_enabled(&self) -> Option<bool> {
+        self.enable_connect_protocol.map(|val| val != 0)
+    }
+
+    pub fn set_enable_connect_protocol(&mut self, val: Option<u32>) {
+        self.enable_connect_protocol = val;
+    }
+
     pub fn header_table_size(&self) -> Option<u32> {
         self.header_table_size
     }
@@ -181,6 +191,14 @@ impl Settings {
                 Some(MaxHeaderListSize(val)) => {
                     settings.max_header_list_size = Some(val);
                 }
+                Some(EnableConnectProtocol(val)) => match val {
+                    0 | 1 => {
+                        settings.enable_connect_protocol = Some(val);
+                    }
+                    _ => {
+                        return Err(Error::InvalidSettingValue);
+                    }
+                },
                 None => {}
             }
         }
@@ -236,6 +254,10 @@ impl Settings {
         if let Some(v) = self.max_header_list_size {
             f(MaxHeaderListSize(v));
         }
+
+        if let Some(v) = self.enable_connect_protocol {
+            f(EnableConnectProtocol(v));
+        }
     }
 }
 
@@ -269,6 +291,9 @@ impl fmt::Debug for Settings {
             Setting::MaxHeaderListSize(v) => {
                 builder.field("max_header_list_size", &v);
             }
+            Setting::EnableConnectProtocol(v) => {
+                builder.field("enable_connect_protocol", &v);
+            }
         });
 
         builder.finish()
@@ -291,6 +316,7 @@ impl Setting {
             4 => Some(InitialWindowSize(val)),
             5 => Some(MaxFrameSize(val)),
             6 => Some(MaxHeaderListSize(val)),
+            8 => Some(EnableConnectProtocol(val)),
             _ => None,
         }
     }
@@ -322,6 +348,7 @@ impl Setting {
             InitialWindowSize(v) => (4, v),
             MaxFrameSize(v) => (5, v),
             MaxHeaderListSize(v) => (6, v),
+            EnableConnectProtocol(v) => (8, v),
         };
 
         dst.put_u16(kind);