diff --git a/src/client.rs b/src/client.rs index 9d0ff31d9..594b21993 100644 --- a/src/client.rs +++ b/src/client.rs @@ -139,7 +139,7 @@ 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}; +use crate::{FlowControl, PeerSettings, PingPong, RecvStream, SendStream}; use bytes::{Buf, Bytes}; use http::{uri, HeaderMap, Method, Request, Response, Version}; @@ -540,6 +540,42 @@ where }) } + /// Polls for the `SETTINGS` parameters the server peer sent. + /// + /// A client is permitted to send requests without waiting for the peer's + /// `SETTINGS` ([RFC 9113 Section 3.4][1]), so a fresh connection has not + /// seen them. This returns `Pending` until that frame has been received + /// and acknowledged, and errors if the connection ends first. + /// + /// [1]: https://datatracker.ietf.org/doc/html/rfc9113#section-3.4 + pub fn poll_peer_settings(&self, cx: &Context) -> Poll> { + self.inner.poll_peer_settings(cx) + } + + /// Waits for the `SETTINGS` parameters the server peer sent. + /// + /// The `async` form of [`poll_peer_settings`]. + /// + /// # Examples + /// + /// ``` + /// # use bytes::Bytes; + /// # use h2::client::SendRequest; + /// # async fn doc(client: SendRequest) -> Result<(), h2::Error> { + /// let settings = client.peer_settings().await?; + /// + /// if settings.get(0x4d44) == Some(1) { + /// // the peer advertised whatever 0x4d44 means to your extension + /// } + /// # Ok(()) + /// # } + /// ``` + /// + /// [`poll_peer_settings`]: SendRequest::poll_peer_settings + pub async fn peer_settings(&self) -> Result { + crate::poll_fn(|cx| self.poll_peer_settings(cx)).await + } + /// Returns whether the [extended CONNECT protocol][1] is enabled or not. /// /// This setting is configured by the server peer by sending the @@ -707,6 +743,63 @@ impl Builder { self } + /// Sets the `SETTINGS` parameter `id` to `value`. + /// + /// [RFC 9113 Section 5.5][1] permits an extension to define new settings + /// and makes using one safe without prior arrangement: implementations + /// "MUST ignore unknown or unsupported values in all extensible protocol + /// elements". So an identifier h2 knows nothing about is sent unexamined, + /// and is safe to send to a peer that knows nothing about it either. + /// Identifiers are allocated in the IANA ["HTTP/2 Settings" registry][reg]; + /// h2 carries the number and does not define what it means. + /// + /// An identifier h2 does define is routed to the setting it names and + /// checked the same way, so `setting(4, n)` is another spelling of + /// [`initial_window_size`](Builder::initial_window_size) rather than a + /// second copy of it, and one identifier can never be sent twice in one + /// frame. What the peer sent is read with [`SendRequest::peer_settings`]. + /// + /// # Examples + /// + /// ``` + /// # use tokio::io::{AsyncRead, AsyncWrite}; + /// # use h2::client::*; + /// # use bytes::Bytes; + /// # + /// # async fn doc(my_io: T) + /// # -> Result<((SendRequest, Connection)), h2::Error> + /// # { + /// // An extension names its own identifiers; h2 only carries them. + /// const ENABLE_METADATA: u16 = 0x4d44; + /// + /// let client_fut = Builder::new() + /// .setting(ENABLE_METADATA, 1) + /// .handshake(my_io); + /// # client_fut.await + /// # } + /// # pub fn main() {} + /// ``` + /// + /// # Panics + /// + /// Panics on a value the method named for that setting would panic on: + /// `MAX_FRAME_SIZE` (`5`) outside the range + /// [`Builder::max_frame_size`] accepts. + /// + /// Panics if twelve identifiers h2 does not itself define are already set + /// and this is not one of them, since a setting a caller asked for must not + /// be quietly left unsent. The twelve are what a `SETTINGS` frame has to + /// give here once room is reserved for every identifier h2 defines, so + /// those can never panic. + /// + /// [1]: https://datatracker.ietf.org/doc/html/rfc9113#section-5.5 + /// [reg]: https://www.iana.org/assignments/http2-parameters/http2-parameters.xhtml + /// [`SendRequest::peer_settings`]: SendRequest::peer_settings + pub fn setting(&mut self, id: u16, value: u32) -> &mut Self { + self.settings.set(id, Some(value)); + self + } + /// Indicates the initial window size (in octets) for connection-level flow control /// for received data. /// diff --git a/src/ext.rs b/src/ext.rs index cf383a495..0c1a0d9b6 100644 --- a/src/ext.rs +++ b/src/ext.rs @@ -1,4 +1,37 @@ //! Extensions specific to the HTTP/2 protocol. +//! +//! [RFC 9113 Section 5.5][ext] permits an extension to define new frame types, +//! new settings, or new error codes, and obliges an endpoint to ignore what it +//! does not understand, which is what makes an extension point usable +//! "without prior arrangement or negotiation". h2 implements no extension. What +//! it offers is the means to carry one, and this module is only part of that: +//! +//! - **`SETTINGS` parameters.** Identifiers come from the IANA ["HTTP/2 +//! Settings" registry][reg]; h2 carries the number without defining it. +//! [`client::Builder::setting`] and +//! [`server::Builder::setting`] send one by identifier; +//! [`SendRequest::peer_settings`] and [`server::Connection::peer_settings`] +//! read back what the peer sent. h2 applies no rule to an identifier it does +//! not define, and rejects no value of one. +//! - **Error codes.** [`Reason`] is a `u32` either way, so a code this crate +//! does not name is carried verbatim. +//! - **Frame types.** A frame whose type h2 does not know is discarded, as 5.5 +//! requires. Sending one is not supported. +//! +//! [`Protocol`] below is a different thing: not an HTTP/2 extension point, but +//! the `:protocol` pseudo-header of [RFC 8441][8441], the extension that lets +//! a CONNECT name another protocol to carry on its stream. h2 checks the name +//! is UTF-8 and does not otherwise interpret it, so a protocol h2 has never +//! heard of works without a change here. +//! +//! [ext]: https://datatracker.ietf.org/doc/html/rfc9113#section-5.5 +//! [reg]: https://www.iana.org/assignments/http2-parameters/http2-parameters.xhtml +//! [8441]: https://datatracker.ietf.org/doc/html/rfc8441#section-4 +//! [`client::Builder::setting`]: crate::client::Builder::setting +//! [`server::Builder::setting`]: crate::server::Builder::setting +//! [`SendRequest::peer_settings`]: crate::client::SendRequest::peer_settings +//! [`server::Connection::peer_settings`]: crate::server::Connection::peer_settings +//! [`Reason`]: crate::Reason use crate::hpack::BytesStr; diff --git a/src/frame/mod.rs b/src/frame/mod.rs index 4a0c03036..0323421cb 100644 --- a/src/frame/mod.rs +++ b/src/frame/mod.rs @@ -59,6 +59,7 @@ pub use self::priority::{Priority, StreamDependency}; pub use self::reason::Reason; pub use self::reset::Reset; pub use self::settings::Settings; +pub(crate) use self::settings::MAX_SETTINGS; pub use self::stream_id::{StreamId, StreamIdOverflow}; pub use self::window_update::WindowUpdate; diff --git a/src/frame/settings.rs b/src/frame/settings.rs index 484498a9d..ceaa6cd41 100644 --- a/src/frame/settings.rs +++ b/src/frame/settings.rs @@ -3,39 +3,43 @@ use std::fmt; use crate::frame::{util, Error, Frame, FrameSize, Head, Kind, StreamId}; use bytes::{BufMut, BytesMut}; +/// How many settings one frame holds, the numerically lowest being kept. +/// +/// Eight covers every identifier this crate defines; twenty leaves a dozen for +/// extensions. 6.5.2 permits ignoring what is not understood, and what is +/// dropped always sorts above the twenty lowest, never one this crate acts on. +pub(crate) const MAX_SETTINGS: usize = 20; + +/// One slot per identifier this crate defines, so extensions a caller adds can +/// never crowd out a setting [`Settings::set`] must hold. What is left for +/// [`Settings::set`] to give is twelve. +const RESERVED: usize = SETTINGS_ENABLE_CONNECT_PROTOCOL as usize; + #[derive(Clone, Default, Eq, PartialEq)] pub struct Settings { flags: SettingsFlags, - // Fields - header_table_size: Option, - enable_push: Option, - max_concurrent_streams: Option, - initial_window_size: Option, - max_frame_size: Option, - max_header_list_size: Option, - enable_connect_protocol: Option, + len: u8, + /// The identifiers held, ascending; `ids[..len]` is live. + /// + /// Ordered by identifier, not by the order a caller set them in, so a + /// frame's bytes are a function of the values it carries. + ids: [u16; MAX_SETTINGS], + vals: [u32; MAX_SETTINGS], } -/// An enum that lists all valid settings that can be sent in a SETTINGS -/// frame. +/// The flags a SETTINGS frame can carry: 6.5 defines exactly one. /// -/// Each setting has a value that is a 32 bit unsigned integer (6.5.1.). -#[derive(Debug)] -pub enum Setting { - HeaderTableSize(u32), - EnablePush(u32), - MaxConcurrentStreams(u32), - InitialWindowSize(u32), - MaxFrameSize(u32), - MaxHeaderListSize(u32), - EnableConnectProtocol(u32), +/// Two states rather than eight bits, which leaves 254 unused bit patterns for +/// `Option` to put its `None` in, so the pending frame a +/// connection holds stays the size of a frame. +#[derive(Clone, Copy, Default, Eq, PartialEq)] +enum SettingsFlags { + #[default] + Empty, + Ack, } -#[derive(Copy, Clone, Eq, PartialEq, Default)] -pub struct SettingsFlags(u8); - const ACK: u8 = 0x1; -const ALL: u8 = ACK; /// The default value of SETTINGS_HEADER_TABLE_SIZE pub const DEFAULT_SETTINGS_HEADER_TABLE_SIZE: usize = 4_096; @@ -52,12 +56,22 @@ pub const MAX_INITIAL_WINDOW_SIZE: usize = (1 << 31) - 1; /// MAX_FRAME_SIZE upper bound pub const MAX_MAX_FRAME_SIZE: FrameSize = (1 << 24) - 1; +// The identifiers this crate defines, from RFC 9113 6.5.2 and RFC 8441 3, so +// that each number is written once. +const SETTINGS_HEADER_TABLE_SIZE: u16 = 0x1; +const SETTINGS_ENABLE_PUSH: u16 = 0x2; +const SETTINGS_MAX_CONCURRENT_STREAMS: u16 = 0x3; +const SETTINGS_INITIAL_WINDOW_SIZE: u16 = 0x4; +const SETTINGS_MAX_FRAME_SIZE: u16 = 0x5; +const SETTINGS_MAX_HEADER_LIST_SIZE: u16 = 0x6; +const SETTINGS_ENABLE_CONNECT_PROTOCOL: u16 = 0x8; + // ===== impl Settings ===== impl Settings { pub fn ack() -> Settings { Settings { - flags: SettingsFlags::ack(), + flags: SettingsFlags::Ack, ..Settings::default() } } @@ -67,67 +81,177 @@ impl Settings { } pub fn initial_window_size(&self) -> Option { - self.initial_window_size + self.get(SETTINGS_INITIAL_WINDOW_SIZE) } pub fn set_initial_window_size(&mut self, size: Option) { - self.initial_window_size = size; + self.set(SETTINGS_INITIAL_WINDOW_SIZE, size) } pub fn max_concurrent_streams(&self) -> Option { - self.max_concurrent_streams + self.get(SETTINGS_MAX_CONCURRENT_STREAMS) } pub fn set_max_concurrent_streams(&mut self, max: Option) { - self.max_concurrent_streams = max; + self.set(SETTINGS_MAX_CONCURRENT_STREAMS, max) } pub fn max_frame_size(&self) -> Option { - self.max_frame_size + self.get(SETTINGS_MAX_FRAME_SIZE) } pub fn set_max_frame_size(&mut self, size: Option) { - if let Some(val) = size { - assert!(DEFAULT_MAX_FRAME_SIZE <= val && val <= MAX_MAX_FRAME_SIZE); - } - self.max_frame_size = size; + self.set(SETTINGS_MAX_FRAME_SIZE, size) } pub fn max_header_list_size(&self) -> Option { - self.max_header_list_size + self.get(SETTINGS_MAX_HEADER_LIST_SIZE) } pub fn set_max_header_list_size(&mut self, size: Option) { - self.max_header_list_size = size; + self.set(SETTINGS_MAX_HEADER_LIST_SIZE, size) } pub fn is_push_enabled(&self) -> Option { - self.enable_push.map(|val| val != 0) + self.get(SETTINGS_ENABLE_PUSH).map(|val| val != 0) } pub fn set_enable_push(&mut self, enable: bool) { - self.enable_push = Some(enable as u32); + self.set(SETTINGS_ENABLE_PUSH, Some(enable as u32)) } pub fn is_extended_connect_protocol_enabled(&self) -> Option { - self.enable_connect_protocol.map(|val| val != 0) + self.get(SETTINGS_ENABLE_CONNECT_PROTOCOL) + .map(|val| val != 0) } pub fn set_enable_connect_protocol(&mut self, val: Option) { - self.enable_connect_protocol = val; + self.set(SETTINGS_ENABLE_CONNECT_PROTOCOL, val) } pub fn header_table_size(&self) -> Option { - self.header_table_size + self.get(SETTINGS_HEADER_TABLE_SIZE) } pub fn set_header_table_size(&mut self, size: Option) { - self.header_table_size = size; + self.set(SETTINGS_HEADER_TABLE_SIZE, size) } - pub fn load(head: Head, payload: &[u8]) -> Result { - use self::Setting::*; + pub fn get(&self, id: u16) -> Option { + self.position(id).ok().map(|i| self.vals[i]) + } + + /// How many identifiers held are not ones this crate defines. + fn extensions(&self) -> usize { + self.ids[..self.len as usize] + .iter() + .filter(|id| **id > SETTINGS_ENABLE_CONNECT_PROTOCOL) + .count() + } + + /// + /// # Panics + /// + /// On a `MAX_FRAME_SIZE` value outside the range specified by HTTP/2 + /// ยง6.5.2. With debug assertions on, also on an identifier this crate does + /// not define when twelve such are already set and `id` is not among them; + /// without them that setting is dropped and logged. Nothing already set is + /// displaced either way. + /// + /// Peer SETTINGS go through [`Settings::load`], which validates fallibly + /// and never panics. + pub fn set(&mut self, id: u16, val: Option) { + let Some(val) = val else { + self.remove(id); + return; + }; + + bounds(id, val); + + let full = id > SETTINGS_ENABLE_CONNECT_PROTOCOL + && self.position(id).is_err() + && self.extensions() >= MAX_SETTINGS - RESERVED; + debug_assert!( + !full, + "a SETTINGS frame carries {} extension identifiers here, so \ + 0x{id:04x} would not be sent", + MAX_SETTINGS - RESERVED, + ); + if full { + tracing::debug!("settings full; not sending id=0x{:04x}", id); + return; + } + + self.insert(id, val); + } + + /// Takes every value held by `src`, replacing any held for the same + /// identifier here. + pub fn merge(&mut self, src: &Settings) { + for (id, val) in src.entries() { + self.insert(id, val); + } + } + + pub(crate) fn entries(&self) -> impl Iterator + '_ { + let len = self.len as usize; + self.ids[..len] + .iter() + .copied() + .zip(self.vals[..len].iter().copied()) + } + + fn position(&self, id: u16) -> Result { + self.ids[..self.len as usize].binary_search(&id) + } + + /// Holds `val` for `id`, keeping the numerically lowest identifiers when + /// there is no room for them all. + /// + /// Best effort by design: this is the receive path, and 6.5.2 lets an + /// endpoint ignore an identifier it does not understand. What it may never + /// do is lose one it does, which keeping the lowest guarantees. + fn insert(&mut self, id: u16, val: u32) { + let len = self.len as usize; + + match self.position(id) { + // 6.5.3: "Each parameter in a SETTINGS frame replaces any existing + // value for that parameter." + Ok(i) => self.vals[i] = val, + Err(i) if len < MAX_SETTINGS => self.insert_at(i, id, val), + Err(i) if i < MAX_SETTINGS => { + self.len -= 1; + self.insert_at(i, id, val); + } + Err(_) => { + tracing::debug!("settings full; not retaining id=0x{:04x}", id); + } + } + } + + /// Writes `(id, val)` at `i`, shifting the rest up. `self.len` must be + /// below [`MAX_SETTINGS`]. + fn insert_at(&mut self, i: usize, id: u16, val: u32) { + let len = self.len as usize; + debug_assert!(len < MAX_SETTINGS); + + self.ids.copy_within(i..len, i + 1); + self.vals.copy_within(i..len, i + 1); + self.ids[i] = id; + self.vals[i] = val; + self.len += 1; + } + + fn remove(&mut self, id: u16) { + if let Ok(i) = self.position(id) { + let len = self.len as usize; + self.ids.copy_within(i + 1..len, i); + self.vals.copy_within(i + 1..len, i); + self.len -= 1; + } + } + pub fn load(head: Head, payload: &[u8]) -> Result { debug_assert_eq!(head.kind(), crate::frame::Kind::Settings); if !head.stream_id().is_zero() { @@ -157,57 +281,24 @@ impl Settings { debug_assert!(!settings.flags.is_ack()); for raw in payload.chunks(6) { - match Setting::load(raw) { - Some(HeaderTableSize(val)) => { - settings.header_table_size = Some(val); - } - Some(EnablePush(val)) => match val { - 0 | 1 => { - settings.enable_push = Some(val); - } - _ => { - return Err(Error::InvalidSettingValue); - } - }, - Some(MaxConcurrentStreams(val)) => { - settings.max_concurrent_streams = Some(val); - } - Some(InitialWindowSize(val)) => { - if val as usize > MAX_INITIAL_WINDOW_SIZE { - return Err(Error::InvalidSettingValue); - } else { - settings.initial_window_size = Some(val); - } - } - Some(MaxFrameSize(val)) => { - if DEFAULT_MAX_FRAME_SIZE <= val && val <= MAX_MAX_FRAME_SIZE { - settings.max_frame_size = Some(val); - } else { - return Err(Error::InvalidSettingValue); - } - } - 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 => {} + let id: u16 = (u16::from(raw[0]) << 8) | u16::from(raw[1]); + let val: u32 = unpack_octets_4!(raw, 2, u32); + + // 0x0 is Reserved in the registry, so it names nothing; 6.5.2 says + // to ignore an identifier we do not understand, and this is one. + if id == 0 { + continue; } + + validate(id, val)?; + settings.insert(id, val); } Ok(settings) } fn payload_len(&self) -> usize { - let mut len = 0; - self.for_each(|_| len += 6); - len + self.len as usize * 6 } pub fn encode(&self, dst: &mut BytesMut) { @@ -219,43 +310,54 @@ impl Settings { head.encode(payload_len, dst); - // Encode the settings - self.for_each(|setting| { - tracing::trace!("encoding setting; val={:?}", setting); - setting.encode(dst) - }); - } - - fn for_each(&self, mut f: F) { - use self::Setting::*; - - if let Some(v) = self.header_table_size { - f(HeaderTableSize(v)); - } - - if let Some(v) = self.enable_push { - f(EnablePush(v)); + for (id, val) in self.entries() { + tracing::trace!("encoding setting; id=0x{:04x} val={}", id, val); + dst.put_u16(id); + dst.put_u32(val); } + } +} - if let Some(v) = self.max_concurrent_streams { - f(MaxConcurrentStreams(v)); - } +/// The bound a caller's own value must satisfy, which only one setting has. +fn bounds(id: u16, val: u32) { + if id == SETTINGS_MAX_FRAME_SIZE { + assert!(DEFAULT_MAX_FRAME_SIZE <= val && val <= MAX_MAX_FRAME_SIZE); + } +} - if let Some(v) = self.initial_window_size { - f(InitialWindowSize(v)); +/// The bounds RFC 9113 6.5.2 states for the settings this crate defines. +/// +/// An identifier it does not define has no bound to apply here: the endpoint +/// implementing the extension that defines it is the one that can see whether +/// it asked for it. +fn validate(id: u16, val: u32) -> Result<(), Error> { + match id { + SETTINGS_ENABLE_PUSH | SETTINGS_ENABLE_CONNECT_PROTOCOL if val > 1 => { + Err(Error::InvalidSettingValue) } - - if let Some(v) = self.max_frame_size { - f(MaxFrameSize(v)); + SETTINGS_INITIAL_WINDOW_SIZE if val as usize > MAX_INITIAL_WINDOW_SIZE => { + Err(Error::InvalidSettingValue) } - - if let Some(v) = self.max_header_list_size { - f(MaxHeaderListSize(v)); + SETTINGS_MAX_FRAME_SIZE + if !(DEFAULT_MAX_FRAME_SIZE..=MAX_MAX_FRAME_SIZE).contains(&val) => + { + Err(Error::InvalidSettingValue) } + _ => Ok(()), + } +} - if let Some(v) = self.enable_connect_protocol { - f(EnableConnectProtocol(v)); - } +/// The name RFC 9113 and RFC 8441 give an identifier, for `Debug`. +fn name(id: u16) -> Option<&'static str> { + match id { + SETTINGS_HEADER_TABLE_SIZE => Some("header_table_size"), + SETTINGS_ENABLE_PUSH => Some("enable_push"), + SETTINGS_MAX_CONCURRENT_STREAMS => Some("max_concurrent_streams"), + SETTINGS_INITIAL_WINDOW_SIZE => Some("initial_window_size"), + SETTINGS_MAX_FRAME_SIZE => Some("max_frame_size"), + SETTINGS_MAX_HEADER_LIST_SIZE => Some("max_header_list_size"), + SETTINGS_ENABLE_CONNECT_PROTOCOL => Some("enable_connect_protocol"), + _ => None, } } @@ -270,120 +372,380 @@ impl fmt::Debug for Settings { let mut builder = f.debug_struct("Settings"); builder.field("flags", &self.flags); - self.for_each(|setting| match setting { - Setting::EnablePush(v) => { - builder.field("enable_push", &v); - } - Setting::HeaderTableSize(v) => { - builder.field("header_table_size", &v); - } - Setting::InitialWindowSize(v) => { - builder.field("initial_window_size", &v); - } - Setting::MaxConcurrentStreams(v) => { - builder.field("max_concurrent_streams", &v); - } - Setting::MaxFrameSize(v) => { - builder.field("max_frame_size", &v); - } - Setting::MaxHeaderListSize(v) => { - builder.field("max_header_list_size", &v); - } - Setting::EnableConnectProtocol(v) => { - builder.field("enable_connect_protocol", &v); - } - }); + for (id, val) in self.entries() { + match name(id) { + Some(name) => builder.field(name, &val), + None => builder.field(&format!("0x{id:04x}"), &val), + }; + } builder.finish() } } -// ===== impl Setting ===== - -impl Setting { - /// Creates a new `Setting` with the correct variant corresponding to the - /// given setting id, based on the settings IDs defined in section - /// 6.5.2. - pub fn from_id(id: u16, val: u32) -> Option { - use self::Setting::*; - - match 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, +// ===== impl SettingsFlags ===== + +impl SettingsFlags { + fn load(bits: u8) -> SettingsFlags { + if bits & ACK == ACK { + SettingsFlags::Ack + } else { + SettingsFlags::Empty } } - /// Creates a new `Setting` by parsing the given buffer of 6 bytes, which - /// contains the raw byte representation of the setting, according to the - /// "SETTINGS format" defined in section 6.5.1. - /// - /// The `raw` parameter should have length at least 6 bytes, since the - /// length of the raw setting is exactly 6 bytes. - /// - /// # Panics - /// - /// If given a buffer shorter than 6 bytes, the function will panic. - fn load(raw: &[u8]) -> Option { - let id: u16 = (u16::from(raw[0]) << 8) | u16::from(raw[1]); - let val: u32 = unpack_octets_4!(raw, 2, u32); + fn is_ack(&self) -> bool { + matches!(self, SettingsFlags::Ack) + } +} - Setting::from_id(id, val) +impl From for u8 { + fn from(src: SettingsFlags) -> u8 { + match src { + SettingsFlags::Empty => 0, + SettingsFlags::Ack => ACK, + } } +} - fn encode(&self, dst: &mut BytesMut) { - use self::Setting::*; +impl fmt::Debug for SettingsFlags { + fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result { + util::debug_flags(f, u8::from(*self)) + .flag_if(self.is_ack(), "ACK") + .finish() + } +} - let (kind, val) = match *self { - HeaderTableSize(v) => (1, v), - EnablePush(v) => (2, v), - MaxConcurrentStreams(v) => (3, v), - InitialWindowSize(v) => (4, v), - MaxFrameSize(v) => (5, v), - MaxHeaderListSize(v) => (6, v), - EnableConnectProtocol(v) => (8, v), - }; +#[cfg(test)] +mod tests { + use super::*; - dst.put_u16(kind); - dst.put_u32(val); + /// The payload of `settings`, without the nine byte frame head. + fn payload(settings: &Settings) -> Vec { + let mut dst = BytesMut::new(); + settings.encode(&mut dst); + dst[9..].to_vec() } -} -// ===== impl SettingsFlags ===== + /// A SETTINGS payload carrying each `(identifier, value)` pair in order. + fn raw(pairs: &[(u16, u32)]) -> Vec { + let mut payload = Vec::new(); + for (id, val) in pairs { + payload.extend_from_slice(&id.to_be_bytes()); + payload.extend_from_slice(&val.to_be_bytes()); + } + payload + } -impl SettingsFlags { - pub fn empty() -> SettingsFlags { - SettingsFlags(0) + fn load(pairs: &[(u16, u32)]) -> Settings { + let head = Head::new(Kind::Settings, 0, StreamId::zero()); + Settings::load(head, &raw(pairs)).expect("a well formed SETTINGS payload") } - pub fn load(bits: u8) -> SettingsFlags { - SettingsFlags(bits & ALL) + #[test] + fn an_undefined_identifier_encodes_as_it_was_set() { + let mut settings = Settings::default(); + settings.set(0x9, Some(1)); + + assert_eq!(payload(&settings), raw(&[(0x9, 1)])); } - pub fn ack() -> SettingsFlags { - SettingsFlags(ACK) + #[test] + fn an_undefined_identifier_loads_as_it_arrived() { + assert_eq!(load(&[(0x9, 7)]).get(0x9), Some(7)); } - pub fn is_ack(&self) -> bool { - self.0 & ACK == ACK + /// The encoded order is the identifier order, not the order values were + /// set in; otherwise the mock io tests, which compare frames byte for + /// byte, would depend on the order a caller happened to use. + #[test] + fn undefined_identifiers_encode_in_identifier_order() { + let mut descending = Settings::default(); + descending.set(0x4d45, Some(4)); + descending.set(0x4d44, Some(2)); + descending.set(0x9, Some(1)); + + let mut ascending = Settings::default(); + ascending.set(0x9, Some(1)); + ascending.set(0x4d44, Some(2)); + ascending.set(0x4d45, Some(4)); + + assert_eq!( + payload(&descending), + raw(&[(0x9, 1), (0x4d44, 2), (0x4d45, 4)]), + ); + assert_eq!(payload(&descending), payload(&ascending)); + assert_eq!(descending, ascending, "and the two compare equal"); } -} -impl From for u8 { - fn from(src: SettingsFlags) -> u8 { - src.0 + /// RFC 9113 6.5.3: "Each parameter in a SETTINGS frame replaces any + /// existing value for that parameter." + #[test] + fn a_duplicate_undefined_identifier_takes_the_last_value() { + assert_eq!(load(&[(0x9, 1), (0x9, 2)]).get(0x9), Some(2)); + + let mut settings = Settings::default(); + settings.set(0x9, Some(1)); + settings.set(0x9, Some(2)); + assert_eq!(payload(&settings), raw(&[(0x9, 2)])); } -} -impl fmt::Debug for SettingsFlags { - fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result { - util::debug_flags(f, self.0) - .flag_if(self.is_ack(), "ACK") - .finish() + /// One identifier occupies one slot however it is addressed: the named + /// setter and `set` reach the same place, so the payload carries `0x4` + /// once, with the value written last. + #[test] + fn one_identifier_occupies_one_slot() { + let mut settings = Settings::default(); + settings.set_initial_window_size(Some(1_000)); + settings.set(4, Some(2_000)); + + assert_eq!(settings.initial_window_size(), Some(2_000)); + assert_eq!(settings.get(4), Some(2_000)); + assert_eq!(payload(&settings), raw(&[(4, 2_000)])); + + settings.set_initial_window_size(Some(3_000)); + assert_eq!(settings.get(4), Some(3_000)); + } + + /// `set` is exactly as strict as the setter it routes to, which for + /// `MAX_FRAME_SIZE` means refusing a value outside the legal range. + #[test] + #[should_panic] + fn a_defined_identifier_set_generically_keeps_its_own_bounds() { + Settings::default().set(5, Some(1)); + } + + #[test] + fn merge_takes_the_later_value_for_every_identifier() { + let mut settings = load(&[(0x9, 1), (0x10, 2), (4, 1_000)]); + settings.merge(&load(&[(0x10, 3), (0x4d44, 4), (4, 2_000)])); + + assert_eq!(settings.get(0x9), Some(1), "untouched"); + assert_eq!(settings.get(0x10), Some(3), "replaced"); + assert_eq!(settings.get(0x4d44), Some(4), "added"); + assert_eq!(settings.get(4), Some(2_000), "replaced, in its own field"); + } + + /// Outbound, a full frame refuses a new identifier rather than displacing + /// one already configured. Inbound prefers the lowest identifier; a sender + /// has no such licence, every identifier it holds having been asked for. + #[test] + fn a_full_frame_refuses_rather_than_displace_what_was_configured() { + let mut settings = Settings::default(); + for i in 0..(MAX_SETTINGS - RESERVED) as u16 { + settings.set(0x200 - i, Some(1)); + } + + settings.set(0x200, Some(2)); + assert_eq!( + settings.get(0x200), + Some(2), + "replacing one already held needs no room", + ); + } + + /// The capacity bound. 6.5.2 obliges an endpoint to ignore an identifier + /// it does not understand, and past the twentieth every identifier is one + /// of those: seventeen exist, ten registered and seven drafted. So the + /// excess is dropped rather than growing the frame or the connection. + #[test] + fn a_frame_beyond_capacity_ignores_the_excess() { + let over: Vec<(u16, u32)> = (0..MAX_SETTINGS as u32 + 5) + .map(|i| (0x1000 + i as u16, i)) + .collect(); + + let settings = load(&over); + + assert_eq!(settings.entries().count(), MAX_SETTINGS); + assert_eq!(settings.get(0x1000), Some(0), "the first is kept"); + assert_eq!( + settings.get(0x1000 + MAX_SETTINGS as u16 - 1), + Some(MAX_SETTINGS as u32 - 1), + "and so is the last that fits", + ); + assert_eq!( + settings.get(0x1000 + MAX_SETTINGS as u16), + None, + "the next is not" + ); + assert_eq!( + payload(&settings).len(), + MAX_SETTINGS * 6, + "nor is it emitted" + ); + } + + /// `None` means "send nothing for this", which the named setters took to + /// mean clearing the field. With one store it has to mean removing the + /// entry, including from the middle of the sorted run. + #[test] + fn clearing_a_setting_removes_it() { + let mut settings = Settings::default(); + settings.set_header_table_size(Some(1)); + settings.set_initial_window_size(Some(2)); + settings.set_max_header_list_size(Some(3)); + + settings.set_initial_window_size(None); + + assert_eq!(settings.initial_window_size(), None); + assert_eq!( + payload(&settings), + raw(&[(1, 1), (6, 3)]), + "and the rest keep order" + ); + + settings.set_header_table_size(None); + settings.set_max_header_list_size(None); + assert!(payload(&settings).is_empty()); + } + + /// `MAX_SETTINGS` high identifiers, then every identifier this crate defines. + /// Each must still be applied: a peer cannot suppress a setting h2 acts on + /// by padding the frame ahead of it. + #[test] + fn a_full_frame_still_admits_every_identifier_this_crate_defines() { + let mut pairs: Vec<(u16, u32)> = (0..MAX_SETTINGS as u32) + .map(|i| (0x1000 + i as u16, i)) + .collect(); + for id in [1u16, 2, 3, 4, 5, 6, 8] { + pairs.push(( + id, + if id == SETTINGS_MAX_FRAME_SIZE { + 16_384 + } else { + 1 + }, + )); + } + + let settings = load(&pairs); + + assert_eq!(settings.header_table_size(), Some(1)); + assert_eq!(settings.is_push_enabled(), Some(true)); + assert_eq!(settings.max_concurrent_streams(), Some(1)); + assert_eq!(settings.initial_window_size(), Some(1)); + assert_eq!(settings.max_frame_size(), Some(16_384)); + assert_eq!(settings.max_header_list_size(), Some(1)); + assert_eq!(settings.is_extended_connect_protocol_enabled(), Some(true)); + } + + /// The same, with `0x7` -- unassigned, so this crate does not define it -- + /// taking the eighth of the low slots. All eight are below everything else, + /// so all eight are kept. + #[test] + fn the_unassigned_low_identifier_crowds_nothing_out() { + let mut pairs: Vec<(u16, u32)> = (0..MAX_SETTINGS as u32) + .map(|i| (0x1000 + i as u16, i)) + .collect(); + pairs.push((0x7, 7)); + for id in [1u16, 2, 3, 4, 5, 6, 8] { + pairs.push(( + id, + if id == SETTINGS_MAX_FRAME_SIZE { + 16_384 + } else { + 1 + }, + )); + } + + let settings = load(&pairs); + + assert_eq!(settings.get(0x7), Some(7)); + assert_eq!(settings.initial_window_size(), Some(1)); + assert_eq!(settings.is_extended_connect_protocol_enabled(), Some(true)); + } + + /// What is kept is the numerically lowest, not the earliest: a low + /// identifier arriving last displaces the highest one held. + #[test] + fn the_lowest_identifiers_are_the_ones_retained() { + let mut pairs: Vec<(u16, u32)> = (0..MAX_SETTINGS as u32) + .map(|i| (0x2000 + i as u16, i)) + .collect(); + pairs.push((0x1000, 99)); + + let settings = load(&pairs); + + assert_eq!(settings.get(0x1000), Some(99), "the lowest is kept"); + assert_eq!( + settings.get(0x2000 + MAX_SETTINGS as u16 - 1), + None, + "and the highest it displaced is gone", + ); + assert_eq!(settings.entries().count(), MAX_SETTINGS); + let ids: Vec = settings.entries().map(|(id, _)| id).collect(); + assert!(ids.windows(2).all(|w| w[0] < w[1]), "still ascending"); + } + + /// 6.5.3 holds across the overflow: a repeated identifier takes its last + /// value even when the frame is full. + #[test] + fn a_duplicate_takes_its_last_value_in_a_full_frame() { + let mut pairs: Vec<(u16, u32)> = (0..MAX_SETTINGS as u32) + .map(|i| (0x1000 + i as u16, i)) + .collect(); + pairs.push((SETTINGS_INITIAL_WINDOW_SIZE, 1)); + pairs.push((SETTINGS_INITIAL_WINDOW_SIZE, 2)); + pairs.push((0x1000, 1)); + pairs.push((0x1000, 2)); + + let settings = load(&pairs); + + assert_eq!(settings.initial_window_size(), Some(2)); + assert_eq!(settings.get(0x1000), Some(2)); + } + + /// Outbound, dropping is never right: a caller asked for the setting to be + /// sent. `set` panics instead of quietly omitting it. + #[test] + #[cfg(debug_assertions)] + #[should_panic(expected = "would not be sent")] + fn a_setting_that_would_not_be_sent_panics() { + let mut settings = Settings::default(); + for i in 0..(MAX_SETTINGS - RESERVED) as u32 { + settings.set(0x1000 + i as u16, Some(i)); + } + settings.set(0x9000, Some(1)); + } + + /// The release half of the case above, which panics under the debug + /// assertions `cargo test` builds with. + #[test] + #[cfg(not(debug_assertions))] + fn a_setting_that_would_not_be_sent_is_dropped() { + let mut settings = Settings::default(); + for i in 0..(MAX_SETTINGS - RESERVED) as u32 { + settings.set(0x1000 + i as u16, Some(i)); + } + // Below everything held, the ordering that would displace one already + // configured if a full frame were allowed to insert. + settings.set(0x999, Some(1)); + + assert_eq!(settings.get(0x999), None, "the one that would not be sent"); + let highest = 0x1000 + (MAX_SETTINGS - RESERVED) as u16 - 1; + assert_eq!( + settings.get(highest), + Some((MAX_SETTINGS - RESERVED) as u32 - 1), + "and nothing already configured displaced", + ); + } + + /// And the reservation is what keeps that panic away from the identifiers + /// this crate defines, however many extensions a caller added. + #[test] + fn a_defined_identifier_always_has_room() { + let mut settings = Settings::default(); + for i in 0..(MAX_SETTINGS - RESERVED) as u32 { + settings.set(0x1000 + i as u16, Some(i)); + } + + settings.set(SETTINGS_INITIAL_WINDOW_SIZE, Some(1_000)); + assert_eq!(settings.initial_window_size(), Some(1_000)); + assert_eq!( + settings.get(0x1000 + (MAX_SETTINGS - RESERVED) as u16 - 1), + Some(MAX_SETTINGS as u32 - RESERVED as u32 - 1), + "and it displaced none of them", + ); } } diff --git a/src/lib.rs b/src/lib.rs index c04742dfd..bc7d0d519 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -136,7 +136,10 @@ mod share; pub mod fuzz_bridge; pub use crate::error::{Error, Reason}; -pub use crate::share::{FlowControl, Ping, PingPong, Pong, RecvStream, SendStream, StreamId}; +pub use crate::share::{ + FlowControl, PeerSettings, Ping, PingPong, Pong, RecvStream, SendStream, StreamId, + MAX_PEER_SETTINGS, +}; #[cfg(feature = "unstable")] pub use codec::{Codec, SendError, UserError}; diff --git a/src/proto/connection.rs b/src/proto/connection.rs index bb72d6068..94f1410c4 100644 --- a/src/proto/connection.rs +++ b/src/proto/connection.rs @@ -194,6 +194,15 @@ where self.inner.settings.send_settings(settings) } + /// Whether the peer's opening SETTINGS frame has been applied. + pub(crate) fn has_received_remote_initial_settings(&self) -> bool { + self.inner.settings.has_received_remote_initial_settings() + } + + pub(crate) fn peer_settings(&self) -> Option { + self.inner.streams.peer_settings() + } + /// Returns the maximum number of concurrent streams that may be initiated /// by this peer. pub(crate) fn max_send_streams(&self) -> usize { @@ -384,11 +393,22 @@ where } ready!(self.poll_ready(cx))?; - match self - .inner - .as_dyn() - .recv_frame(ready!(Pin::new(&mut self.codec).poll_next(cx)?))? + let frame = ready!(Pin::new(&mut self.codec).poll_next(cx)?); + + // RFC 9113 3.4: a client's preface "MUST be followed by a SETTINGS + // frame", a server's "MUST be the first frame the server sends", + // and "Clients and servers MUST treat an invalid connection + // preface as a connection error ... of type PROTOCOL_ERROR". So + // nothing may precede it, and a peer that sends something anyway + // is told so rather than quietly having no settings to report. + if !self.inner.settings.has_received_remote_initial_settings() + && !matches!(frame, None | Some(Frame::Settings(_))) { + proto_err!(conn: "frame before the peer's SETTINGS; frame={:?}", frame); + return Poll::Ready(Err(Error::library_go_away(Reason::PROTOCOL_ERROR))); + } + + match self.inner.as_dyn().recv_frame(frame)? { ReceivedFrame::Settings(frame) => { self.inner.settings.recv_settings( frame, @@ -559,6 +579,7 @@ where fn recv_frame(&mut self, frame: Option) -> Result { use crate::frame::Frame::*; + match frame { Some(Headers(frame)) => { tracing::trace!(?frame, "recv HEADERS"); diff --git a/src/proto/settings.rs b/src/proto/settings.rs index d6155fc3d..0f8ba1d77 100644 --- a/src/proto/settings.rs +++ b/src/proto/settings.rs @@ -99,6 +99,11 @@ impl Settings { } } + /// Whether the peer's opening SETTINGS frame has been applied. + pub(crate) fn has_received_remote_initial_settings(&self) -> bool { + self.has_received_remote_initial_settings + } + /// Sets `true` to `self.has_received_remote_initial_settings`. /// Returns `true` if this method is called for the first time. /// (i.e. it is the initial SETTINGS frame from the remote peer) diff --git a/src/proto/streams/streams.rs b/src/proto/streams/streams.rs index ea81ee0d7..140c162a4 100644 --- a/src/proto/streams/streams.rs +++ b/src/proto/streams/streams.rs @@ -101,6 +101,26 @@ struct Actions { /// Task that calls `poll_complete`. task: Option, + /// Whoever is waiting for the peer's opening SETTINGS to arrive. + peer_settings_task: Option, + + /// Set when the half of `Streams` the connection task holds goes away + /// without an EOF, so that a waiter is not left parked on settings that + /// can no longer arrive. + connection_gone: bool, + + /// Every SETTINGS value the peer has sent, with the last value for an + /// identifier winning, or `None` until its opening frame has been applied. + /// + /// `None` rather than an empty frame, so that "the peer has not spoken + /// yet" and "the peer sent nothing" cannot be confused: every value read + /// out of an absent frame would be a default rather than an answer. + /// + /// Held because a setting outlives the frame that carried it, and because + /// this is the only part of the connection a `SendRequest` can reach. The + /// frame is a fixed size, so what a peer sends cannot grow it. + peer_settings: Option, + /// If the connection errors, a copy is kept for any StreamRefs. conn_error: Option, } @@ -245,6 +265,12 @@ where let send_buffer = &mut *send_buffer; me.counts.apply_remote_settings(frame, is_initial); + if is_initial { + me.actions.peer_settings = Some(frame.clone()); + me.actions.wake_peer_settings(); + } else if let Some(held) = &mut me.actions.peer_settings { + held.merge(frame); + } me.actions.send.apply_remote_settings( frame, @@ -359,6 +385,46 @@ where )) } + /// Resolves once the peer's opening SETTINGS frame has been applied. + /// + /// Every value `peer_setting` reports reads as absent until then, so this + /// is the point at which those answers become the peer's rather than + /// nothing at all. An HTTP/2 peer must send SETTINGS as its first frame, + /// so the wait is bounded by the connection: it ends in an error if the + /// connection ends first. + pub(crate) fn poll_peer_settings( + &self, + cx: &Context, + ) -> Poll> { + let mut me = self.inner.lock().unwrap(); + let me = &mut *me; + + if let Some(settings) = &me.actions.peer_settings { + return Poll::Ready(Ok(crate::PeerSettings::new(settings.clone()))); + } + me.actions.ensure_no_conn_error()?; + if me.actions.connection_gone { + return Poll::Ready(Err(proto::Error::from(io::Error::new( + io::ErrorKind::BrokenPipe, + "connection closed before the peer's SETTINGS arrived", + )) + .into())); + } + me.actions.peer_settings_task = Some(cx.waker().clone()); + Poll::Pending + } + + /// Every setting the peer has sent, copied out from under the lock. + pub(crate) fn peer_settings(&self) -> Option { + self.inner + .lock() + .unwrap() + .actions + .peer_settings + .clone() + .map(crate::PeerSettings::new) + } + pub(crate) fn is_extended_connect_protocol_enabled(&self) -> bool { self.inner .lock() @@ -460,6 +526,9 @@ impl Inner { recv: Recv::new(peer, &config), send: Send::new(&config), task: None, + peer_settings_task: None, + connection_gone: false, + peer_settings: None, conn_error: None, }, store: Store::new(), @@ -795,6 +864,7 @@ impl Inner { }); actions.conn_error = Some(err); + actions.wake_peer_settings(); last_processed_id } @@ -826,6 +896,7 @@ impl Inner { }); actions.conn_error = Some(err); + actions.wake_peer_settings(); Ok(()) } @@ -954,6 +1025,8 @@ impl Inner { ); } + actions.wake_peer_settings(); + tracing::trace!("Streams::recv_eof"); self.store.for_each(|stream| { @@ -1182,6 +1255,8 @@ where if let Some(task) = inner.actions.task.take() { task.wake(); } + inner.actions.connection_gone = true; + inner.actions.wake_peer_settings(); } } } @@ -1833,6 +1908,17 @@ impl Actions { } } + /// Wake whoever is waiting on the peer's opening SETTINGS. + /// + /// Called both when they arrive and when the connection ends without + /// them, because a waiter parked on settings that will never come never + /// wakes. + fn wake_peer_settings(&mut self) { + if let Some(task) = self.peer_settings_task.take() { + task.wake(); + } + } + fn ensure_no_conn_error(&self) -> Result<(), proto::Error> { if let Some(ref err) = self.conn_error { Err(err.clone()) diff --git a/src/server.rs b/src/server.rs index 931661a19..df26e1b4a 100644 --- a/src/server.rs +++ b/src/server.rs @@ -118,7 +118,7 @@ use crate::codec::{Codec, UserError}; use crate::frame::{self, Pseudo, PushPromiseHeaderError, Reason, Settings, StreamId}; use crate::proto::{self, Config, Error, Prioritized}; -use crate::{FlowControl, PingPong, RecvStream, SendStream}; +use crate::{FlowControl, PeerSettings, PingPong, RecvStream, SendStream}; use bytes::{Buf, Bytes}; use http::{HeaderMap, Method, Request, Response}; @@ -314,9 +314,16 @@ impl fmt::Debug for SendPushedResponse { enum Handshaking { /// State 1. Connection is flushing pending SETTINGS frame. Flushing(Instrumented>>), - /// State 2. Connection is waiting for the client preface. + /// State 2. Connection is waiting for the twenty-four byte client preface. ReadingPreface(Instrumented>>), - /// State 3. Handshake is done, polling again would panic. + /// State 3. Connection is waiting for the `SETTINGS` frame that RFC 9113 + /// 3.4 makes the rest of that preface. + /// + /// Boxed: a `Connection` is the largest thing here by some way, and one + /// allocation per handshake is cheaper than carrying its width in every + /// state. + ReadingSettings(Box>), + /// State 4. Handshake is done, polling again would panic. Done, } @@ -503,6 +510,39 @@ where Ok(()) } + /// The `SETTINGS` parameters the client peer sent. + /// + /// No wait, unlike the client side, and nothing optional about it: RFC + /// 9113 3.4 makes the client's `SETTINGS` part of the connection preface, + /// so [`handshake`] does not yield a connection until it has arrived. A + /// peer that sends any other frame first has sent an invalid preface, + /// which is a `PROTOCOL_ERROR` and no connection at all. + /// + /// A peer that sends an *empty* `SETTINGS`, which 3.4 permits, is reported + /// as such: see [`PeerSettings::is_empty`]. + /// + /// # Examples + /// + /// ``` + /// # use tokio::io::{AsyncRead, AsyncWrite}; + /// # use bytes::Buf; + /// # fn doc(conn: &h2::server::Connection) + /// # where T: AsyncRead + AsyncWrite + Unpin + /// # { + /// // The client's opening SETTINGS have been applied by the time a + /// // connection exists at all, so there is nothing here to be absent. + /// for (id, value) in conn.peer_settings().iter() { + /// println!("{id:#06x} = {value}"); + /// } + /// # } + /// # pub fn main() {} + /// ``` + pub fn peer_settings(&self) -> PeerSettings { + self.connection + .peer_settings() + .expect("the handshake does not complete without the client's SETTINGS") + } + /// Returns `Ready` when the underlying connection has closed. /// /// If any new inbound streams are received during a call to `poll_closed`, @@ -700,6 +740,63 @@ impl Builder { self } + /// Sets the `SETTINGS` parameter `id` to `value`. + /// + /// [RFC 9113 Section 5.5][1] permits an extension to define new settings + /// and makes using one safe without prior arrangement: implementations + /// "MUST ignore unknown or unsupported values in all extensible protocol + /// elements". So an identifier h2 knows nothing about is sent unexamined, + /// and is safe to send to a peer that knows nothing about it either. + /// Identifiers are allocated in the IANA ["HTTP/2 Settings" registry][reg]; + /// h2 carries the number and does not define what it means. + /// + /// An identifier h2 does define is routed to the setting it names and + /// checked the same way, so `setting(4, n)` is another spelling of + /// [`initial_window_size`](Builder::initial_window_size) rather than a + /// second copy of it, and one identifier can never be sent twice in one + /// frame. What the peer sent is read with + /// [`Connection::peer_settings`](Connection::peer_settings). + /// + /// # Examples + /// + /// ``` + /// # use tokio::io::{AsyncRead, AsyncWrite}; + /// # use h2::server::*; + /// # + /// # fn doc(my_io: T) + /// # -> Handshake + /// # { + /// // An extension names its own identifiers; h2 only carries them. + /// const ENABLE_METADATA: u16 = 0x4d44; + /// + /// let server_fut = Builder::new() + /// .setting(ENABLE_METADATA, 1) + /// .handshake(my_io); + /// # server_fut + /// # } + /// # + /// # pub fn main() {} + /// ``` + /// + /// # Panics + /// + /// Panics on a value the method named for that setting would panic on: + /// `MAX_FRAME_SIZE` (`5`) outside the range + /// [`Builder::max_frame_size`] accepts. + /// + /// Panics if twelve identifiers h2 does not itself define are already set + /// and this is not one of them, since a setting a caller asked for must not + /// be quietly left unsent. The twelve are what a `SETTINGS` frame has to + /// give here once room is reserved for every identifier h2 defines, so + /// those can never panic. + /// + /// [1]: https://datatracker.ietf.org/doc/html/rfc9113#section-5.5 + /// [reg]: https://www.iana.org/assignments/http2-parameters/http2-parameters.xhtml + pub fn setting(&mut self, id: u16, value: u32) -> &mut Self { + self.settings.set(id, Some(value)); + self + } + /// Indicates the initial window size (in octets) for connection-level flow control /// for received data. /// @@ -1538,19 +1635,62 @@ where }, ); - tracing::trace!("connection established!"); let mut c = Connection { connection }; if let Some(sz) = self.builder.initial_target_connection_window_size { c.set_target_window_size(sz); } - return Poll::Ready(Ok(c)); + self.state = Handshaking::ReadingSettings(Box::new(c)); + } + Handshaking::ReadingSettings(c) => { + // RFC 9113 3.4: the twenty-four octets "MUST be followed + // by a SETTINGS frame", so the preface is not complete + // until that frame has been applied, and a connection + // handed over before then would report its peer's + // parameters as absent rather than as what they are. + // + // No timeout: 3.4 specifies none, and this crate cannot + // impose one. It reads a clock -- reset-stream expiry + // compares `Instant::now()` -- but only where something + // else already polls it. A peer that sends the preface and + // stops produces no events, so a deadline here would need a + // timer to wake at, and this crate has none. Bounding the + // wait is the caller's, and a caller that does not bound it + // holds the connection for as long as the peer stays silent. + if !c.connection.has_received_remote_initial_settings() { + match c.poll_closed(cx) { + Poll::Ready(Ok(())) => { + self.state = Handshaking::Done; + return Poll::Ready(Err(crate::Error::from_io(io::Error::new( + io::ErrorKind::UnexpectedEof, + "closed before the client's SETTINGS", + )))); + } + Poll::Ready(Err(e)) => { + self.state = Handshaking::Done; + return Poll::Ready(Err(e)); + } + Poll::Pending => { + if !c.connection.has_received_remote_initial_settings() { + return Poll::Pending; + } + } + } + } + + tracing::trace!("connection established!"); + break; } Handshaking::Done => { panic!("Handshaking::poll() called again after handshaking was complete") } } } + + match std::mem::replace(&mut self.state, Handshaking::Done) { + Handshaking::ReadingSettings(c) => Poll::Ready(Ok(*c)), + _ => unreachable!("only ReadingSettings breaks the loop"), + } } } @@ -1782,6 +1922,7 @@ where match *self { Handshaking::Flushing(_) => f.write_str("Flushing(_)"), Handshaking::ReadingPreface(_) => f.write_str("ReadingPreface(_)"), + Handshaking::ReadingSettings(_) => f.write_str("ReadingSettings(_)"), Handshaking::Done => f.write_str("Done"), } } diff --git a/src/share.rs b/src/share.rs index 101987342..6ec0c04ae 100644 --- a/src/share.rs +++ b/src/share.rs @@ -1,5 +1,5 @@ use crate::codec::UserError; -use crate::frame::Reason; +use crate::frame::{self, Reason}; use crate::proto::{self, WindowSize}; use bytes::{Buf, Bytes}; @@ -196,6 +196,25 @@ pub struct FlowControl { inner: proto::OpaqueStreamRef, } +/// The `SETTINGS` parameters a peer has sent. +/// +/// Taken from [`client::SendRequest::peer_settings`] or +/// [`server::Connection::peer_settings`]. A snapshot can only be taken once +/// the peer's opening `SETTINGS` frame has been applied, so an identifier +/// absent from one is an identifier the peer sent no value for, and not one +/// whose frame has yet to arrive. +/// +/// Identifiers this crate defines are included, as the peer sent them rather +/// than as this crate went on to act on them. A repeated identifier appears +/// once, with the last value received, as [RFC 9113 Section 6.5.3][1] +/// requires. +/// +/// [`client::SendRequest::peer_settings`]: crate::client::SendRequest::peer_settings +/// [`server::Connection::peer_settings`]: crate::server::Connection::peer_settings +/// [1]: https://datatracker.ietf.org/doc/html/rfc9113#section-6.5.3 +#[derive(Clone, Debug, Default, Eq, PartialEq)] +pub struct PeerSettings(frame::Settings); + /// A handle to send and receive PING frames with the peer. // NOT Clone on purpose pub struct PingPong { @@ -384,6 +403,56 @@ impl SendStream { // ===== impl StreamId ===== +/// How many of a peer's `SETTINGS` parameters are retained. +/// +/// The numerically lowest are kept, so every identifier this crate defines is +/// always among them; what a peer can cost itself is the visibility of its +/// higher ones. A caller mirroring a [`PeerSettings`] into storage of its own +/// needs this to size it. +pub const MAX_PEER_SETTINGS: usize = frame::MAX_SETTINGS; + +impl PeerSettings { + pub(crate) fn new(settings: frame::Settings) -> Self { + PeerSettings(settings) + } + + /// The value the peer sent for the `SETTINGS` identifier `id`, or `None` + /// if it sent none. + pub fn get(&self, id: u16) -> Option { + self.0.get(id) + } + + /// Every parameter the peer sent, ascending by identifier, which is not + /// the order the peer sent them in. + pub fn iter(&self) -> impl Iterator + '_ { + self.0.entries() + } + + /// Whether the peer advertised the [extended CONNECT protocol][1]. + /// + /// `SETTINGS_ENABLE_CONNECT_PROTOCOL`, whose initial value RFC 8441 gives + /// as `0`, so a peer that sent no value for it has not advertised it. + /// + /// Unlike [`client::SendRequest::is_extended_connect_protocol_enabled`], + /// this cannot be asked too early: a snapshot exists only once the peer's + /// opening frame has been applied, so `false` here is the peer's answer + /// and not the absence of one. + /// + /// [1]: https://datatracker.ietf.org/doc/html/rfc8441#section-4 + /// [`client::SendRequest::is_extended_connect_protocol_enabled`]: crate::client::SendRequest::is_extended_connect_protocol_enabled + pub fn is_extended_connect_protocol_enabled(&self) -> bool { + self.0 + .is_extended_connect_protocol_enabled() + .unwrap_or(false) + } + + /// Whether the peer sent no parameters at all, which an empty `SETTINGS` + /// frame is entitled to do. + pub fn is_empty(&self) -> bool { + self.0.entries().next().is_none() + } +} + impl StreamId { pub(crate) fn from_internal(id: crate::frame::StreamId) -> Self { StreamId(id.into()) diff --git a/tests/h2-support/src/frames.rs b/tests/h2-support/src/frames.rs index 5454ebd79..f6a8d000f 100644 --- a/tests/h2-support/src/frames.rs +++ b/tests/h2-support/src/frames.rs @@ -390,6 +390,11 @@ impl Mock { self } + pub fn setting(mut self, id: u16, val: u32) -> Self { + self.0.set(id, Some(val)); + self + } + pub fn header_table_size(mut self, val: u32) -> Self { self.0.set_header_table_size(Some(val)); self diff --git a/tests/h2-tests/tests/client_request.rs b/tests/h2-tests/tests/client_request.rs index 189f6917c..6bd613d73 100644 --- a/tests/h2-tests/tests/client_request.rs +++ b/tests/h2-tests/tests/client_request.rs @@ -1721,6 +1721,362 @@ async fn invalid_connect_protocol_enabled_setting() { join(srv, h2).await; } +/// Three identifiers IANA has assigned that h2 does not implement -- the case +/// this mechanism exists for. h2 defines no meaning for any of them, and the +/// point of every test below is that it does not need to. +const NO_RFC7540_PRIORITIES: u16 = 0x9; // RFC 9218 +const TLS_RENEG_PERMITTED: u16 = 0x10; // MS-HTTP2EXT +const ENABLE_METADATA: u16 = 0x4d44; // draft-beky-httpbis-metadata + +#[tokio::test] +async fn arbitrary_settings_are_advertised_during_the_handshake() { + h2_support::trace_init!(); + let (io, mut srv) = mock::new(); + + let srv = async move { + let settings = srv.assert_client_handshake().await; + + // Set in descending order below and written in ascending order here: + // the frame's bytes are a function of the values, not of the calls. + let mut bytes = BytesMut::new(); + settings.encode(&mut bytes); + assert_eq!( + &bytes[..], + &[ + 0, 0, 18, // payload length: three settings + 4, // SETTINGS + 0, // no flags + 0, 0, 0, 0, // stream 0 + 0, 0x09, 0, 0, 0, 1, // + 0, 0x10, 0, 0, 0x03, 0xe8, // + 0x4d, 0x44, 0, 0, 0x27, 0x10, // + ][..], + ); + + srv.recv_frame( + frames::headers(1) + .request("GET", "https://example.com/") + .eos(), + ) + .await; + srv.send_frame(frames::headers(1).response(200).eos()).await; + }; + + let h2 = async move { + let (mut client, mut h2) = client::Builder::new() + .setting(ENABLE_METADATA, 10_000) + .setting(TLS_RENEG_PERMITTED, 1_000) + .setting(NO_RFC7540_PRIORITIES, 1) + .handshake::<_, Bytes>(io) + .await + .unwrap(); + + let request = Request::get("https://example.com/").body(()).unwrap(); + let (response, _) = client.send_request(request, true).unwrap(); + h2.drive(response).await.unwrap(); + }; + + join(srv, h2).await; +} + +#[tokio::test] +async fn arbitrary_settings_the_server_sent_are_read_back() { + h2_support::trace_init!(); + let (io, mut srv) = mock::new(); + + let srv = async move { + let settings = srv + .assert_client_handshake_with_settings( + frames::settings() + .setting(NO_RFC7540_PRIORITIES, 1) + .setting(TLS_RENEG_PERMITTED, 1_000) + .setting(ENABLE_METADATA, u32::MAX), + ) + .await; + assert_default_settings!(settings); + + srv.recv_frame( + frames::headers(1) + .request("GET", "https://example.com/") + .eos(), + ) + .await; + srv.send_frame(frames::headers(1).response(200).eos()).await; + }; + + let h2 = async move { + let (mut client, mut h2) = client::handshake(io).await.unwrap(); + + // A request only to drive the connection far enough that the server's + // SETTINGS have been received and acknowledged. No GOAWAY: there is no + // value of an identifier h2 does not define that h2 may reject, and + // `u32::MAX` is the proof. + let request = Request::get("https://example.com/").body(()).unwrap(); + let (response, _) = client.send_request(request, true).unwrap(); + let response = h2.drive(response).await.unwrap(); + assert_eq!(response.status(), 200); + + let settings = h2.drive(client.peer_settings()).await.unwrap(); + assert_eq!(settings.get(NO_RFC7540_PRIORITIES), Some(1)); + assert_eq!(settings.get(TLS_RENEG_PERMITTED), Some(1_000)); + assert_eq!(settings.get(ENABLE_METADATA), Some(u32::MAX)); + // And one the server said nothing about stays unanswered, rather than + // reading as a zero the server never granted. + assert_eq!(settings.get(0x4d45), None); + + // The same three, without having to know their identifiers first. + assert_eq!( + settings.iter().collect::>(), + vec![ + (NO_RFC7540_PRIORITIES, 1), + (TLS_RENEG_PERMITTED, 1_000), + (ENABLE_METADATA, u32::MAX) + ], + ); + }; + + join(srv, h2).await; +} + +#[tokio::test] +async fn an_identifier_the_server_sent_no_value_for_is_absent() { + h2_support::trace_init!(); + let (io, mut srv) = mock::new(); + + let srv = async move { + let settings = srv.assert_client_handshake().await; + assert_default_settings!(settings); + + srv.recv_frame( + frames::headers(1) + .request("GET", "https://example.com/") + .eos(), + ) + .await; + srv.send_frame(frames::headers(1).response(200).eos()).await; + }; + + let h2 = async move { + let (mut client, mut h2) = client::handshake(io).await.unwrap(); + + let request = Request::get("https://example.com/").body(()).unwrap(); + let (response, _) = client.send_request(request, true).unwrap(); + h2.drive(response).await.unwrap(); + + // The server's SETTINGS carried none of these, and absence here can + // only mean that: a snapshot cannot be taken before the frame lands. + let settings = h2.drive(client.peer_settings()).await.unwrap(); + assert_eq!(settings.get(NO_RFC7540_PRIORITIES), None); + assert!(settings.is_empty()); + }; + + join(srv, h2).await; +} + +#[tokio::test] +async fn the_last_value_of_a_repeated_identifier_is_the_one_read_back() { + h2_support::trace_init!(); + let (io, mut srv) = mock::new(); + + // RFC 9113 6.5.3. Written as bytes because the frame type this crate + // encodes with cannot hold one identifier twice, which is the point -- + // only a peer can send such a frame. + let srv = async move { + srv.send_bytes(&[ + 0, 0, 12, // payload length: two settings + 4, // SETTINGS + 0, // no flags + 0, 0, 0, 0, // stream 0 + 0, 0x09, 0, 0, 0, 1, // + 0, 0x09, 0, 0, 0, 2, // + ]) + .await; + srv.read_preface().await.unwrap(); + + let settings = assert_settings!(srv.next().await.expect("unexpected EOF").unwrap()); + assert_default_settings!(settings); + srv.send(frame::Settings::ack().into()).await.unwrap(); + assert!(assert_settings!(srv.next().await.unwrap().unwrap()).is_ack()); + + srv.recv_frame( + frames::headers(1) + .request("GET", "https://example.com/") + .eos(), + ) + .await; + srv.send_frame(frames::headers(1).response(200).eos()).await; + }; + + let h2 = async move { + let (mut client, mut h2) = client::handshake(io).await.unwrap(); + + let request = Request::get("https://example.com/").body(()).unwrap(); + let (response, _) = client.send_request(request, true).unwrap(); + h2.drive(response).await.unwrap(); + + let settings = h2.drive(client.peer_settings()).await.unwrap(); + assert_eq!(settings.get(NO_RFC7540_PRIORITIES), Some(2)); + }; + + join(srv, h2).await; +} + +#[tokio::test] +async fn a_later_settings_frame_replaces_only_what_it_carries() { + h2_support::trace_init!(); + let (io, mut srv) = mock::new(); + + let srv = async move { + let settings = srv + .assert_client_handshake_with_settings( + frames::settings() + .setting(NO_RFC7540_PRIORITIES, 1) + .setting(TLS_RENEG_PERMITTED, 1_000), + ) + .await; + assert_default_settings!(settings); + + srv.recv_frame( + frames::headers(1) + .request("GET", "https://example.com/") + .eos(), + ) + .await; + srv.send_frame(frames::headers(1).response(200).eos()).await; + srv.send_frame(frames::settings().setting(TLS_RENEG_PERMITTED, 2_000)) + .await; + srv.recv_frame(frames::settings_ack()).await; + }; + + let h2 = async move { + let (mut client, mut h2) = client::handshake(io).await.unwrap(); + + let request = Request::get("https://example.com/").body(()).unwrap(); + let (response, _) = client.send_request(request, true).unwrap(); + h2.drive(response).await.unwrap(); + + let request = Request::get("https://example.com/").body(()).unwrap(); + let (response, _) = client.send_request(request, true).unwrap(); + h2.drive(async { + let _ = response.await; + }) + .await; + + let settings = h2.drive(client.peer_settings()).await.unwrap(); + assert_eq!(settings.get(NO_RFC7540_PRIORITIES), Some(1), "untouched"); + assert_eq!(settings.get(TLS_RENEG_PERMITTED), Some(2_000), "replaced"); + }; + + join(srv, h2).await; +} + +#[tokio::test] +async fn an_identifier_h2_defines_is_set_and_read_in_its_own_field() { + h2_support::trace_init!(); + let (io, mut srv) = mock::new(); + + // `0x4` is `SETTINGS_INITIAL_WINDOW_SIZE`, which h2 defines. Set through + // both doors it is still one setting: the frame carries it once, with the + // value set last, and both doors read that value back. + let srv = async move { + let settings = srv + .assert_client_handshake_with_settings(frames::settings().initial_window_size(4_000)) + .await; + + let mut bytes = BytesMut::new(); + settings.encode(&mut bytes); + assert_eq!( + &bytes[..], + &[ + 0, 0, 6, // payload length: one setting + 4, // SETTINGS + 0, // no flags + 0, 0, 0, 0, // stream 0 + 0, 4, 0, 0, 0x07, 0xd0, // INITIAL_WINDOW_SIZE = 2_000 + ][..], + ); + assert_eq!(settings.get(4), Some(2_000)); + + srv.recv_frame( + frames::headers(1) + .request("GET", "https://example.com/") + .eos(), + ) + .await; + srv.send_frame(frames::headers(1).response(200).eos()).await; + }; + + let h2 = async move { + let (mut client, mut h2) = client::Builder::new() + .initial_window_size(1_000) + .setting(4, 2_000) + .handshake::<_, Bytes>(io) + .await + .unwrap(); + + let request = Request::get("https://example.com/").body(()).unwrap(); + let (response, _) = client.send_request(request, true).unwrap(); + h2.drive(response).await.unwrap(); + + let settings = h2.drive(client.peer_settings()).await.unwrap(); + assert_eq!(settings.get(4), Some(4_000)); + }; + + join(srv, h2).await; +} + +#[tokio::test] +async fn peer_settings_resolve_without_a_request_being_sent() { + h2_support::trace_init!(); + let (io, mut srv) = mock::new(); + + let srv = async move { + let settings = srv + .assert_client_handshake_with_settings( + frames::settings().setting(NO_RFC7540_PRIORITIES, 1), + ) + .await; + assert_default_settings!(settings); + srv.recv_eof().await; + }; + + let h2 = async move { + let (client, mut h2) = client::handshake(io).await.unwrap(); + + // Nothing is sent on this connection and nothing needs to be. The + // peer's SETTINGS is the first frame it sends, so waiting for it is + // the whole of finding out what the peer said. + let settings = h2.drive(client.peer_settings()).await.unwrap(); + + assert_eq!(settings.get(NO_RFC7540_PRIORITIES), Some(1)); + }; + + join(srv, h2).await; +} + +#[tokio::test] +async fn peer_settings_end_in_an_error_if_the_connection_does() { + h2_support::trace_init!(); + let (io, mut srv) = mock::new(); + + let srv = async move { + srv.read_preface().await.unwrap(); + let settings = assert_settings!(srv.next().await.expect("unexpected EOF").unwrap()); + assert_default_settings!(settings); + // And then nothing. A waiter parked on settings that will never + // arrive must not be left parked. + }; + + let h2 = async move { + let (client, mut h2) = client::handshake(io).await.unwrap(); + + let err = h2.drive(client.peer_settings()).await.unwrap_err(); + assert!(err.is_io(), "{err:?}"); + }; + + join(srv, h2).await; +} + #[tokio::test] async fn extended_connect_request() { h2_support::trace_init!(); diff --git a/tests/h2-tests/tests/server.rs b/tests/h2-tests/tests/server.rs index a1952fbbb..9f9a89551 100644 --- a/tests/h2-tests/tests/server.rs +++ b/tests/h2-tests/tests/server.rs @@ -1712,6 +1712,113 @@ async fn extended_connect_protocol_enabled_during_handshake() { join(client, srv).await; } +#[tokio::test] +async fn arbitrary_settings_are_advertised_during_handshake() { + h2_support::trace_init!(); + + let (io, mut client) = mock::new(); + + // The identifiers are the ones draft-ietf-webtrans-http2 registers, used + // here as plausible extension codepoints: h2 defines no meaning for them, + // and nothing below needs it to. + let client = async move { + let settings = client.assert_server_handshake().await; + + // Set in descending order below and written in ascending order here: + // the frame's bytes are a function of the values, not of the calls. + let mut bytes = BytesMut::new(); + settings.encode(&mut bytes); + assert_eq!( + &bytes[..], + &[ + 0, 0, 18, // payload length: three settings + 4, // SETTINGS + 0, // no flags + 0, 0, 0, 0, // stream 0 + 0, 0x09, 0, 0, 0, 1, // + 0, 0x10, 0, 0, 0x03, 0xe8, // + 0x4d, 0x44, 0, 0, 0x27, 0x10, // + ][..], + ); + }; + + let srv = async move { + let mut builder = server::Builder::new(); + builder.setting(0x4d44, 10_000); + builder.setting(0x10, 1_000); + builder.setting(0x9, 1); + + let mut srv = builder.handshake::<_, Bytes>(io).await.expect("handshake"); + + poll_fn(move |cx| srv.poll_closed(cx)) + .await + .expect("server"); + }; + + join(client, srv).await; +} + +#[tokio::test] +async fn arbitrary_settings_the_client_sent_are_read_back() { + h2_support::trace_init!(); + + let (io, mut client) = mock::new(); + + let client = async move { + client + .assert_server_handshake_with_settings( + frames::settings() + .setting(0x9, 1) + .setting(0x10, 1_000) + .setting(0x4d44, u32::MAX), + ) + .await; + client + .send_frame( + frames::headers(1) + .request("GET", "https://example.com/") + .eos(), + ) + .await; + client + .recv_frame(frames::headers(1).response(200).eos()) + .await; + }; + + let srv = async move { + let mut srv = server::handshake(io).await.expect("handshake"); + + // No check that the settings have arrived: `handshake` does not + // resolve until they have, which 3.4 makes part of the preface. + // A client's SETTINGS is the first frame of a connection, so by the + // time this server has a request to answer it has the client's values. + // No GOAWAY: there is no value of an identifier h2 does not define + // that h2 may reject, and `u32::MAX` is the proof. + let (_req, mut respond) = srv.next().await.unwrap().expect("request"); + respond + .send_response(Response::new(()), true) + .expect("respond"); + + let settings = srv.peer_settings(); + assert_eq!(settings.get(0x9), Some(1)); + assert_eq!(settings.get(0x10), Some(1_000)); + assert_eq!(settings.get(0x4d44), Some(u32::MAX)); + assert_eq!(settings.get(0x4d45), None); + + // The same three, without having to know their identifiers first. + assert_eq!( + settings.iter().collect::>(), + vec![(0x9, 1), (0x10, 1_000), (0x4d44, u32::MAX)], + ); + + poll_fn(move |cx| srv.poll_closed(cx)) + .await + .expect("server"); + }; + + join(client, srv).await; +} + #[tokio::test] async fn reject_pseudo_protocol_on_non_connect_request() { h2_support::trace_init!(); @@ -1994,3 +2101,34 @@ async fn remote_reset_does_not_panic_connection_driver() { .expect("server task timed out") .expect("server task panicked"); } + +/// 3.4: the preface "MUST be followed by a SETTINGS frame", and an invalid +/// connection preface is a `PROTOCOL_ERROR`. A request that arrives instead is +/// not a connection with no settings; it is not a connection. +#[tokio::test] +async fn a_frame_before_the_client_settings_is_a_protocol_error() { + h2_support::trace_init!(); + let (io, mut client) = mock::new(); + + let client = async move { + client.write_preface().await; + let _ = client.next().await; // the server's own SETTINGS + client + .send_frame( + frames::headers(1) + .request("GET", "https://example.com/") + .eos(), + ) + .await; + client.recv_frame(frames::go_away(0).protocol_error()).await; + }; + + let srv = async move { + // The handshake itself refuses it: 3.4 makes the client's SETTINGS + // part of the preface, so this never becomes a connection. + let err = server::handshake(io).await.expect_err("no connection"); + assert_eq!(err.reason(), Some(Reason::PROTOCOL_ERROR)); + }; + + join(client, srv).await; +}