feat(iroh-relay)!: Implement new handshake protocol, refactor frame types (#3331)

## Description

This PR changes the iroh relay handshake and relaying protocol.
The main goals are:
- Actually authenticate the client by verifying a signed challenge
(previously only the client info was signed)
- We also allow using TLS extracted keying material as the subject to
sign and sending it in an HTTP header, if that is available. When this
works (it does so most of the time), it's 1RTT faster.
- Use a better encoding for messages (we now use QUIC varints for frame
type, use websocket framing for the protocol)
- Remove outdated and unused frames (KeepAlive, NotePreferred)

There's also some secondary changes:
- Conceptually, the relay handshake and relaying protocols are now
split: There's a `protos::handshake` module for the handshake, and a
`protos::relay` for the send/recv protocol.
- I've refactored the streams we have in iroh-relay. We now use this
stack:
- On the client, we establish TLS streams using
`MaybeTlsStream<ProxyStream>`. `MaybeTlsStream` gives us the TLS
handshake and `ProxyStream` gives us HTTPS proxy support.
- On the server we establish rate-limited TLS streams using
`RateLimit<MaybeTlsStream>`.
- On top of these two, we stack `WsBytesFramed`, which internally uses
`tokio_websockets::WebSocketStream` natively and on the server and
`ws_stream_wasm::WsStream` in browsers. This wrapper does the websocket
framing and translates opaque `Bytes` messages into websocket bytes
frames.
- This `WsBytesFramed` is then directly used with the `handshake`
protocol on the server and client side.
- For the `send_recv` protocol, we then upgrade the server side to
`RelayedStream` and the client side to the `Conn` type.
- Instead of having `Frame`, `ReceivedMessage` and `SendMessage` as
types, we now only have `RelayToClientMsg` and `ClientToRelayMsg` for
the send/recv protocol.

## Breaking Changes

In iroh:
- Connections to older relays don't work anymore.

In iroh-relay:
- Removed `iroh_relay::client::SendMessage` and
`iroh_relay::client::ReceivedMessage` in favor of `ClientToRelayMsg` and
`RelayToClientMsg` respectively.
- `impl Stream for Client` now produces `RelayToClientMsg` instead of
`ReceivedMessage`
- `Client` now `impl Sink<ClientToRelayMsg>` instead of `impl
Sink<SendMessage>`
- Removed `ClientBuilder::is_prober`
- Moved `protos::relay::FrameType` to `protos::common::FrameType` and
adjusted frame types to those of the current set of protocols

## Change checklist
<!-- Remove any that are not relevant. -->
- [x] Self-review.
- [x] Documentation updates following the [style
guide](https://rust-lang.github.io/rfcs/1574-more-api-documentation-conventions.html#appendix-a-full-conventions-text),
if relevant.
- [x] Tests if relevant.
- [x] All breaking changes documented.

Author: matheus23

Cargo.lock

@@ -95,6 +95,8 @@ toml = { version = "0.8", optional = true }
 tracing-subscriber = { version = "0.3", features = [
     "env-filter",
 ], optional = true }
+blake3 = "1.8.2"
+serde_bytes = "0.11.17"
 
 # non-wasm-in-browser dependencies
 [target.'cfg(not(all(target_family = "wasm", target_os = "unknown")))'.dependencies]

iroh-relay/Cargo.toml

@@ -5,3 +5,4 @@
 # It is recommended to check this file in to source control so that
 # everyone who runs the test benefits from these saved cases.
 cc 9295f5287162dfb180e5826e563c2cea08b477b803ef412ff8351eb5c3eb45ef # shrinks to frame = KeepAlive
+cc 753aabcf8ae2b4e4a52f451d58339aab85a4b61108afdf4b9600f97b3a33bf42 # shrinks to frame = Health { problem: None }

iroh-relay/proptest-regressions/protos/relay.txt

@@ -0,0 +1,9 @@
+# Seeds for failure cases proptest has generated in the past. It is
+# automatically read and these particular cases re-run before any
+# novel cases are generated.
+#
+# It is recommended to check this file in to source control so that
+# everyone who runs the test benefits from these saved cases.
+cc 8f4f94b7c917bb0f52d31b529c3d580728ea57954ca45d91768cf4ae745e6eb9 # shrinks to frame = ReceivedDatagrams { remote_node_id: PublicKey(3b6a27bcceb6a42d62a3a8d02a6f0d73653215771de243a63ac048a18b59da29), datagrams: Datagrams { ecn: None, segment_size: Some(43846), .. } }
+cc e40ca61e22386f1c76f717f2a6dbba367ea05d906317b3f979b46031567edbca # shrinks to frame = SendDatagrams { dst_node_id: PublicKey(3b6a27bcceb6a42d62a3a8d02a6f0d73653215771de243a63ac048a18b59da29), datagrams: Datagrams { ecn: None, segment_size: Some(44811), .. } }
+cc 435ec32fc803db22bf4688a6356878073752b58fcd0b4422876fb3ab2a622684 # shrinks to a huge frame = Health { .. }

iroh-relay/proptest-regressions/protos/send_recv.txt

@@ -22,10 +22,17 @@ use tracing::warn;
 use tracing::{debug, event, trace, Level};
 use url::Url;
 
-pub use self::conn::{ReceivedMessage, RecvError, SendError, SendMessage};
+pub use self::conn::{RecvError, SendError};
 #[cfg(not(wasm_browser))]
 use crate::dns::{DnsError, DnsResolver};
-use crate::{http::RELAY_PATH, protos::relay::SendError as SendRelayError, KeyCache};
+use crate::{
+    http::RELAY_PATH,
+    protos::{
+        handshake,
+        relay::{ClientToRelayMsg, RelayToClientMsg},
+    },
+    KeyCache,
+};
 
 pub(crate) mod conn;
 #[cfg(not(wasm_browser))]
@@ -60,7 +67,7 @@ pub enum ConnectError {
         source: ws_stream_wasm::WsErr,
     },
     #[snafu(transparent)]
-    Handshake { source: SendRelayError },
+    Handshake { source: handshake::Error },
     #[snafu(transparent)]
     Dial { source: DialError },
     #[snafu(display("Unexpected status during upgrade: {code}"))]
@@ -117,8 +124,6 @@ pub struct ClientBuilder {
     /// Default is None
     #[debug("address family selector callback")]
     address_family_selector: Option<Arc<dyn Fn() -> bool + Send + Sync>>,
-    /// Default is false
-    is_prober: bool,
     /// Server url.
     url: RelayUrl,
     /// Allow self-signed certificates from relay servers
@@ -144,7 +149,6 @@ impl ClientBuilder {
     ) -> Self {
         ClientBuilder {
             address_family_selector: None,
-            is_prober: false,
             url: url.into(),
 
             #[cfg(any(test, feature = "test-utils"))]
@@ -172,12 +176,6 @@ impl ClientBuilder {
         self
     }
 
-    /// Indicates this client is a prober
-    pub fn is_prober(mut self, is: bool) -> Self {
-        self.is_prober = is;
-        self
-    }
-
     /// Skip the verification of the relay server's SSL certificates.
     ///
     /// May only be used in tests.
@@ -202,9 +200,13 @@ impl ClientBuilder {
     /// Establishes a new connection to the relay server.
     #[cfg(not(wasm_browser))]
     pub async fn connect(&self) -> Result<Client, ConnectError> {
+        use http::header::SEC_WEBSOCKET_PROTOCOL;
         use tls::MaybeTlsStreamBuilder;
 
-        use crate::protos::relay::MAX_FRAME_SIZE;
+        use crate::{
+            http::{CLIENT_AUTH_HEADER, RELAY_PROTOCOL_VERSION},
+            protos::{handshake::KeyMaterialClientAuth, relay::MAX_FRAME_SIZE},
+        };
 
         let mut dial_url = (*self.url).clone();
         dial_url.set_path(RELAY_PATH);
@@ -240,17 +242,33 @@ impl ClientBuilder {
             .as_ref()
             .local_addr()
             .map_err(|_| NoLocalAddrSnafu.build())?;
-        let (conn, response) = tokio_websockets::ClientBuilder::new()
+        let mut builder = tokio_websockets::ClientBuilder::new()
             .uri(dial_url.as_str())
             .map_err(|_| {
                 InvalidRelayUrlSnafu {
                     url: dial_url.clone(),
                 }
                 .build()
             })?
+            .add_header(
+                SEC_WEBSOCKET_PROTOCOL,
+                http::HeaderValue::from_static(RELAY_PROTOCOL_VERSION),
+            )
+            .expect("valid header name and value")
             .limits(tokio_websockets::Limits::default().max_payload_len(Some(MAX_FRAME_SIZE)))
-            .connect_on(stream)
-            .await?;
+            // We turn off automatic flushing after a threshold (the default would be after 8KB).
+            // This means we need to flush manually, which we do by calling `Sink::send_all` or
+            // `Sink::send` (which calls `Sink::flush`) in the `ActiveRelayActor`.
+            .config(tokio_websockets::Config::default().flush_threshold(usize::MAX));
+        if let Some(client_auth) = KeyMaterialClientAuth::new(&self.secret_key, &stream) {
+            debug!("Using TLS key export for relay client authentication");
+            builder = builder
+                .add_header(CLIENT_AUTH_HEADER, client_auth.into_header_value())
+                .expect(
+                    "impossible: CLIENT_AUTH_HEADER isn't a disallowed header value for websockets",
+                );
+        }
+        let (conn, response) = builder.connect_on(stream).await?;
 
         if response.status() != hyper::StatusCode::SWITCHING_PROTOCOLS {
             UnexpectedUpgradeStatusSnafu {
@@ -291,6 +309,8 @@ impl ClientBuilder {
     /// Establishes a new connection to the relay server.
     #[cfg(wasm_browser)]
     pub async fn connect(&self) -> Result<Client, ConnectError> {
+        use crate::http::RELAY_PROTOCOL_VERSION;
+
         let mut dial_url = (*self.url).clone();
         dial_url.set_path(RELAY_PATH);
         // The relay URL is exchanged with the http(s) scheme in tickets and similar.
@@ -310,7 +330,9 @@ impl ClientBuilder {
 
         debug!(%dial_url, "Dialing relay by websocket");
 
-        let (_, ws_stream) = ws_stream_wasm::WsMeta::connect(dial_url.as_str(), None).await?;
+        let (_, ws_stream) =
+            ws_stream_wasm::WsMeta::connect(dial_url.as_str(), Some(vec![RELAY_PROTOCOL_VERSION]))
+                .await?;
         let conn = Conn::new(ws_stream, self.key_cache.clone(), &self.secret_key).await?;
 
         event!(
@@ -350,49 +372,49 @@ impl Client {
 }
 
 impl Stream for Client {
-    type Item = Result<ReceivedMessage, RecvError>;
+    type Item = Result<RelayToClientMsg, RecvError>;
 
     fn poll_next(mut self: Pin<&mut Self>, cx: &mut task::Context<'_>) -> Poll<Option<Self::Item>> {
         Pin::new(&mut self.conn).poll_next(cx)
     }
 }
 
-impl Sink<SendMessage> for Client {
+impl Sink<ClientToRelayMsg> for Client {
     type Error = SendError;
 
     fn poll_ready(
         mut self: Pin<&mut Self>,
         cx: &mut task::Context<'_>,
     ) -> Poll<Result<(), Self::Error>> {
-        <Conn as Sink<SendMessage>>::poll_ready(Pin::new(&mut self.conn), cx)
+        Pin::new(&mut self.conn).poll_ready(cx)
     }
 
-    fn start_send(mut self: Pin<&mut Self>, item: SendMessage) -> Result<(), Self::Error> {
+    fn start_send(mut self: Pin<&mut Self>, item: ClientToRelayMsg) -> Result<(), Self::Error> {
         Pin::new(&mut self.conn).start_send(item)
     }
 
     fn poll_flush(
         mut self: Pin<&mut Self>,
         cx: &mut task::Context<'_>,
     ) -> Poll<Result<(), Self::Error>> {
-        <Conn as Sink<SendMessage>>::poll_flush(Pin::new(&mut self.conn), cx)
+        Pin::new(&mut self.conn).poll_flush(cx)
     }
 
     fn poll_close(
         mut self: Pin<&mut Self>,
         cx: &mut task::Context<'_>,
     ) -> Poll<Result<(), Self::Error>> {
-        <Conn as Sink<SendMessage>>::poll_close(Pin::new(&mut self.conn), cx)
+        Pin::new(&mut self.conn).poll_close(cx)
     }
 }
 
 /// The send half of a relay client.
 #[derive(Debug)]
 pub struct ClientSink {
-    sink: SplitSink<Conn, SendMessage>,
+    sink: SplitSink<Conn, ClientToRelayMsg>,
 }
 
-impl Sink<SendMessage> for ClientSink {
+impl Sink<ClientToRelayMsg> for ClientSink {
     type Error = SendError;
 
     fn poll_ready(
@@ -402,7 +424,7 @@ impl Sink<SendMessage> for ClientSink {
         Pin::new(&mut self.sink).poll_ready(cx)
     }
 
-    fn start_send(mut self: Pin<&mut Self>, item: SendMessage) -> Result<(), Self::Error> {
+    fn start_send(mut self: Pin<&mut Self>, item: ClientToRelayMsg) -> Result<(), Self::Error> {
         Pin::new(&mut self.sink).start_send(item)
     }
 
@@ -436,7 +458,7 @@ impl ClientStream {
 }
 
 impl Stream for ClientStream {
-    type Item = Result<ReceivedMessage, RecvError>;
+    type Item = Result<RelayToClientMsg, RecvError>;
 
     fn poll_next(mut self: Pin<&mut Self>, cx: &mut task::Context<'_>) -> Poll<Option<Self::Item>> {
         Pin::new(&mut self.stream).poll_next(cx)