diff --git a/src/client.rs b/src/client.rs index 9d0ff31d9..9f7343ac4 100644 --- a/src/client.rs +++ b/src/client.rs @@ -1432,6 +1432,24 @@ where self.inner.take_user_pings().map(PingPong::new) } + /// Returns the origins the server announced with ORIGIN frames so far, in + /// no particular order. + /// + /// # Note + /// + /// Origins are only added while the connection is polled. At most 256 + /// distinct origins are kept, further ones are ignored. + pub fn received_origins(&self) -> Vec { + self.inner.received_origins().iter().cloned().collect() + } + + /// Returns whether the server announced `origin` with an ORIGIN frame. + /// + /// See [`received_origins`](Connection::received_origins). + pub fn has_received_origin(&self, origin: &str) -> bool { + self.inner.received_origins().contains(origin) + } + /// Returns the maximum number of concurrent streams that may be initiated /// by this client. /// diff --git a/src/codec/framed_read.rs b/src/codec/framed_read.rs index 1ec0e2c1d..b5c38a452 100644 --- a/src/codec/framed_read.rs +++ b/src/codec/framed_read.rs @@ -401,6 +401,14 @@ fn decode_frame(decoder: &mut FrameDecoder, mut bytes: BytesMut) -> Result { + // An ORIGIN frame on a stream other than 0x0 MUST be ignored. + if head.stream_id() != 0 { + return Ok(None); + } + + frame::Origin::load(head, &bytes[frame::HEADER_LEN..]).into() + } Kind::Unknown => { // Unknown frames are ignored return Ok(None); diff --git a/src/codec/framed_write.rs b/src/codec/framed_write.rs index efd1966cf..eaee24170 100644 --- a/src/codec/framed_write.rs +++ b/src/codec/framed_write.rs @@ -277,6 +277,10 @@ where v.encode(self.buf.get_mut()); tracing::trace!(rem = self.buf.remaining(), "encoded settings"); } + Frame::Origin(v) => { + v.encode(self.buf.get_mut()); + tracing::trace!(rem = self.buf.remaining(), "encoded origin"); + } Frame::GoAway(v) => { v.encode(self.buf.get_mut()); tracing::trace!(rem = self.buf.remaining(), "encoded go_away"); diff --git a/src/frame/head.rs b/src/frame/head.rs index 38be2f697..1c2999015 100644 --- a/src/frame/head.rs +++ b/src/frame/head.rs @@ -22,6 +22,7 @@ pub enum Kind { GoAway = 7, WindowUpdate = 8, Continuation = 9, + Origin = 12, Unknown, } @@ -88,6 +89,7 @@ impl Kind { 7 => Kind::GoAway, 8 => Kind::WindowUpdate, 9 => Kind::Continuation, + 12 => Kind::Origin, _ => Kind::Unknown, } } diff --git a/src/frame/mod.rs b/src/frame/mod.rs index 4a0c03036..bb0fe8ce1 100644 --- a/src/frame/mod.rs +++ b/src/frame/mod.rs @@ -39,6 +39,7 @@ mod data; mod go_away; mod head; mod headers; +mod origin; mod ping; mod priority; mod reason; @@ -54,6 +55,7 @@ pub use self::head::{Head, Kind}; pub use self::headers::{ parse_u64, Continuation, Headers, Pseudo, PushPromise, PushPromiseHeaderError, }; +pub use self::origin::Origin; pub use self::ping::Ping; pub use self::priority::{Priority, StreamDependency}; pub use self::reason::Reason; @@ -87,6 +89,7 @@ pub enum Frame { GoAway(GoAway), WindowUpdate(WindowUpdate), Reset(Reset), + Origin(Origin), } impl Frame { @@ -106,6 +109,7 @@ impl Frame { GoAway(frame) => frame.into(), WindowUpdate(frame) => frame.into(), Reset(frame) => frame.into(), + Origin(frame) => frame.into(), } } } @@ -124,6 +128,7 @@ impl fmt::Debug for Frame { GoAway(ref frame) => fmt::Debug::fmt(frame, fmt), WindowUpdate(ref frame) => fmt::Debug::fmt(frame, fmt), Reset(ref frame) => fmt::Debug::fmt(frame, fmt), + Origin(ref frame) => fmt::Debug::fmt(frame, fmt), } } } diff --git a/src/frame/origin.rs b/src/frame/origin.rs new file mode 100644 index 000000000..8810553ab --- /dev/null +++ b/src/frame/origin.rs @@ -0,0 +1,125 @@ +use bytes::BufMut; + +use crate::frame::{self, Head, Kind, StreamId}; + +#[derive(Debug, Clone, Eq, PartialEq)] +pub struct Origin { + origins: Vec, +} + +impl Origin { + /// Create a new ORIGIN frame. + /// + /// Entries that are not ASCII or too long to encode are skipped. + pub fn new(origins: I) -> Origin + where + I: IntoIterator, + S: Into, + { + Origin { + origins: Self::valid_origins(origins).collect(), + } + } + + /// Returns the origins that can be encoded in an ORIGIN frame. + pub(crate) fn valid_origins(origins: I) -> impl Iterator + where + I: IntoIterator, + S: Into, + { + origins + .into_iter() + .map(Into::into) + .filter(|origin| origin.is_ascii() && origin.len() <= u16::MAX as usize) + } + + pub fn into_origins(self) -> Vec { + self.origins + } + + /// Builds an `Origin` frame from a raw frame. + pub fn load(head: Head, mut payload: &[u8]) -> Origin { + debug_assert_eq!(head.kind(), crate::frame::Kind::Origin); + + let mut origins = Vec::new(); + + // Entries that are not ASCII are skipped, and so is a truncated + // entry at the end of the payload. + while payload.len() >= 2 { + let len = u16::from_be_bytes([payload[0], payload[1]]) as usize; + if payload.len() < 2 + len { + break; + } + let entry = &payload[2..2 + len]; + payload = &payload[2 + len..]; + if let Ok(origin) = std::str::from_utf8(entry) { + if origin.is_ascii() { + origins.push(origin.to_owned()); + } + } + } + + Origin { origins } + } + + pub fn encode(&self, dst: &mut B) { + let len = self.origins.iter().map(|origin| 2 + origin.len()).sum(); + tracing::trace!("encoding ORIGIN; len={}", len); + let head = Head::new(Kind::Origin, 0, StreamId::zero()); + head.encode(len, dst); + for origin in &self.origins { + dst.put_u16(origin.len() as u16); + dst.put_slice(origin.as_bytes()); + } + } +} + +impl From for frame::Frame { + fn from(src: Origin) -> Self { + frame::Frame::Origin(src) + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn encodes_the_entries() { + let mut buf = Vec::new(); + Origin::new([ + "https://a.example", + "https://b.example:8443", + "https://ä.example", + ]) + .encode(&mut buf); + + let mut want = vec![0, 0, 2 + 17 + 2 + 22, 0x0c, 0, 0, 0, 0, 0, 0, 17]; + want.extend_from_slice(b"https://a.example"); + want.extend_from_slice(&[0, 22]); + want.extend_from_slice(b"https://b.example:8443"); + assert_eq!(buf, want); + } + + #[test] + fn loads_the_entries() { + let mut buf = Vec::new(); + Origin::new(["https://a.example", "https://b.example:8443"]).encode(&mut buf); + // A truncated entry at the end is dropped. + buf.extend_from_slice(&[0, 9, b'a']); + + let head = Head::parse(&buf[..frame::HEADER_LEN]); + let origin = Origin::load(head, &buf[frame::HEADER_LEN..]); + assert_eq!( + origin.into_origins(), + ["https://a.example", "https://b.example:8443"] + ); + } + + #[test] + fn empty_frame() { + let mut buf = Vec::new(); + Origin::new(Vec::::new()).encode(&mut buf); + assert_eq!(buf, [0, 0, 0, 0x0c, 0, 0, 0, 0, 0]); + } +} diff --git a/src/proto/connection.rs b/src/proto/connection.rs index bb72d6068..171da4bde 100644 --- a/src/proto/connection.rs +++ b/src/proto/connection.rs @@ -7,6 +7,7 @@ use crate::proto::*; use bytes::Bytes; use futures_core::Stream; +use std::collections::HashSet; use std::io; use std::marker::PhantomData; use std::pin::Pin; @@ -14,6 +15,9 @@ use std::task::{Context, Poll}; use std::time::Duration; use tokio::io::AsyncRead; +// Limits the origins kept from a peer, further ones are ignored. +const MAX_RECEIVED_ORIGINS: usize = 256; + /// An H2 connection #[derive(Debug)] pub(crate) struct Connection @@ -48,6 +52,12 @@ where /// A refused stream to reset before receiving another frame. pending_refusal: Option, + /// Origins to send in an ORIGIN frame. + pending_origins: Vec, + + /// Origins announced by the peer in ORIGIN frames. + received_origins: HashSet, + /// Ping/pong handler ping_pong: PingPong, @@ -73,6 +83,9 @@ struct DynConnection<'a, B: Buf = Bytes> { error: &'a mut Option, + // Only a client keeps the ORIGIN frames of a server. + received_origins: Option<&'a mut HashSet>, + ping_pong: &'a mut PingPong, } @@ -164,6 +177,8 @@ where error: None, go_away: GoAway::new(), pending_refusal: None, + pending_origins: Vec::new(), + received_origins: HashSet::new(), ping_pong: PingPong::new(), settings: Settings::new(config.settings), streams, @@ -236,6 +251,30 @@ where self.inner.pending_refusal = None; } + let pending_origins = &mut self.inner.pending_origins; + while !pending_origins.is_empty() { + ready!(self.codec.poll_ready(cx))?; + // Split the origins into frames which fit the peer's max frame size. + let max = self.codec.max_send_frame_size(); + let mut size = 0; + let count = pending_origins + .iter() + .take_while(|origin| { + size += 2 + origin.len(); + size <= max + }) + .count(); + if count == 0 { + // This origin can never be sent. + pending_origins.remove(0); + continue; + } + let origins = pending_origins.drain(..count); + self.codec + .buffer(frame::Origin::new(origins).into()) + .expect("invalid ORIGIN frame"); + } + Poll::Ready(Ok(())) } @@ -424,6 +463,7 @@ where go_away, streams, error, + received_origins, ping_pong, .. } = self; @@ -433,6 +473,7 @@ where go_away, streams, error, + received_origins: (!P::r#dyn().is_server()).then_some(received_origins), ping_pong, } } @@ -616,6 +657,17 @@ where tracing::trace!(?frame, "recv PRIORITY"); // TODO: handle } + Some(Origin(frame)) => { + tracing::trace!(?frame, "recv ORIGIN"); + if let Some(received_origins) = &mut self.received_origins { + for origin in frame.into_origins() { + if received_origins.len() >= MAX_RECEIVED_ORIGINS { + break; + } + received_origins.insert(origin); + } + } + } None => { tracing::trace!("codec closed"); self.streams.recv_eof(false).expect("mutex poisoned"); @@ -641,6 +693,10 @@ where pub(crate) fn streams(&self) -> &Streams { &self.inner.streams } + + pub(crate) fn received_origins(&self) -> &HashSet { + &self.inner.received_origins + } } impl Connection @@ -652,6 +708,17 @@ where self.inner.streams.next_incoming() } + /// Queue an ORIGIN frame to be written. + pub fn send_origins(&mut self, origins: I) + where + I: IntoIterator, + S: Into, + { + self.inner + .pending_origins + .extend(frame::Origin::valid_origins(origins)); + } + // Graceful shutdown only makes sense for server peers. pub fn go_away_gracefully(&mut self) { if self.inner.go_away.is_going_away() { diff --git a/src/server.rs b/src/server.rs index 931661a19..8e369c549 100644 --- a/src/server.rs +++ b/src/server.rs @@ -550,6 +550,26 @@ where self.connection.go_away_gracefully(); } + /// Sends an ORIGIN frame ([RFC 8336]) listing the origins this connection + /// may be used for. + /// + /// Each origin is an ASCII serialization such as `https://example.com`. + /// Entries that can't be encoded are skipped. + /// + /// Origins added before the connection is next polled are sent together, + /// split into several frames if they exceed the peer's max frame size. + /// + /// Must continue being polled to write the frame. + /// + /// [RFC 8336]: https://www.rfc-editor.org/rfc/rfc8336 + pub fn send_origins(&mut self, origins: I) + where + I: IntoIterator, + S: Into, + { + self.connection.send_origins(origins); + } + /// Takes a `PingPong` instance from the connection. /// /// # Note diff --git a/tests/h2-support/src/frames.rs b/tests/h2-support/src/frames.rs index 5454ebd79..1d61b87d3 100644 --- a/tests/h2-support/src/frames.rs +++ b/tests/h2-support/src/frames.rs @@ -77,6 +77,14 @@ pub fn ping(payload: [u8; 8]) -> Mock { Mock(frame::Ping::new(payload)) } +pub fn origin(origins: I) -> Mock +where + I: IntoIterator, + S: Into, +{ + Mock(frame::Origin::new(origins)) +} + // === Generic helpers of all frame types pub struct Mock(T); diff --git a/tests/h2-tests/tests/origin.rs b/tests/h2-tests/tests/origin.rs new file mode 100644 index 000000000..b885a0140 --- /dev/null +++ b/tests/h2-tests/tests/origin.rs @@ -0,0 +1,247 @@ +use futures::channel::oneshot; +use futures::StreamExt; +use h2_support::prelude::*; + +#[tokio::test] +async fn server_sends_origins() { + h2_support::trace_init!(); + let (io, mut client) = mock::new(); + let (tx, rx) = oneshot::channel(); + + let client = async move { + let settings = client.assert_server_handshake().await; + assert_default_settings!(settings); + tx.send(()).unwrap(); + client + .recv_frame(frames::origin([ + "https://a.example", + "https://b.example:8443", + ])) + .await; + client.ping_pong([1; 8]).await; + }; + + let srv = async move { + let mut srv = server::handshake(io).await.expect("handshake"); + // The origins are queued once the handshake is done. + tokio::select! { + _ = rx => {} + _ = srv.next() => panic!("connection closed"), + } + srv.send_origins(["https://a.example", "https://b.example:8443"]); + assert!(srv.next().await.is_none()); + }; + + join(client, srv).await; +} + +#[tokio::test] +async fn server_origins_added_before_poll_share_a_frame() { + h2_support::trace_init!(); + let (io, mut client) = mock::new(); + let (tx, rx) = oneshot::channel(); + + let client = async move { + let settings = client.assert_server_handshake().await; + assert_default_settings!(settings); + tx.send(()).unwrap(); + client + .recv_frame(frames::origin(["https://a.example", "https://b.example"])) + .await; + client.ping_pong([1; 8]).await; + }; + + let srv = async move { + let mut srv = server::handshake(io).await.expect("handshake"); + // The origins are queued once the handshake is done. + tokio::select! { + _ = rx => {} + _ = srv.next() => panic!("connection closed"), + } + srv.send_origins(["https://a.example"]); + srv.send_origins(["https://b.example"]); + assert!(srv.next().await.is_none()); + }; + + join(client, srv).await; +} + +#[tokio::test] +async fn server_skips_origins_that_cannot_be_encoded() { + h2_support::trace_init!(); + let (io, mut client) = mock::new(); + + let client = async move { + let settings = client.assert_server_handshake().await; + assert_default_settings!(settings); + // No ORIGIN frame is written for an origin which is skipped. + client.ping_pong([1; 8]).await; + }; + + let srv = async move { + let mut srv = server::handshake(io).await.expect("handshake"); + srv.send_origins(["https://ä.example"]); + assert!(srv.next().await.is_none()); + }; + + join(client, srv).await; +} + +#[tokio::test] +async fn server_splits_origins_to_fit_max_frame_size() { + h2_support::trace_init!(); + let (io, mut client) = mock::new(); + let (tx, rx) = oneshot::channel(); + + let origins: Vec = (0..1000) + .map(|i| format!("https://{:04}.example", i)) + .collect(); + + let client = { + let origins = origins.clone(); + async move { + let settings = client.assert_server_handshake().await; + assert_default_settings!(settings); + tx.send(()).unwrap(); + + let mut frames = 0; + let mut received = Vec::new(); + while received.len() < origins.len() { + match client.next().await.unwrap().unwrap() { + frame::Frame::Origin(origin) => received.extend(origin.into_origins()), + frame => panic!("unexpected frame {:?}", frame), + } + frames += 1; + } + assert!(frames > 1); + assert_eq!(received, origins); + client.ping_pong([1; 8]).await; + } + }; + + let srv = async move { + let mut srv = server::handshake(io).await.expect("handshake"); + // The origins are queued once the handshake is done. + tokio::select! { + _ = rx => {} + _ = srv.next() => panic!("connection closed"), + } + srv.send_origins(origins); + assert!(srv.next().await.is_none()); + }; + + join(client, srv).await; +} + +#[tokio::test] +async fn server_ignores_origin_frame() { + h2_support::trace_init!(); + let (io, mut client) = mock::new(); + + let client = async move { + let settings = client.assert_server_handshake().await; + assert_default_settings!(settings); + client + .send_frame(frames::origin(["https://a.example"])) + .await; + client.ping_pong([1; 8]).await; + }; + + let srv = async move { + let mut srv = server::handshake(io).await.expect("handshake"); + assert!(srv.next().await.is_none()); + }; + + join(client, srv).await; +} + +#[tokio::test] +async fn client_reads_origins() { + h2_support::trace_init!(); + let (io, mut srv) = mock::new(); + + let srv = async move { + let _ = srv.assert_client_handshake().await; + srv.send_frame(frames::origin(["https://a.example", "https://b.example"])) + .await; + srv.send_frame(frames::origin(["https://b.example", "https://c.example"])) + .await; + }; + + let h2 = async move { + let (_client, mut conn) = client::handshake(io).await.unwrap(); + assert!(conn.received_origins().is_empty()); + assert!(!conn.has_received_origin("https://a.example")); + (&mut conn).await.unwrap(); + + let mut origins = conn.received_origins(); + origins.sort(); + assert_eq!( + origins, + [ + "https://a.example", + "https://b.example", + "https://c.example" + ] + ); + assert!(conn.has_received_origin("https://a.example")); + assert!(!conn.has_received_origin("https://d.example")); + }; + + join(srv, h2).await; +} + +#[tokio::test] +async fn client_caps_origins() { + h2_support::trace_init!(); + let (io, mut srv) = mock::new(); + + let origins: Vec = (0..300) + .map(|i| format!("https://{:03}.example", i)) + .collect(); + + let srv = { + let origins = origins.clone(); + async move { + let _ = srv.assert_client_handshake().await; + // Duplicates are only counted once. + srv.send_frame(frames::origin(origins.clone())).await; + srv.send_frame(frames::origin(origins)).await; + } + }; + + let h2 = async move { + let (_client, mut conn) = client::handshake(io).await.unwrap(); + (&mut conn).await.unwrap(); + + assert_eq!(conn.received_origins().len(), 256); + assert!(conn.has_received_origin(&origins[0])); + assert!(!conn.has_received_origin(&origins[299])); + }; + + join(srv, h2).await; +} + +#[tokio::test] +async fn client_ignores_origin_frame_on_a_stream() { + h2_support::trace_init!(); + let (io, mut srv) = mock::new(); + + let srv = async move { + let _ = srv.assert_client_handshake().await; + // An ORIGIN frame with stream id 1, which has to be ignored. + let mut frame = vec![0, 0, 19, 0x0c, 0, 0, 0, 0, 1, 0, 17]; + frame.extend_from_slice(b"https://a.example"); + srv.send_bytes(&frame).await; + srv.send_frame(frames::origin(["https://b.example"])).await; + }; + + let h2 = async move { + let (_client, mut conn) = client::handshake(io).await.unwrap(); + (&mut conn).await.unwrap(); + + assert_eq!(conn.received_origins(), ["https://b.example"]); + }; + + join(srv, h2).await; +}