feat: add support for ORIGIN frames (RFC 8336)

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<String> {
+        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.
     ///

src/codec/framed_read.rs

@@ -401,6 +401,14 @@ fn decode_frame(decoder: &mut FrameDecoder, mut bytes: BytesMut) -> Result<Optio
                 return Ok(None);
             }
         }
+        Kind::Origin => {
+            // 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);

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");

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,
         }
     }

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<T = Bytes> {
     GoAway(GoAway),
     WindowUpdate(WindowUpdate),
     Reset(Reset),
+    Origin(Origin),
 }
 
 impl<T> Frame<T> {
@@ -106,6 +109,7 @@ impl<T> Frame<T> {
             GoAway(frame) => frame.into(),
             WindowUpdate(frame) => frame.into(),
             Reset(frame) => frame.into(),
+            Origin(frame) => frame.into(),
         }
     }
 }
@@ -124,6 +128,7 @@ impl<T> fmt::Debug for Frame<T> {
             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),
         }
     }
 }

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<String>,
+}
+
+impl Origin {
+    /// Create a new ORIGIN frame.
+    ///
+    /// Entries that are not ASCII or too long to encode are skipped.
+    pub fn new<I, S>(origins: I) -> Origin
+    where
+        I: IntoIterator<Item = S>,
+        S: Into<String>,
+    {
+        Origin {
+            origins: Self::valid_origins(origins).collect(),
+        }
+    }
+
+    /// Returns the origins that can be encoded in an ORIGIN frame.
+    pub(crate) fn valid_origins<I, S>(origins: I) -> impl Iterator<Item = String>
+    where
+        I: IntoIterator<Item = S>,
+        S: Into<String>,
+    {
+        origins
+            .into_iter()
+            .map(Into::into)
+            .filter(|origin| origin.is_ascii() && origin.len() <= u16::MAX as usize)
+    }
+
+    pub fn into_origins(self) -> Vec<String> {
+        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<B: BufMut>(&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<B> From<Origin> for frame::Frame<B> {
+    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::<String>::new()).encode(&mut buf);
+        assert_eq!(buf, [0, 0, 0, 0x0c, 0, 0, 0, 0, 0]);
+    }
+}

src/proto/connection.rs

@@ -7,13 +7,17 @@ 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;
 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<T, P, B: Buf = Bytes>
@@ -48,6 +52,12 @@ where
     /// A refused stream to reset before receiving another frame.
     pending_refusal: Option<StreamId>,
 
+    /// Origins to send in an ORIGIN frame.
+    pending_origins: Vec<String>,
+
+    /// Origins announced by the peer in ORIGIN frames.
+    received_origins: HashSet<String>,
+
     /// Ping/pong handler
     ping_pong: PingPong,
 
@@ -73,6 +83,9 @@ struct DynConnection<'a, B: Buf = Bytes> {
 
     error: &'a mut Option<frame::GoAway>,
 
+    // Only a client keeps the ORIGIN frames of a server.
+    received_origins: Option<&'a mut HashSet<String>>,
+
     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<B, client::Peer> {
         &self.inner.streams
     }
+
+    pub(crate) fn received_origins(&self) -> &HashSet<String> {
+        &self.inner.received_origins
+    }
 }
 
 impl<T, B> Connection<T, server::Peer, B>
@@ -652,6 +708,17 @@ where
         self.inner.streams.next_incoming()
     }
 
+    /// Queue an ORIGIN frame to be written.
+    pub fn send_origins<I, S>(&mut self, origins: I)
+    where
+        I: IntoIterator<Item = S>,
+        S: Into<String>,
+    {
+        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() {