diff --git a/easytier-core/src/connectivity/direct/mod.rs b/easytier-core/src/connectivity/direct/mod.rs index e4325630..35379bc1 100644 --- a/easytier-core/src/connectivity/direct/mod.rs +++ b/easytier-core/src/connectivity/direct/mod.rs @@ -1,14 +1,12 @@ use std::{ collections::HashSet, - hash::Hash, net::{IpAddr, Ipv4Addr, Ipv6Addr, SocketAddr}, sync::{Arc, Weak}, - time::{Duration, Instant as StdInstant}, + time::Duration, }; use anyhow::Context; use async_trait::async_trait; -use dashmap::DashMap; use quanta::Instant; use rand::Rng; use serde::{Deserialize, Serialize}; @@ -26,6 +24,7 @@ use crate::{ }, transport::{self, ConnectedTransport, UdpSessionMode}, }, + foundation::expiring_set::ExpiringSet, foundation::task::{PeerTaskLauncher, PeerTaskManager}, host::dns::DnsResolver, peers::{ @@ -157,50 +156,6 @@ impl DirectConnectorOptions { } } -#[derive(Debug)] -struct ExpiringSet -where - K: Eq + Hash, -{ - entries: DashMap, -} - -impl Default for ExpiringSet -where - K: Eq + Hash, -{ - fn default() -> Self { - Self { - entries: DashMap::new(), - } - } -} - -impl ExpiringSet -where - K: Eq + Hash + Clone, -{ - fn insert(&self, key: K, ttl: Duration) { - self.entries.insert(key, StdInstant::now() + ttl); - } - - fn contains(&self, key: &K) -> bool { - let active = self - .entries - .get(key) - .is_some_and(|expires_at| *expires_at > StdInstant::now()); - if !active { - self.entries.remove(key); - } - active - } - - fn cleanup(&self) { - let now = StdInstant::now(); - self.entries.retain(|_, expires_at| *expires_at > now); - } -} - #[derive(Debug, Hash, Eq, PartialEq, Clone)] struct ListenerBlacklistKey(PeerId, String); diff --git a/easytier-core/src/foundation/expiring_set.rs b/easytier-core/src/foundation/expiring_set.rs new file mode 100644 index 00000000..9cf10f04 --- /dev/null +++ b/easytier-core/src/foundation/expiring_set.rs @@ -0,0 +1,71 @@ +use std::{hash::Hash, time::Duration, time::Instant}; + +use dashmap::DashMap; + +/// A thread-safe set whose entries expire after a per-insert TTL. +/// +/// `contains` lazily removes expired entries, so periodic `cleanup` calls +/// are only needed to reclaim memory for keys that stop being read. +#[derive(Debug, Clone)] +pub struct ExpiringSet +where + K: Eq + Hash, +{ + entries: DashMap, +} + +impl Default for ExpiringSet +where + K: Eq + Hash, +{ + fn default() -> Self { + Self { + entries: DashMap::new(), + } + } +} + +impl ExpiringSet +where + K: Eq + Hash + Clone, +{ + pub fn insert(&self, key: K, ttl: Duration) { + self.entries.insert(key, Instant::now() + ttl); + } + + pub fn contains(&self, key: &K) -> bool { + match self + .entries + .remove_if(key, |_, expires_at| *expires_at <= Instant::now()) + { + // Existed and expired: removed while holding the shard lock, so a + // concurrent insert of the same key cannot be dropped by us. + Some(_) => false, + // Not removed: either absent, or still fresh. + None => self.entries.contains_key(key), + } + } + + pub fn cleanup(&self) { + let now = Instant::now(); + self.entries.retain(|_, expires_at| *expires_at > now); + } +} + +#[cfg(test)] +mod tests { + use std::time::Duration; + + use super::*; + + #[test] + fn expired_entries_are_reported_absent() { + let set: ExpiringSet = ExpiringSet::default(); + set.insert(1, Duration::ZERO); + set.insert(2, Duration::from_secs(3600)); + assert!(!set.contains(&1)); + assert!(set.contains(&2)); + set.cleanup(); + assert!(!set.entries.contains_key(&1)); + } +} diff --git a/easytier-core/src/foundation/mod.rs b/easytier-core/src/foundation/mod.rs index effe2897..e1293aed 100644 --- a/easytier-core/src/foundation/mod.rs +++ b/easytier-core/src/foundation/mod.rs @@ -3,6 +3,7 @@ //! Everything in `foundation` may be used by any layer, and nothing here may //! depend on a domain Module. See `CONTEXT.md` "Module layers". +pub mod expiring_set; #[cfg(any( feature = "proxy-smoltcp-stack", test, diff --git a/easytier/src/gateway/quic_proxy.rs b/easytier/src/gateway/quic_proxy.rs index ab06d1ce..060a8889 100644 --- a/easytier/src/gateway/quic_proxy.rs +++ b/easytier/src/gateway/quic_proxy.rs @@ -1,6 +1,8 @@ -use super::hedge::HedgeExt; +use super::hedge::{ErrorCollection, HedgeExt}; use crate::proto::peer_rpc::KcpConnData as QuicConnData; -use crate::tunnel::quic::{client_config, endpoint_config, server_config}; +use crate::tunnel::quic::{ + QUIC_VERSION_ETQ1, client_config, endpoint_config, etq1_client_config, server_config, +}; use anyhow::{Context, Error, anyhow, ensure}; use atomic_refcell::AtomicRefCell; use bytes::{BufMut, Bytes, BytesMut}; @@ -15,6 +17,7 @@ use quinn::{ WriteError, default_runtime, }; use std::cmp::min; +use std::future::Future; use std::io::IoSliceMut; use std::net::{IpAddr, Ipv4Addr, SocketAddr}; use std::pin::Pin; @@ -33,6 +36,7 @@ use tokio_util::sync::{CancellationToken, PollSender}; use tracing::{debug, error, info, instrument, trace, warn}; use easytier_core::{ + foundation::expiring_set::ExpiringSet, gateway::proxy::traits::TcpProxyStream, gateway::proxy::wrapped_transport::{ WrappedTransportAcceptedStream, WrappedTransportConnect, WrappedTransportDatagram, @@ -261,13 +265,89 @@ impl From<(SendStream, RecvStream)> for QuicStream { } //endregion +/// How long to keep dialing a peer with legacy QUIC version 1 after it +/// rejected `QUIC_VERSION_ETQ1`, before probing ETQ1 again. Bounds the set +/// and lets peers that upgrade mid-flight pick up the fix. +const LEGACY_VERSION_TTL: Duration = Duration::from_secs(3600); + #[derive(Debug, Clone)] pub struct NatDstQuicConnector { pub(crate) endpoint: Endpoint, pub(crate) conn_map: Cache, + /// Peers that recently rejected `QUIC_VERSION_ETQ1` via version + /// negotiation and are dialed with legacy version 1 until the entry + /// expires. + pub(crate) legacy_version_peers: Arc>, } impl NatDstQuicConnector { + fn connect_hedge( + &self, + dst_peer: PeerId, + version: u32, + ) -> impl Future>> + '_ { + let endpoint = self.endpoint.clone(); + let legacy_version_peers = self.legacy_version_peers.clone(); + (0..5) + .map(move |_| { + let endpoint = endpoint.clone(); + let legacy_version_peers = legacy_version_peers.clone(); + async move { + let config = if version == QUIC_VERSION_ETQ1 { + etq1_client_config() + } else { + client_config() + }; + let ret = endpoint + .connect_with( + config, + QuicAddr::new(dst_peer, PacketType::QuicSrc).into(), + "", + ) + .context("failed to create connection")? + .await + .context("connection failed"); + if let Some(ConnectionError::VersionMismatch) = + ret.as_ref().err().and_then(|e| e.downcast_ref()) + { + // Remote only supports legacy version 1; remember it + // so later connects skip the rejected version for a + // while. + legacy_version_peers.insert(dst_peer, LEGACY_VERSION_TTL); + } + ret + } + }) + .hedge(Duration::from_millis(200)) + } + + async fn connect(&self, dst_peer: PeerId) -> anyhow::Result { + self.conn_map.invalidate(&dst_peer).await; + self.legacy_version_peers.cleanup(); + + if !self.legacy_version_peers.contains(&dst_peer) { + let result = self + .conn_map + .try_get_with(dst_peer, self.connect_hedge(dst_peer, QUIC_VERSION_ETQ1)) + .await; + match result { + Ok(conn) => return Ok(conn), + Err(_) if self.legacy_version_peers.contains(&dst_peer) => { + debug!( + ?dst_peer, + "remote rejected ETQ1, falling back to quic version 1" + ); + } + Err(errors) => return Err(anyhow!("failed to connect to peer: {errors}")), + } + } + + self.conn_map + .try_get_with(dst_peer, self.connect_hedge(dst_peer, 1)) + .await + .map_err(|errors| anyhow!("failed to connect to peer: {errors}")) + } + async fn connect_to_peer( &self, dst_peer: PeerId, @@ -293,27 +373,7 @@ impl NatDstQuicConnector { buf.freeze() }; - let reconnect = || async move { - self.conn_map.invalidate(&dst_peer).await; - - let connect = (0..5) - .map(|_| { - let endpoint = self.endpoint.clone(); - async move { - endpoint - .connect(QuicAddr::new(dst_peer, PacketType::QuicSrc).into(), "") - .context("failed to create connection")? - .await - .context("connection failed") - } - }) - .hedge(Duration::from_millis(200)); - - self.conn_map - .try_get_with(dst_peer, connect) - .await - .context("failed to connect to peer") - }; + let reconnect = || async move { self.connect(dst_peer).await }; let mut reconnected = false; @@ -664,6 +724,9 @@ impl QuicProxy { default_runtime().unwrap(), ) .unwrap(); // TODO: maybe a different transport config + // Default stays on legacy version 1; ETQ1 is negotiated per attempt in + // NatDstQuicConnector so that peers which only speak version 1 keep + // working. endpoint.set_default_client_config(client_config()); self.endpoint = Some(endpoint.clone()); @@ -690,6 +753,7 @@ impl QuicProxy { .max_capacity(u8::MAX.into()) // cf. quinn transport config (max_concurrent_bidi_streams) .time_to_idle(Duration::from_secs(600)) // cf. quinn transport config (max_idle_timeout) .build(), + legacy_version_peers: Arc::new(ExpiringSet::default()), }); } @@ -845,6 +909,9 @@ mod tests { use super::*; use bytes::Buf; use quanta::Instant; + use quinn::EndpointConfig; + + use crate::tunnel::quic::{connect_with_etq1, endpoint_config_with_versions}; /// Helper function: Create a pair of interconnected QuicSockets. /// Data sent by socket_a will enter socket_b's rx, and vice versa. @@ -1262,4 +1329,183 @@ mod tests { let chunk3_data = &payload[chunk3_start..chunk3_start + segment_size]; assert_eq!(chunk3_data[0], 3u8, "Chunk 3 corrupted"); } + + fn endpoint_pair_with_configs( + client_endpoint_config: EndpointConfig, + server_endpoint_config: EndpointConfig, + ) -> (Endpoint, Endpoint) { + let server_config = server_config(); + let client_config = client_config(); + + let (socket_client, socket_server) = make_socket_pair(); + + let mut client_endpoint = Endpoint::new_with_abstract_socket( + client_endpoint_config, + Some(server_config.clone()), + Arc::new(socket_client), + default_runtime().unwrap(), + ) + .unwrap(); + client_endpoint.set_default_client_config(client_config.clone()); + + let mut server_endpoint = Endpoint::new_with_abstract_socket( + server_endpoint_config, + Some(server_config), + Arc::new(socket_server), + default_runtime().unwrap(), + ) + .unwrap(); + server_endpoint.set_default_client_config(client_config); + + (client_endpoint, server_endpoint) + } + + async fn assert_stream_roundtrip(connection: &Connection) -> anyhow::Result<()> { + let (mut send, mut recv) = connection.open_bi().await?; + send.write_all(b"ping").await?; + send.finish()?; + let mut buf = vec![0u8; 4]; + recv.read_exact(&mut buf).await?; + assert_eq!(&buf, b"ping"); + Ok(()) + } + + #[tokio::test] + async fn etq1_handshake_streams_both_ways() -> anyhow::Result<()> { + // New client and new server negotiate ETQ1 and echo a stream in both + // directions, exercising the packet-number-bound checksum on + // Initial, Handshake and 1-RTT packets. + let (client_endpoint, server_endpoint) = + endpoint_pair_with_configs(endpoint_config(), endpoint_config()); + let server_addr = server_endpoint.local_addr()?; + + let server = tokio::spawn(async move { + let Some(incoming) = server_endpoint.accept().await else { + panic!("no incoming connection"); + }; + let connection = incoming.await.unwrap(); + let (mut send, mut recv) = connection.accept_bi().await.unwrap(); + let mut buf = vec![0u8; 4]; + recv.read_exact(&mut buf).await.unwrap(); + send.write_all(&buf).await.unwrap(); + send.finish().unwrap(); + let _ = connection.closed().await; + }); + + let connection = client_endpoint + .connect_with(etq1_client_config(), server_addr, "localhost")? + .await?; + + let (mut send, mut recv) = connection.open_bi().await?; + send.write_all(b"ping").await?; + send.finish()?; + let mut buf = vec![0u8; 4]; + recv.read_exact(&mut buf).await?; + assert_eq!(&buf, b"ping"); + + connection.close(0u32.into(), b"done"); + let _ = tokio::time::timeout(Duration::from_secs(2), server).await; + Ok(()) + } + + #[tokio::test] + async fn etq1_against_legacy_server_falls_back_to_version_1() -> anyhow::Result<()> { + // A new client dialing a legacy (version 1 only) server must see + // VersionMismatch for ETQ1 and succeed after falling back to the + // legacy client config, mirroring NatDstQuicConnector::connect. + let (client_endpoint, server_endpoint) = + endpoint_pair_with_configs(endpoint_config(), endpoint_config_with_versions(vec![1])); + let server_addr = server_endpoint.local_addr()?; + + let server = tokio::spawn(async move { + let Some(incoming) = server_endpoint.accept().await else { + return; + }; + let connection = incoming.await.unwrap(); + if let Ok((mut send, mut recv)) = connection.accept_bi().await { + let mut buf = vec![0u8; 4]; + recv.read_exact(&mut buf).await.unwrap(); + send.write_all(&buf).await.unwrap(); + send.finish().unwrap(); + } + let _ = connection.closed().await; + }); + + let err = client_endpoint + .connect_with(etq1_client_config(), server_addr, "localhost")? + .await + .expect_err("legacy server must reject ETQ1"); + assert!( + matches!(err, ConnectionError::VersionMismatch), + "unexpected error: {err:?}" + ); + + let connection = client_endpoint + .connect_with(client_config(), server_addr, "localhost")? + .await?; + assert_stream_roundtrip(&connection).await?; + + connection.close(0u32.into(), b"done"); + let _ = tokio::time::timeout(Duration::from_secs(2), server).await; + Ok(()) + } + + #[tokio::test] + async fn legacy_client_against_etq1_server() -> anyhow::Result<()> { + // An old client keeps working against a new (dual version) server. + let (client_endpoint, server_endpoint) = + endpoint_pair_with_configs(endpoint_config_with_versions(vec![1]), endpoint_config()); + let server_addr = server_endpoint.local_addr()?; + + let server = tokio::spawn(async move { + let Some(incoming) = server_endpoint.accept().await else { + return; + }; + let connection = incoming.await.unwrap(); + if let Ok((mut send, mut recv)) = connection.accept_bi().await { + let mut buf = vec![0u8; 4]; + recv.read_exact(&mut buf).await.unwrap(); + send.write_all(&buf).await.unwrap(); + send.finish().unwrap(); + } + let _ = connection.closed().await; + }); + + let connection = client_endpoint.connect(server_addr, "localhost")?.await?; + assert_stream_roundtrip(&connection).await?; + + connection.close(0u32.into(), b"done"); + let _ = tokio::time::timeout(Duration::from_secs(2), server).await; + Ok(()) + } + #[tokio::test] + async fn connect_with_etq1_falls_back_for_legacy_server() -> anyhow::Result<()> { + // The shared ETQ1-first dial helper used by the quic:// tunnel: a + // legacy server rejects ETQ1 via version negotiation and the helper + // must transparently fall back to version 1. + let (client_endpoint, server_endpoint) = + endpoint_pair_with_configs(endpoint_config(), endpoint_config_with_versions(vec![1])); + let server_addr = server_endpoint.local_addr()?; + + let server = tokio::spawn(async move { + let Some(incoming) = server_endpoint.accept().await else { + return; + }; + let connection = incoming.await.unwrap(); + if let Ok((mut send, mut recv)) = connection.accept_bi().await { + let mut buf = vec![0u8; 4]; + recv.read_exact(&mut buf).await.unwrap(); + send.write_all(&buf).await.unwrap(); + send.finish().unwrap(); + } + let _ = connection.closed().await; + }); + + let connection = connect_with_etq1(&client_endpoint, server_addr, "localhost").await?; + assert_stream_roundtrip(&connection).await?; + + connection.close(0u32.into(), b"done"); + let _ = tokio::time::timeout(Duration::from_secs(2), server).await; + Ok(()) + } } diff --git a/easytier/src/tunnel/quic.rs b/easytier/src/tunnel/quic.rs index a4ba36b4..339b82a2 100644 --- a/easytier/src/tunnel/quic.rs +++ b/easytier/src/tunnel/quic.rs @@ -17,8 +17,9 @@ use easytier_core::{ }, }; use quinn::{ - AsyncUdpSocket, ClientConfig, Connecting, Connection, Endpoint, EndpointConfig, Incoming, - ServerConfig, TransportConfig, congestion::BbrConfig, default_runtime, + AsyncUdpSocket, ClientConfig, Connecting, Connection, ConnectionError, Endpoint, + EndpointConfig, Incoming, ServerConfig, TransportConfig, congestion::BbrConfig, + default_runtime, }; use std::{net::SocketAddr, sync::Arc, time::Duration}; use tokio::{ @@ -35,6 +36,7 @@ pub(crate) use session_socket::QuicUdpSessionSocket; // region config mod crypto { + use crate::tunnel::quic::QUIC_VERSION_ETQ1; use crate::utils::BoxExt; use bytes::{Buf, BytesMut}; use quinn_proto::crypto::{ @@ -52,9 +54,20 @@ mod crypto { use tracing::{error, instrument, trace}; #[derive(Debug, Clone, Copy)] - struct CryptoKey; + struct CryptoKey { + /// Bind the packet checksum to the packet number so that a truncated + /// packet number decoded out of window (RFC 9000 Appendix A) fails + /// authentication instead of acknowledging an unsent packet number. + pn_bound: bool, + } impl CryptoKey { + fn for_version(version: u32) -> Self { + Self { + pn_bound: version == QUIC_VERSION_ETQ1, + } + } + fn header(self) -> KeyPair> { KeyPair { local: Box::new(self), @@ -94,6 +107,28 @@ mod crypto { } hasher.finish() } + + // The packet number is mixed into the checksum before the slices. Real + // QUIC derives the AEAD nonce from the packet number, so a number + // decoded differently from how it was encoded fails authentication; + // this mirrors that property for the SeaHash integrity check. + fn checksum_with_packet(packet: u64, slices: &[&[u8]]) -> u64 { + let mut hasher = SeaHasher::default(); + hasher.write(&packet.to_le_bytes()); + for slice in slices { + hasher.write(&(slice.len() as u64).to_le_bytes()); + hasher.write(slice); + } + hasher.finish() + } + + fn checksum_for(&self, packet: u64, slices: &[&[u8]]) -> u64 { + if self.pn_bound { + Self::checksum_with_packet(packet, slices) + } else { + Self::checksum(slices) + } + } } impl PacketKey for CryptoKey { @@ -101,7 +136,7 @@ mod crypto { fn encrypt(&self, packet: u64, buf: &mut [u8], header_len: usize) { let (header, rest) = buf.split_at_mut(header_len); let (payload, tag) = rest.split_at_mut(rest.len() - self.tag_len()); - let checksum = Self::checksum(&[header, payload]); + let checksum = self.checksum_for(packet, &[header, payload]); tag.copy_from_slice(&checksum.to_be_bytes()); trace!(checksum, ?header, ?payload, ?tag); } @@ -115,7 +150,7 @@ mod crypto { ) -> Result<(), CryptoError> { let tag = payload.split_off(payload.len() - self.tag_len()).get_u64(); trace!(tag, ?payload); - let checksum = Self::checksum(&[header, payload]); + let checksum = self.checksum_for(packet, &[header, payload]); if checksum != tag { error!(tag, checksum, "checksum mismatch"); return Err(CryptoError); @@ -146,15 +181,17 @@ mod crypto { #[derive(Debug)] struct QuicSession { side: Side, + key: CryptoKey, state: HandshakeState, local: TransportParameters, remote: Option, } impl QuicSession { - fn new(side: Side, params: TransportParameters) -> Self { + fn new(side: Side, version: u32, params: TransportParameters) -> Self { Self { side, + key: CryptoKey::for_version(version), state: HandshakeState::EmitInitial, local: params, remote: None, @@ -164,7 +201,7 @@ mod crypto { impl Session for QuicSession { fn initial_keys(&self, _: &ConnectionId, _: Side) -> Keys { - CryptoKey.keys() + self.key.keys() } fn handshake_data(&self) -> Option> { @@ -212,21 +249,21 @@ mod crypto { self.local.write(buf); } self.state = HandshakeState::EmitHandshake; - Some(CryptoKey.keys()) + Some(self.key.keys()) } HandshakeState::EmitHandshake => { if self.side.is_server() { self.local.write(buf); } self.state = HandshakeState::Done; - Some(CryptoKey.keys()) + Some(self.key.keys()) } HandshakeState::Done => None, } } fn next_1rtt_keys(&mut self) -> Option>> { - Some(CryptoKey.packet()) + Some(self.key.packet()) } fn is_valid_retry(&self, _: &ConnectionId, _: &[u8], _: &[u8]) -> bool { @@ -254,13 +291,13 @@ mod crypto { server_name: &str, params: &TransportParameters, ) -> Result, ConnectError> { - Ok(Box::new(QuicSession::new(Side::Client, *params))) + Ok(Box::new(QuicSession::new(Side::Client, version, *params))) } } impl ServerConfig for CryptoConfig { - fn initial_keys(&self, _: u32, _: &ConnectionId) -> Result { - Ok(CryptoKey.keys()) + fn initial_keys(&self, version: u32, _: &ConnectionId) -> Result { + Ok(CryptoKey::for_version(version).keys()) } fn retry_tag(&self, _: u32, _: &ConnectionId, _: &[u8]) -> [u8; 16] { @@ -273,11 +310,59 @@ mod crypto { version: u32, params: &TransportParameters, ) -> Box { - Box::new(QuicSession::new(Side::Server, *params)) + Box::new(QuicSession::new(Side::Server, version, *params)) + } + } + + #[cfg(test)] + mod tests { + use super::*; + + fn roundtrip(pn_bound: bool, encrypt_pn: u64, decrypt_pn: u64) -> Result<(), CryptoError> { + let key = CryptoKey { pn_bound }; + let mut buf = vec![7u8; 8 + 32 + key.tag_len()]; + quinn_proto::crypto::PacketKey::encrypt(&key, encrypt_pn, &mut buf, 8); + let header = buf[..8].to_vec(); + let mut payload = BytesMut::from(&buf[8..]); + quinn_proto::crypto::PacketKey::decrypt(&key, decrypt_pn, &header, &mut payload) + } + + #[test] + fn legacy_checksum_ignores_packet_number() { + // Documents the original flaw: a packet number decoded differently + // from how it was encoded still passes authentication. + assert!(roundtrip(false, 100, 356).is_ok()); + } + + #[test] + fn pn_bound_checksum_rejects_shifted_packet_number() { + assert!(roundtrip(true, 100, 100).is_ok()); + // A 1-byte truncated packet number decoded one window (+256) ahead, + // as happens when reordering exceeds the RFC 9000 Appendix A + // decode window, must fail authentication. + assert!(roundtrip(true, 100, 356).is_err()); } } } +/// EasyTier custom QUIC version "ETQ1" (0x45545131, outside the reserved +/// `0x??a?a?a?a` grease pattern). +/// +/// Connections negotiated on this version bind the packet integrity checksum +/// to the packet number. With the legacy checksum, a 1-byte-encoded packet +/// number that arrives re-ordered beyond the decode window (RFC 9000 +/// Appendix A) is decoded as a future packet number; the checksum still +/// passes, the peer ACKs a number that was never sent, and the connection is +/// torn down with `PROTOCOL_VIOLATION("unsent packet acked")`. Binding the +/// checksum to the packet number makes such a misdecode fail authentication, +/// mirroring real QUIC where the AEAD nonce is derived from the packet +/// number. +/// +/// Used by both the QUIC proxy and the quic:// tunnel. Peers that only speak +/// version 1 reject it via version negotiation, so dialers must fall back to +/// [`client_config`] on `VersionMismatch` (see [`connect_with_etq1`]). +pub const QUIC_VERSION_ETQ1: u32 = 0x45545131; + pub fn transport_config() -> Arc { let mut config = TransportConfig::default(); @@ -305,11 +390,48 @@ pub fn client_config() -> ClientConfig { config } -pub fn endpoint_config() -> EndpointConfig { +/// Client config negotiating [`QUIC_VERSION_ETQ1`], the first choice of +/// dialers that support the fallback (see [`connect_with_etq1`]). +pub fn etq1_client_config() -> ClientConfig { + let mut config = client_config(); + config.version(QUIC_VERSION_ETQ1); + config +} + +pub(crate) fn endpoint_config_with_versions(versions: Vec) -> EndpointConfig { let mut config = EndpointConfig::default(); config.max_udp_payload_size(1200).unwrap(); + config.supported_versions(versions); config } + +/// Endpoint config accepting both [`QUIC_VERSION_ETQ1`] and legacy version +/// 1, so new dialers negotiate ETQ1 while old peers keep working. +pub fn endpoint_config() -> EndpointConfig { + endpoint_config_with_versions(vec![QUIC_VERSION_ETQ1, 1]) +} + +/// Dial `endpoint` preferring [`QUIC_VERSION_ETQ1`], falling back to legacy +/// version 1 when the remote rejects ETQ1 via version negotiation. +pub(crate) async fn connect_with_etq1( + endpoint: &Endpoint, + addr: SocketAddr, + server_name: &str, +) -> anyhow::Result { + match endpoint + .connect_with(etq1_client_config(), addr, server_name) + .with_context(|| format!("failed to start connection to {addr}"))? + .await + { + Ok(connection) => Ok(connection), + Err(ConnectionError::VersionMismatch) => endpoint + .connect_with(client_config(), addr, server_name) + .with_context(|| format!("failed to start connection to {addr}"))? + .await + .with_context(|| format!("failed to connect to {addr}")), + Err(error) => Err(error).with_context(|| format!("failed to connect to {addr}")), + } +} //endregion const QUIC_ACCEPT_COMPLETION_TIMEOUT: Duration = Duration::from_secs(10); @@ -338,13 +460,7 @@ pub(crate) async fn upgrade_connected( let mut endpoint = Endpoint::new_with_abstract_socket(endpoint_config(), None, socket, runtime)?; endpoint.set_default_client_config(client_config()); - let connecting = endpoint - .connect(remote_addr, "localhost") - .map_err(anyhow::Error::new) - .with_context(|| format!("failed to start connection to {remote_addr}"))?; - let connection = connecting - .await - .with_context(|| format!("failed to connect to {remote_addr}"))?; + let connection = connect_with_etq1(&endpoint, remote_addr, "localhost").await?; let (write, read) = connection .open_bi() .await