Implement the extended CONNECT protocol from RFC 8441

Author: nox

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};
@@ -517,6 +518,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>
@@ -1246,11 +1260,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 +1429,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 +1449,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>,
@@ -288,6 +290,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 +531,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 +552,7 @@ impl Pseudo {
             scheme: None,
             authority: None,
             path: Some(path).filter(|p| !p.is_empty()),
+            protocol,
             status: None,
         };
 
@@ -571,6 +578,7 @@ impl Pseudo {
             scheme: None,
             authority: None,
             path: None,
+            protocol: None,
             status: Some(status),
         }
     }
@@ -589,6 +597,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 +690,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 +892,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);

src/hpack/header.rs

@@ -1,4 +1,5 @@
 use super::{DecoderError, NeedMore};
+use crate::ext::Protocol;
 
 use bytes::Bytes;
 use http::header::{HeaderName, HeaderValue};
@@ -14,6 +15,7 @@ pub enum Header<T = HeaderName> {
     Method(Method),
     Scheme(BytesStr),
     Path(BytesStr),
+    Protocol(Protocol),
     Status(StatusCode),
 }
 
@@ -25,6 +27,7 @@ pub enum Name<'a> {
     Method,
     Scheme,
     Path,
+    Protocol,
     Status,
 }
 
@@ -51,6 +54,7 @@ impl Header<Option<HeaderName>> {
             Method(v) => Method(v),
             Scheme(v) => Scheme(v),
             Path(v) => Path(v),
+            Protocol(v) => Protocol(v),
             Status(v) => Status(v),
         })
     }
@@ -79,6 +83,10 @@ impl Header {
                     let value = BytesStr::try_from(value)?;
                     Ok(Header::Path(value))
                 }
+                b"protocol" => {
+                    let value = Protocol::try_from(value)?;
+                    Ok(Header::Protocol(value))
+                }
                 b"status" => {
                     let status = StatusCode::from_bytes(&value)?;
                     Ok(Header::Status(status))
@@ -104,6 +112,7 @@ impl Header {
             Header::Method(ref v) => 32 + 7 + v.as_ref().len(),
             Header::Scheme(ref v) => 32 + 7 + v.len(),
             Header::Path(ref v) => 32 + 5 + v.len(),
+            Header::Protocol(ref v) => 32 + 9 + v.as_str().len(),
             Header::Status(_) => 32 + 7 + 3,
         }
     }
@@ -116,6 +125,7 @@ impl Header {
             Header::Method(..) => Name::Method,
             Header::Scheme(..) => Name::Scheme,
             Header::Path(..) => Name::Path,
+            Header::Protocol(..) => Name::Protocol,
             Header::Status(..) => Name::Status,
         }
     }
@@ -127,6 +137,7 @@ impl Header {
             Header::Method(ref v) => v.as_ref().as_ref(),
             Header::Scheme(ref v) => v.as_ref(),
             Header::Path(ref v) => v.as_ref(),
+            Header::Protocol(ref v) => v.as_ref(),
             Header::Status(ref v) => v.as_str().as_ref(),
         }
     }
@@ -156,6 +167,10 @@ impl Header {
                 Header::Path(ref b) => a == b,
                 _ => false,
             },
+            Header::Protocol(ref a) => match *other {
+                Header::Protocol(ref b) => a == b,
+                _ => false,
+            },
             Header::Status(ref a) => match *other {
                 Header::Status(ref b) => a == b,
                 _ => false,
@@ -205,6 +220,7 @@ impl From<Header> for Header<Option<HeaderName>> {
             Header::Method(v) => Header::Method(v),
             Header::Scheme(v) => Header::Scheme(v),
             Header::Path(v) => Header::Path(v),
+            Header::Protocol(v) => Header::Protocol(v),
             Header::Status(v) => Header::Status(v),
         }
     }
@@ -221,6 +237,7 @@ impl<'a> Name<'a> {
             Name::Method => Ok(Header::Method(Method::from_bytes(&*value)?)),
             Name::Scheme => Ok(Header::Scheme(BytesStr::try_from(value)?)),
             Name::Path => Ok(Header::Path(BytesStr::try_from(value)?)),
+            Name::Protocol => Ok(Header::Protocol(Protocol::try_from(value)?)),
             Name::Status => {
                 match StatusCode::from_bytes(&value) {
                     Ok(status) => Ok(Header::Status(status)),
@@ -238,6 +255,7 @@ impl<'a> Name<'a> {
             Name::Method => b":method",
             Name::Scheme => b":scheme",
             Name::Path => b":path",
+            Name::Protocol => b":protocol",
             Name::Status => b":status",
         }
     }

src/hpack/table.rs

@@ -751,6 +751,7 @@ fn index_static(header: &Header) -> Option<(usize, bool)> {
             "/index.html" => Some((5, true)),
             _ => Some((4, false)),
         },
+        Header::Protocol(..) => None,
         Header::Status(ref v) => match u16::from(*v) {
             200 => Some((8, true)),
             204 => Some((9, true)),