Implement the extended CONNECT protocol from RFC 8441

Author: nox

src/client.rs

@@ -517,6 +517,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 +1259,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.
     ///
@@ -1426,6 +1438,7 @@ impl Peer {
                 uri,
                 headers,
                 version,
+                extensions,
                 ..
             },
             _,
@@ -1435,7 +1448,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, extensions);
 
         if pseudo.scheme.is_none() {
             // If the scheme is not set, then there are a two options.

src/ext.rs

@@ -0,0 +1,63 @@
+//! Extensions specific to the HTTP/2.0 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 PseudoProtocol {
+    value: BytesStr,
+}
+
+impl PseudoProtocol {
+    /// Creates a new pseudo-protocol.
+    ///
+    /// TODO: Validation.
+    pub const fn from_static(value: &'static str) -> Self {
+        Self {
+            value: BytesStr::from_static(value),
+        }
+    }
+
+    pub(crate) 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 PseudoProtocol {
+    fn from(value: &'a str) -> Self {
+        Self {
+            value: BytesStr::from(value),
+        }
+    }
+}
+
+impl std::ops::Deref for PseudoProtocol {
+    type Target = str;
+    fn deref(&self) -> &str {
+        self.as_str()
+    }
+}
+
+impl AsRef<[u8]> for PseudoProtocol {
+    fn as_ref(&self) -> &[u8] {
+        self.value.as_ref()
+    }
+}
+
+impl fmt::Debug for PseudoProtocol {
+    fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
+        self.value.fmt(f)
+    }
+}

src/frame/headers.rs

@@ -1,9 +1,10 @@
 use super::{util, StreamDependency, StreamId};
+use crate::ext::PseudoProtocol;
 use crate::frame::{Error, Frame, Head, Kind};
 use crate::hpack::{self, BytesStr};
 
 use http::header::{self, HeaderName, HeaderValue};
-use http::{uri, HeaderMap, Method, Request, StatusCode, Uri};
+use http::{uri, Extensions, HeaderMap, Method, Request, StatusCode, Uri};
 
 use bytes::{BufMut, Bytes, BytesMut};
 
@@ -66,6 +67,7 @@ pub struct Pseudo {
     pub scheme: Option<BytesStr>,
     pub authority: Option<BytesStr>,
     pub path: Option<BytesStr>,
+    pub protocol: Option<PseudoProtocol>,
 
     // Response
     pub status: Option<StatusCode>,
@@ -525,14 +527,16 @@ impl Continuation {
 // ===== impl Pseudo =====
 
 impl Pseudo {
-    pub fn request(method: Method, uri: Uri) -> Self {
+    pub fn request(method: Method, uri: Uri, mut extensions: Extensions) -> Self {
         let parts = uri::Parts::from(uri);
 
         let mut path = parts
             .path_and_query
             .map(|v| BytesStr::from(v.as_str()))
             .unwrap_or(BytesStr::from_static(""));
 
+        let protocol = extensions.remove::<PseudoProtocol>();
+
         match method {
             Method::OPTIONS | Method::CONNECT => {}
             _ if path.is_empty() => {
@@ -546,6 +550,7 @@ impl Pseudo {
             scheme: None,
             authority: None,
             path: Some(path).filter(|p| !p.is_empty()),
+            protocol,
             status: None,
         };
 
@@ -571,6 +576,7 @@ impl Pseudo {
             scheme: None,
             authority: None,
             path: None,
+            protocol: None,
             status: Some(status),
         }
     }
@@ -589,6 +595,11 @@ impl Pseudo {
         self.scheme = Some(bytes_str);
     }
 
+    #[cfg(feature = "unstable")]
+    pub fn set_protocol(&mut self, protocol: PseudoProtocol) {
+        self.protocol = Some(protocol);
+    }
+
     pub fn set_authority(&mut self, authority: BytesStr) {
         self.authority = Some(authority);
     }
@@ -677,6 +688,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 +890,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
     }
@@ -149,7 +159,7 @@ impl Settings {
         debug_assert!(!settings.flags.is_ack());
 
         for raw in payload.chunks(6) {
-            match Setting::load(raw) {
+            match dbg!(Setting::load(raw)) {
                 Some(HeaderTableSize(val)) => {
                     settings.header_table_size = Some(val);
                 }
@@ -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 => {}
             }
         }
@@ -206,7 +224,7 @@ impl Settings {
         // Encode the settings
         self.for_each(|setting| {
             tracing::trace!("encoding setting; val={:?}", setting);
-            setting.encode(dst)
+            dbg!(setting).encode(dst)
         });
     }
 
@@ -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()
@@ -284,13 +309,14 @@ impl Setting {
     pub fn from_id(id: u16, val: u32) -> Option<Setting> {
         use self::Setting::*;
 
-        match id {
+        match dbg!(id) {
             1 => Some(HeaderTableSize(val)),
             2 => Some(EnablePush(val)),
             3 => Some(MaxConcurrentStreams(val)),
             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);