From 9959aac08539e36d69c325e89f90c1f4f3a7773e Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Thu, 16 Jul 2026 08:14:04 -0700 Subject: [PATCH] feat(quiche): add keep-alive, GSO opt-out, and dial/SNI split Upstream enablers for the "QUIC transport controls and routing" gaps in moq-dev/moq#2296, where moq-native currently ignores or rejects these options on the quiche backend. Each is exposed on both the ez (raw QUIC) and WebTransport builders, client and server. Keep-alive (with_keep_alive): quiche has no keep-alive of its own and QuicSettings has no interval, so the ez driver now runs an interval that calls send_ack_eliciting() and returns Ready so the io loop flushes the PING. quiche only adds a PING when the next packet wouldn't already be ack-eliciting, so a tick on a busy connection costs nothing. GSO opt-out (with_gso): the client applied max socket capabilities unconditionally and the server did the same when adopting a socket. apply_all_and_get_compatibility always enables GSO and the flags are pub(crate), so opting out reproduces it via SocketCapabilitiesBuilder minus gso(). The server holds listeners and its own sockets in one insertion-ordered enum and applies capabilities at build time, so with_gso works regardless of call order without clobbering the capabilities on a caller-supplied listener. Dial address vs SNI (with_server_name, connect_to): decouples the dial target from the TLS server name, so a caller can resolve the host itself (address-family-aware selection) and verify against a different name. Also fixes a pre-existing bug this surfaced: Url::host() renders IPv6 in bracketed URL form, so connecting to an IPv6-literal URL always failed DNS resolution. It sits directly in the dial/SNI path, so the feature would not work for its main use case without it. This promotes url from a dev-dependency to a real one. Preferred address remains blocked in quiche. QUIC-LB needs no change here: with_listener already accepts a QuicListener carrying its own cid_generator. The keep-alive test is paired with a control proving an un-kept connection genuinely dies of the idle timeout, and was verified to fail when send_ack_eliciting() is removed. Co-Authored-By: Claude Opus 4.8 --- rs/web-transport-quiche/Cargo.toml | 2 +- rs/web-transport-quiche/src/client.rs | 84 ++++++++++- rs/web-transport-quiche/src/ez/client.rs | 157 ++++++++++++------- rs/web-transport-quiche/src/ez/driver.rs | 57 ++++++- rs/web-transport-quiche/src/ez/mod.rs | 1 + rs/web-transport-quiche/src/ez/server.rs | 118 +++++++++++---- rs/web-transport-quiche/src/ez/socket.rs | 41 +++++ rs/web-transport-quiche/src/server.rs | 40 +++++ rs/web-transport-quiche/tests/transport.rs | 166 +++++++++++++++++++++ rs/web-transport-quiche/tests/verify.rs | 102 ++++++++++++- 10 files changed, 678 insertions(+), 90 deletions(-) create mode 100644 rs/web-transport-quiche/src/ez/socket.rs create mode 100644 rs/web-transport-quiche/tests/transport.rs diff --git a/rs/web-transport-quiche/Cargo.toml b/rs/web-transport-quiche/Cargo.toml index c7510610..fd4c072e 100644 --- a/rs/web-transport-quiche/Cargo.toml +++ b/rs/web-transport-quiche/Cargo.toml @@ -33,6 +33,7 @@ tokio = { version = "1", default-features = false, features = [ tokio-quiche = "0.19" tracing = "0.1" +url = "2" web-transport-proto = { workspace = true } web-transport-trait = { workspace = true } @@ -44,4 +45,3 @@ rustls-pemfile = "2" sha2 = "0.10" tokio = { version = "1", features = ["full"] } tracing-subscriber = { version = "0.3", features = ["env-filter"] } -url = "2" diff --git a/rs/web-transport-quiche/src/client.rs b/rs/web-transport-quiche/src/client.rs index b4427cc6..dc47de15 100644 --- a/rs/web-transport-quiche/src/client.rs +++ b/rs/web-transport-quiche/src/client.rs @@ -98,8 +98,45 @@ impl ClientBuilder { Self(self.0.with_server_certificate_hashes(hashes)) } + /// Override the TLS server name (SNI) used for the handshake and hostname + /// verification. + /// + /// Defaults to the host in the URL passed to [ClientBuilder::connect], which + /// is almost always what you want. Set it explicitly to reach a server whose + /// certificate names a different host than the one you're dialing. + /// + /// The URL is otherwise unaffected: the `:authority` sent in the CONNECT + /// request still comes from the URL. + pub fn with_server_name(self, name: impl Into) -> Self { + Self(self.0.with_server_name(name)) + } + + /// Send a PING on this interval, keeping an idle connection alive. + /// + /// Disabled by default. This must be shorter than the peer's + /// [Settings::max_idle_timeout] to have any effect; a third of it is a + /// reasonable choice. + pub fn with_keep_alive(self, interval: std::time::Duration) -> Self { + Self(self.0.with_keep_alive(interval)) + } + + /// Enable UDP generic segmentation offload (GSO), on by default. + /// + /// GSO cuts syscall overhead at high throughput by handing the kernel + /// several packets at once, but some NICs and virtual network stacks + /// mishandle it. Turn it off if large sends are being dropped. + /// + /// Only Linux supports GSO; elsewhere this does nothing. + pub fn with_gso(self, enabled: bool) -> Self { + Self(self.0.with_gso(enabled)) + } + /// Connect to the WebTransport server at the given URL. /// + /// The URL host is resolved via DNS and the first address is used. Resolve + /// it yourself and call [ClientBuilder::connect_to] to choose the address, + /// for example to prefer a particular address family. + /// /// DNS resolution and socket setup happen eagerly. The returned [Connecting] /// has an [established](Connecting::established) method to complete the full handshake /// (TLS + SETTINGS + CONNECT). @@ -110,21 +147,56 @@ impl ClientBuilder { request: impl Into, ) -> Result { let request = request.into(); + let (host, port) = Self::target(&request)?; - let port = request.url.port().unwrap_or(443); + let connecting = self.0.connect(&host, port).await?; - let host = match request.url.host() { - Some(host) => host.to_string(), - None => return Err(ClientError::InvalidUrl(request.url.to_string())), - }; + Ok(Connecting { + connecting, + request, + }) + } - let connecting = self.0.connect(&host, port).await?; + /// Connect to the WebTransport server at an already-resolved address. + /// + /// The URL still supplies the request itself, including the TLS server name + /// unless [ClientBuilder::with_server_name] overrides it. Only the address + /// to dial is taken from `remote`. + /// + /// This takes ownership because the underlying quiche implementation doesn't support reusing the same socket. + pub async fn connect_to( + mut self, + request: impl Into, + remote: std::net::SocketAddr, + ) -> Result { + let request = request.into(); + let (host, _) = Self::target(&request)?; + + // `ez::connect_to` has no host to fall back on, so pin down the name the + // URL implies unless the caller already chose one. + self.0 = self.0.with_default_server_name(host); + + let connecting = self.0.connect_to(remote).await?; Ok(Connecting { connecting, request, }) } + + /// The host and port to dial for a request. + fn target(request: &ConnectRequest) -> Result<(String, u16), ClientError> { + // `Host` renders IPv6 in URL form, bracketed, which is not what a + // resolver or a TLS server name wants. + let host = match request.url.host() { + Some(url::Host::Domain(host)) => host.to_string(), + Some(url::Host::Ipv4(ip)) => ip.to_string(), + Some(url::Host::Ipv6(ip)) => ip.to_string(), + None => return Err(ClientError::InvalidUrl(request.url.to_string())), + }; + + Ok((host, request.url.port().unwrap_or(443))) + } } /// A WebTransport connection that is still completing the handshake. diff --git a/rs/web-transport-quiche/src/ez/client.rs b/rs/web-transport-quiche/src/ez/client.rs index 9f49d57a..169523e5 100644 --- a/rs/web-transport-quiche/src/ez/client.rs +++ b/rs/web-transport-quiche/src/ez/client.rs @@ -1,9 +1,12 @@ use std::io; +use std::net::SocketAddr; use std::sync::Arc; +use std::time::Duration; use tokio_quiche::settings::{CertificateKind, Hooks, TlsCertificatePaths}; use rustls_pki_types::{CertificateDer, PrivateKeyDer}; +use crate::ez::socket::capabilities; use crate::ez::tls::{ClientHook, ClientVerify}; use crate::ez::DriverState; @@ -24,7 +27,13 @@ pub struct ClientBuilder { socket: Option, tls: Option<(Vec>, PrivateKeyDer<'static>)>, verify: ClientVerify, + // Held for API symmetry with ServerBuilder: `connect_with_config` has no + // metrics hook, so tokio-quiche never reports client-side metrics. + #[allow(dead_code)] metrics: M, + server_name: Option, + keep_alive: Option, + gso: bool, } impl Default for ClientBuilder { @@ -45,23 +54,19 @@ impl ClientBuilder { socket: None, tls: None, verify: ClientVerify::Default, + server_name: None, + keep_alive: None, + gso: true, } } /// Listen for incoming packets on the given socket. /// /// Defaults to an ephemeral port if not specified. - pub fn with_socket(self, socket: std::net::UdpSocket) -> io::Result { + pub fn with_socket(mut self, socket: std::net::UdpSocket) -> io::Result { socket.set_nonblocking(true)?; - let socket = tokio::net::UdpSocket::from_std(socket)?; - - Ok(Self { - socket: Some(socket), - settings: self.settings, - metrics: self.metrics, - tls: self.tls, - verify: self.verify, - }) + self.socket = Some(tokio::net::UdpSocket::from_std(socket)?); + Ok(self) } /// Listen for incoming packets on the given address. @@ -84,17 +89,53 @@ impl ClientBuilder { /// Optional: Use a client certificate for mTLS. pub fn with_single_cert( - self, + mut self, chain: Vec>, key: PrivateKeyDer<'static>, ) -> Self { - Self { - tls: Some((chain, key)), - settings: self.settings, - metrics: self.metrics, - socket: self.socket, - verify: self.verify, - } + self.tls = Some((chain, key)); + self + } + + /// Override the TLS server name (SNI) used for the handshake and hostname + /// verification. + /// + /// [ClientBuilder::connect] defaults this to the host it was given, which is + /// almost always what you want. Set it explicitly to reach a server whose + /// certificate names a different host than the one you're dialing, and to + /// name the peer for [ClientBuilder::connect_to]. + pub fn with_server_name(mut self, name: impl Into) -> Self { + self.server_name = Some(name.into()); + self + } + + /// Set the TLS server name unless [ClientBuilder::with_server_name] already + /// chose one, so an explicit override always wins. + pub(crate) fn with_default_server_name(mut self, name: impl Into) -> Self { + self.server_name.get_or_insert_with(|| name.into()); + self + } + + /// Send a PING on this interval, keeping an idle connection alive. + /// + /// Disabled by default. This must be shorter than the peer's + /// [Settings::max_idle_timeout] to have any effect; a third of it is a + /// reasonable choice. + pub fn with_keep_alive(mut self, interval: Duration) -> Self { + self.keep_alive = Some(interval); + self + } + + /// Enable UDP generic segmentation offload (GSO), on by default. + /// + /// GSO cuts syscall overhead at high throughput by handing the kernel + /// several packets at once, but some NICs and virtual network stacks + /// mishandle it. Turn it off if large sends are being dropped. + /// + /// Only Linux supports GSO; elsewhere this does nothing. + pub fn with_gso(mut self, enabled: bool) -> Self { + self.gso = enabled; + self } /// Verify the server certificate against an explicit set of root @@ -116,49 +157,59 @@ impl ClientBuilder { /// Connect to the QUIC server at the given host and port. /// + /// The host is resolved via DNS and the first address is used. Resolve the + /// host yourself and call [ClientBuilder::connect_to] to choose the address, + /// for example to prefer a particular address family. + /// + /// Unless [ClientBuilder::with_server_name] overrides it, `host` is also the + /// TLS server name. + /// + /// This takes ownership because the underlying quiche implementation doesn't support reusing the same socket. + pub async fn connect(self, host: &str, port: u16) -> io::Result { + let mut remotes = tokio::net::lookup_host((host, port)) + .await + .map_err(|err| io::Error::new(io::ErrorKind::HostUnreachable, err.to_string()))?; + + let remote = remotes.next().ok_or_else(|| { + io::Error::new( + io::ErrorKind::HostUnreachable, + "no addresses found for host", + ) + })?; + + self.with_default_server_name(host).connect_to(remote).await + } + + /// Connect to the QUIC server at an already-resolved address. + /// + /// [ClientBuilder::with_server_name] is required, as there's no host name to + /// derive the TLS server name from. + /// /// This takes ownership because the underlying quiche implementation doesn't support reusing the same socket. - pub async fn connect(mut self, host: &str, port: u16) -> io::Result { + pub async fn connect_to(mut self, remote: SocketAddr) -> io::Result { + let server_name = self.server_name.take().ok_or_else(|| { + io::Error::new( + io::ErrorKind::InvalidInput, + "connect_to requires with_server_name", + ) + })?; + if self.socket.is_none() { self = self.with_bind("[::]:0")?; } let socket = self.socket.take().unwrap(); - - let mut remotes = match tokio::net::lookup_host((host, port)).await { - Ok(remotes) => remotes, - Err(err) => { - return Err(io::Error::new( - io::ErrorKind::HostUnreachable, - err.to_string(), - )); - } - }; - - // Return the first entry. - let remote = match remotes.next() { - Some(remote) => remote, - None => { - return Err(io::Error::new( - io::ErrorKind::HostUnreachable, - "no addresses found for host", - )) - } - }; - socket.connect(remote).await?; - // Connect to the server using the addr we just resolved. - #[cfg_attr(not(target_os = "linux"), allow(unused_mut))] + // Enable the offloads the kernel supports before the socket is wrapped; + // `from_udp` starts with everything disabled. + let capabilities = capabilities(&socket, self.gso); + let mut socket = tokio_quiche::socket::Socket::< Arc, Arc, >::from_udp(socket)?; - - // Enable UDP GSO/GRO offload where the kernel supports it (Linux only). - // This mirrors the server listener and cuts syscall overhead at high - // throughput; it's a no-op if send/recv don't share one FD. - #[cfg(target_os = "linux")] - socket.apply_max_capabilities(); + socket.capabilities = capabilities; // Only the fully-insecure path (no verification of any kind) deserves a // warning; hash- and root-based verification still authenticate the peer. @@ -205,11 +256,13 @@ impl ClientBuilder { dgram_in.0, dgram_out.1, dgram_max.clone(), + self.keep_alive, ); - let conn = tokio_quiche::quic::connect_with_config(socket, Some(host), ¶ms, app) - .await - .map_err(|e| io::Error::other(e.to_string()))?; + let conn = + tokio_quiche::quic::connect_with_config(socket, Some(&server_name), ¶ms, app) + .await + .map_err(|e| io::Error::other(e.to_string()))?; let conn = Connection::new( conn, diff --git a/rs/web-transport-quiche/src/ez/driver.rs b/rs/web-transport-quiche/src/ez/driver.rs index 66f0f5f5..3b000439 100644 --- a/rs/web-transport-quiche/src/ez/driver.rs +++ b/rs/web-transport-quiche/src/ez/driver.rs @@ -6,7 +6,8 @@ use std::{ atomic::{AtomicUsize, Ordering}, Arc, }, - task::{Poll, Waker}, + task::{Context, Poll, Waker}, + time::Duration, }; use tokio_quiche::{ buf_factory::BufFactory, @@ -223,6 +224,42 @@ impl DriverState { } } +/// Periodically asks quiche to make the next packet ack-eliciting, keeping an +/// otherwise silent connection out of the peer's idle timeout and holding NAT +/// bindings open. +struct KeepAlive { + period: Duration, + /// Created on the first poll so the timer registers with the runtime that + /// actually drives the connection, not whoever built the endpoint. + ticker: Option, +} + +impl KeepAlive { + fn new(period: Duration) -> Self { + Self { + period, + ticker: None, + } + } + + /// Returns true when a keep-alive is due. + fn poll(&mut self, cx: &mut Context) -> bool { + let period = self.period; + let ticker = self.ticker.get_or_insert_with(|| { + // The first tick is one period out; `interval` would instead fire + // immediately and ping a connection that just finished handshaking. + let start = tokio::time::Instant::now() + period; + let mut ticker = tokio::time::interval_at(start, period); + // A late tick means the connection was busy, which is exactly when a + // keep-alive is unnecessary. Don't replay the backlog. + ticker.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay); + ticker + }); + + ticker.poll_tick(cx).is_ready() + } +} + pub(super) struct Driver { state: Lock, @@ -240,6 +277,8 @@ pub(super) struct Driver { // Writable datagram size in bytes, published once at handshake. 0 means the // peer didn't negotiate the datagram extension. dgram_max: Arc, + + keep_alive: Option, } impl Driver { @@ -250,6 +289,7 @@ impl Driver { dgram_in: flume::Sender, dgram_out: flume::Receiver, dgram_max: Arc, + keep_alive: Option, ) -> Self { Self { state, @@ -261,6 +301,7 @@ impl Driver { dgram_in, dgram_out, dgram_max, + keep_alive: keep_alive.map(KeepAlive::new), } } @@ -453,6 +494,16 @@ impl Driver { return Poll::Pending; } + // quiche only adds a PING if the next packet wouldn't already be + // ack-eliciting, so a tick on a busy connection costs nothing. + let mut keep_alive = false; + if let Some(k) = self.keep_alive.as_mut() { + if k.poll(&mut Context::from_waker(waker)) { + qconn.send_ack_eliciting()?; + keep_alive = true; + } + } + // Snapshot stats while we hold an immutable view; stored under the lock below. let stats = ConnectionStats::from_quiche(qconn); @@ -516,7 +567,9 @@ impl Driver { self.flush_send(qconn, stream_id)?; } - if sleep { + // Returning Ready hands control back to the io loop, which flushes the + // scheduled PING to the socket. + if sleep && !keep_alive { Poll::Pending } else { Poll::Ready(Ok(())) diff --git a/rs/web-transport-quiche/src/ez/mod.rs b/rs/web-transport-quiche/src/ez/mod.rs index fbb3c3af..9ad24bdd 100644 --- a/rs/web-transport-quiche/src/ez/mod.rs +++ b/rs/web-transport-quiche/src/ez/mod.rs @@ -11,6 +11,7 @@ mod lock; mod recv; mod send; mod server; +mod socket; mod stream; pub mod tls; diff --git a/rs/web-transport-quiche/src/ez/server.rs b/rs/web-transport-quiche/src/ez/server.rs index 307f919e..ece0c863 100644 --- a/rs/web-transport-quiche/src/ez/server.rs +++ b/rs/web-transport-quiche/src/ez/server.rs @@ -1,15 +1,17 @@ use boring::ssl::NameType; use std::net::SocketAddr; use std::sync::Arc; +use std::time::Duration; use std::{io, marker::PhantomData}; use tokio::sync::mpsc; use tokio::task::JoinSet; use tokio_quiche::quic::SimpleConnectionIdGenerator; use tokio_quiche::settings::{CertificateKind, Hooks, TlsCertificatePaths}; -use tokio_quiche::socket::{QuicListener, SocketCapabilities}; +use tokio_quiche::socket::QuicListener; use rustls_pki_types::{CertificateDer, PrivateKeyDer}; +use crate::ez::socket::capabilities; use crate::ez::tls::{DynamicCertHook, StaticCertHook}; use crate::ez::DriverState; @@ -25,7 +27,17 @@ pub struct ServerInit {} /// Used with [ServerBuilder] to require at least one listener. #[derive(Default)] pub struct ServerWithListener { - listeners: Vec, + listeners: Vec, +} + +/// A listener, in insertion order so [Server::local_addrs] matches the order the +/// caller added them. +enum Listener { + /// Supplied by the caller and used as-is. + Ready(QuicListener), + /// Opened by us. It stays a bare socket until the build step, the first + /// point where every option affecting it is known. + Socket(tokio::net::UdpSocket), } /// Construct a QUIC server using sane defaults. @@ -34,6 +46,8 @@ pub struct ServerBuilder { metrics: M, state: S, alpn: Vec>, + keep_alive: Option, + gso: bool, } impl Default for ServerBuilder { @@ -52,6 +66,8 @@ impl ServerBuilder { metrics: m, state: ServerInit {}, alpn: Vec::new(), + keep_alive: None, + gso: true, } } } @@ -61,8 +77,10 @@ impl ServerBuilder { ServerBuilder { settings: self.settings, metrics: self.metrics, - state: ServerWithListener { listeners: vec![] }, + state: ServerWithListener::default(), alpn: self.alpn, + keep_alive: self.keep_alive, + gso: self.gso, } } @@ -92,33 +110,40 @@ impl ServerBuilder { self.settings = settings; self } + + /// Send a PING to each client on this interval, keeping idle connections alive. + /// + /// See [ServerBuilder::with_keep_alive](ServerBuilder::::with_keep_alive). + pub fn with_keep_alive(mut self, interval: Duration) -> Self { + self.keep_alive = Some(interval); + self + } + + /// Enable UDP generic segmentation offload (GSO), on by default. + /// + /// See [ServerBuilder::with_gso](ServerBuilder::::with_gso). + pub fn with_gso(mut self, enabled: bool) -> Self { + self.gso = enabled; + self + } } impl ServerBuilder { /// Configure the server to use the provided QUIC listener. + /// + /// The listener is used as-is: it carries its own capabilities and + /// connection ID generator, so [ServerBuilder::with_gso] does not apply. pub fn with_listener(mut self, listener: QuicListener) -> Self { - self.state.listeners.push(listener); + self.state.listeners.push(Listener::Ready(listener)); self } /// Listen for incoming packets on the given socket. - pub fn with_socket(self, socket: std::net::UdpSocket) -> io::Result { + pub fn with_socket(mut self, socket: std::net::UdpSocket) -> io::Result { socket.set_nonblocking(true)?; let socket = tokio::net::UdpSocket::from_std(socket)?; - - // TODO Modify quiche to add other platform support. - #[cfg(target_os = "linux")] - let capabilities = SocketCapabilities::apply_all_and_get_compatibility(&socket); - #[cfg(not(target_os = "linux"))] - let capabilities = SocketCapabilities::default(); - - let listener = QuicListener { - socket, - cid_generator: Arc::new(SimpleConnectionIdGenerator), - capabilities, - }; - - Ok(self.with_listener(listener)) + self.state.listeners.push(Listener::Socket(socket)); + Ok(self) } /// Listen for incoming packets on the given address. @@ -134,6 +159,31 @@ impl ServerBuilder { self } + /// Send a PING to each client on this interval, keeping idle connections alive. + /// + /// Disabled by default. A server usually wants to let idle clients time out + /// rather than hold them open, so reach for this only when something in the + /// path (a NAT or load balancer) drops silent flows sooner than + /// [Settings::max_idle_timeout] would. + pub fn with_keep_alive(mut self, interval: Duration) -> Self { + self.keep_alive = Some(interval); + self + } + + /// Enable UDP generic segmentation offload (GSO), on by default. + /// + /// GSO cuts syscall overhead at high throughput by handing the kernel + /// several packets at once, but some NICs and virtual network stacks + /// mishandle it. Turn it off if large sends are being dropped. + /// + /// This applies to sockets from [ServerBuilder::with_socket] and + /// [ServerBuilder::with_bind] only, not to a [ServerBuilder::with_listener] + /// listener. Only Linux supports GSO; elsewhere this does nothing. + pub fn with_gso(mut self, enabled: bool) -> Self { + self.gso = enabled; + self + } + /// Configure the server to use a static certificate for TLS. pub fn with_single_cert( mut self, @@ -169,18 +219,29 @@ impl ServerBuilder { connection_hook: Some(hook), }; - // Capture local addresses before the listeners are consumed. - let local_addrs: Vec = self + let listeners: Vec = self .state .listeners - .iter() - .map(|l| l.socket.local_addr().unwrap()) + .into_iter() + .map(|listener| match listener { + Listener::Ready(listener) => listener, + Listener::Socket(socket) => QuicListener { + capabilities: capabilities(&socket, self.gso), + socket, + cid_generator: Arc::new(SimpleConnectionIdGenerator), + }, + }) .collect(); + // Capture local addresses before the listeners are consumed. + let local_addrs: Vec = listeners + .iter() + .map(|l| l.socket.local_addr()) + .collect::>()?; + let params = tokio_quiche::ConnectionParams::new_server(self.settings, dummy_tls, hooks); - let server = - tokio_quiche::listen_with_capabilities(self.state.listeners, params, self.metrics)?; - Ok(Server::new(server, local_addrs)) + let server = tokio_quiche::listen_with_capabilities(listeners, params, self.metrics)?; + Ok(Server::new(server, local_addrs, self.keep_alive)) } } @@ -250,6 +311,7 @@ impl Server { fn new( sockets: Vec>, local_addrs: Vec, + keep_alive: Option, ) -> Self { let mut tasks = JoinSet::default(); @@ -258,7 +320,7 @@ impl Server { for socket in sockets { let accept = accept.0.clone(); // TODO close all when one errors - tasks.spawn(Self::run_socket(socket, accept)); + tasks.spawn(Self::run_socket(socket, accept, keep_alive)); } Self { @@ -272,6 +334,7 @@ impl Server { async fn run_socket( socket: tokio_quiche::QuicConnectionStream, accept: mpsc::Sender, + keep_alive: Option, ) -> io::Result<()> { let mut rx = socket.into_inner(); while let Some(initial) = rx.recv().await { @@ -298,6 +361,7 @@ impl Server { dgram_in.0, dgram_out.1, dgram_max.clone(), + keep_alive, ); let inner = initial.start(session); diff --git a/rs/web-transport-quiche/src/ez/socket.rs b/rs/web-transport-quiche/src/ez/socket.rs new file mode 100644 index 00000000..84787ed4 --- /dev/null +++ b/rs/web-transport-quiche/src/ez/socket.rs @@ -0,0 +1,41 @@ +use tokio_quiche::socket::SocketCapabilities; + +/// Enable the socket options tokio-quiche knows how to use, optionally leaving +/// `UDP_SEGMENT` (GSO) off. +/// +/// All of these are Linux-only; every other platform reports no capabilities. +#[cfg(target_os = "linux")] +pub(super) fn capabilities(socket: &S, gso: bool) -> SocketCapabilities { + use tokio_quiche::socket::SocketCapabilitiesBuilder; + + if gso { + return SocketCapabilities::apply_all_and_get_compatibility(socket); + } + + // `apply_all_and_get_compatibility` always turns GSO on, so opting out means + // reproducing it here minus the `gso()` call. Each option is independently + // best-effort: a kernel that rejects one still gets the rest. + let mut builder = SocketCapabilitiesBuilder::new(socket); + let _ = builder.check_udp_drop(); + let _ = builder.txtime(); + let _ = builder.gro(); + let _ = builder.rcvmark(); + let _ = builder.ip_mtu_discover_probe(); + let _ = builder.ipv6_mtu_discover_probe(); + + // Source-address control is only useful when the socket may send from an + // address it isn't bound to. + if let Ok(true) = builder.allows_nonlocal_source() { + let _ = builder.ipv4_pktinfo(); + let _ = builder.ipv4_recvorigdstaddr(); + let _ = builder.ipv6_pktinfo(); + let _ = builder.ipv6_recvorigdstaddr(); + } + + builder.finish() +} + +#[cfg(not(target_os = "linux"))] +pub(super) fn capabilities(_socket: &S, _gso: bool) -> SocketCapabilities { + SocketCapabilities::default() +} diff --git a/rs/web-transport-quiche/src/server.rs b/rs/web-transport-quiche/src/server.rs index db587737..4a01d0ec 100644 --- a/rs/web-transport-quiche/src/server.rs +++ b/rs/web-transport-quiche/src/server.rs @@ -81,10 +81,27 @@ impl ServerBuilder { pub fn with_settings(self, settings: ez::Settings) -> Self { Self(self.0.with_settings(settings)) } + + /// Send a PING to each client on this interval, keeping idle connections alive. + /// + /// See [ServerBuilder::with_keep_alive](ServerBuilder::::with_keep_alive). + pub fn with_keep_alive(self, interval: std::time::Duration) -> Self { + Self(self.0.with_keep_alive(interval)) + } + + /// Enable UDP generic segmentation offload (GSO), on by default. + /// + /// See [ServerBuilder::with_gso](ServerBuilder::::with_gso). + pub fn with_gso(self, enabled: bool) -> Self { + Self(self.0.with_gso(enabled)) + } } impl ServerBuilder { /// Configure the server to use the provided QUIC listener. + /// + /// The listener is used as-is: it carries its own capabilities and + /// connection ID generator, so [ServerBuilder::with_gso] does not apply. pub fn with_listener(self, listener: tokio_quiche::socket::QuicListener) -> Self { Self(self.0.with_listener(listener)) } @@ -104,6 +121,29 @@ impl ServerBuilder { Self(self.0.with_settings(settings)) } + /// Send a PING to each client on this interval, keeping idle connections alive. + /// + /// Disabled by default. A server usually wants to let idle clients time out + /// rather than hold them open, so reach for this only when something in the + /// path (a NAT or load balancer) drops silent flows sooner than + /// [Settings::max_idle_timeout](ez::Settings::max_idle_timeout) would. + pub fn with_keep_alive(self, interval: std::time::Duration) -> Self { + Self(self.0.with_keep_alive(interval)) + } + + /// Enable UDP generic segmentation offload (GSO), on by default. + /// + /// GSO cuts syscall overhead at high throughput by handing the kernel + /// several packets at once, but some NICs and virtual network stacks + /// mishandle it. Turn it off if large sends are being dropped. + /// + /// This applies to sockets from [ServerBuilder::with_socket] and + /// [ServerBuilder::with_bind] only, not to a [ServerBuilder::with_listener] + /// listener. Only Linux supports GSO; elsewhere this does nothing. + pub fn with_gso(self, enabled: bool) -> Self { + Self(self.0.with_gso(enabled)) + } + /// Configure the server to use a static certificate for TLS. pub fn with_single_cert( self, diff --git a/rs/web-transport-quiche/tests/transport.rs b/rs/web-transport-quiche/tests/transport.rs new file mode 100644 index 00000000..1060835b --- /dev/null +++ b/rs/web-transport-quiche/tests/transport.rs @@ -0,0 +1,166 @@ +//! QUIC transport controls: keep-alive and the GSO opt-out. +//! +//! The load-bearing assertion for keep-alive is the negative one: the same +//! connection *without* it must actually die of the idle timeout, otherwise the +//! positive test would pass no matter what the driver does. + +use std::{ + net::{Ipv4Addr, SocketAddr}, + time::Duration, +}; + +use anyhow::{Context, Result}; +use rcgen::{CertifiedKey, KeyPair}; +use rustls_pki_types::{CertificateDer, PrivateKeyDer, PrivatePkcs8KeyDer}; +use url::Url; +use web_transport_quiche::{ClientBuilder, ServerBuilder, Settings}; + +/// Short enough to keep the test quick, long enough to survive a loaded CI +/// machine stalling the driver task between ticks. +const IDLE_TIMEOUT: Duration = Duration::from_secs(1); +const KEEP_ALIVE: Duration = Duration::from_millis(200); + +/// Long enough that an un-kept connection is certainly gone. +const IDLE_WAIT: Duration = Duration::from_secs(3); + +fn make_self_signed() -> Result<(Vec>, PrivateKeyDer<'static>)> { + let CertifiedKey { cert, signing_key } = + rcgen::generate_simple_self_signed(vec!["localhost".into(), "127.0.0.1".into()]) + .context("rcgen self-signed")?; + + let cert_der = CertificateDer::from(cert.der().to_vec()); + let key_bytes = KeyPair::serialize_der(&signing_key); + let key_der = PrivateKeyDer::Pkcs8(PrivatePkcs8KeyDer::from(key_bytes)); + + Ok((vec![cert_der], key_der)) +} + +fn init_tracing() { + let _ = tracing_subscriber::fmt() + .with_env_filter( + tracing_subscriber::EnvFilter::try_from_default_env() + .unwrap_or_else(|_| tracing_subscriber::EnvFilter::new("warn")), + ) + .with_test_writer() + .try_init(); +} + +fn idle_settings() -> Settings { + let mut settings = Settings::default(); + settings.max_idle_timeout = Some(IDLE_TIMEOUT); + settings +} + +/// Spawn a server that accepts one session and holds it open without sending +/// anything, so the only traffic on the wire is whatever keep-alive produces. +async fn spawn_server(gso: bool) -> Result<(SocketAddr, tokio::task::JoinHandle<()>)> { + let (chain, key) = make_self_signed()?; + + let bind: SocketAddr = (Ipv4Addr::LOCALHOST, 0).into(); + let mut server = ServerBuilder::default() + .with_bind(bind)? + .with_settings(idle_settings()) + .with_gso(gso) + .with_single_cert(chain, key)?; + + let addr = *server + .local_addrs() + .first() + .context("server has no local address")?; + + let handle = tokio::spawn(async move { + if let Some(request) = server.accept().await { + if let Ok(session) = request.ok().await { + let _ = session.closed().await; + } + } + }); + + Ok((addr, handle)) +} + +fn url_for(addr: SocketAddr) -> Result { + Ok(Url::parse(&format!("https://127.0.0.1:{}/", addr.port()))?) +} + +fn client() -> ClientBuilder { + // The cert is self-signed, and the point here is transport behavior rather + // than verification, which verify.rs already covers. + let mut settings = idle_settings(); + settings.verify_peer = false; + + ClientBuilder::default().with_settings(settings) +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn keep_alive_outlives_idle_timeout() -> Result<()> { + init_tracing(); + + let (addr, server) = spawn_server(true).await?; + + let session = client() + .with_bind((Ipv4Addr::LOCALHOST, 0))? + .with_keep_alive(KEEP_ALIVE) + .connect(url_for(addr)?) + .await? + .established() + .await?; + + // Nothing is sent on the session, so only the keep-alive PINGs can hold the + // idle timer open on either end. + if let Ok(err) = tokio::time::timeout(IDLE_WAIT, session.closed()).await { + anyhow::bail!("keep-alive connection closed after {IDLE_WAIT:?}: {err}"); + } + + session.close(0, "bye"); + session.closed().await; + server.abort(); + Ok(()) +} + +/// The control for `keep_alive_outlives_idle_timeout`. +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn idle_timeout_without_keep_alive() -> Result<()> { + init_tracing(); + + let (addr, server) = spawn_server(true).await?; + + let session = client() + .with_bind((Ipv4Addr::LOCALHOST, 0))? + .connect(url_for(addr)?) + .await? + .established() + .await?; + + tokio::time::timeout(IDLE_WAIT, session.closed()) + .await + .map_err(|_| { + anyhow::anyhow!("idle connection survived {IDLE_WAIT:?} with no keep-alive") + })?; + + server.abort(); + Ok(()) +} + +/// GSO is Linux-only, so elsewhere this just pins that the option is accepted +/// rather than rejected at initialization. +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn handshake_without_gso() -> Result<()> { + init_tracing(); + + let (addr, server) = spawn_server(false).await?; + + let session = client() + .with_bind((Ipv4Addr::LOCALHOST, 0))? + .with_gso(false) + .connect(url_for(addr)?) + .await? + .established() + .await + .context("handshake should succeed with GSO disabled")?; + + session.close(0, "bye"); + session.closed().await; + server.abort(); + Ok(()) +} diff --git a/rs/web-transport-quiche/tests/verify.rs b/rs/web-transport-quiche/tests/verify.rs index aeccff86..7d2769fb 100644 --- a/rs/web-transport-quiche/tests/verify.rs +++ b/rs/web-transport-quiche/tests/verify.rs @@ -37,6 +37,18 @@ fn make_ca_chain() -> Result<( CertificateDer<'static>, Vec>, PrivateKeyDer<'static>, +)> { + make_ca_chain_for(vec!["localhost".into(), "127.0.0.1".into()]) +} + +/// [make_ca_chain], but with an explicit set of leaf SANs. +#[allow(clippy::type_complexity)] +fn make_ca_chain_for( + sans: Vec, +) -> Result<( + CertificateDer<'static>, + Vec>, + PrivateKeyDer<'static>, )> { let ca_key = KeyPair::generate().context("ca key")?; let mut ca_params = CertificateParams::new(Vec::new()).context("ca params")?; @@ -48,8 +60,7 @@ fn make_ca_chain() -> Result<( let ca_cert = ca_params.self_signed(&ca_key).context("self-sign ca")?; let leaf_key = KeyPair::generate().context("leaf key")?; - let mut leaf_params = CertificateParams::new(vec!["localhost".into(), "127.0.0.1".into()]) - .context("leaf params")?; + let mut leaf_params = CertificateParams::new(sans).context("leaf params")?; leaf_params .distinguished_name .push(DnType::CommonName, "localhost"); @@ -254,3 +265,90 @@ async fn custom_roots_reject() -> Result<()> { server.abort(); Ok(()) } + +/// A URL for the literal address, which the `localhost`-only leaf below does +/// *not* cover. +fn ip_url_for(addr: SocketAddr) -> Result { + Ok(Url::parse(&format!("https://{addr}/"))?) +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn server_name_override_accept() -> Result<()> { + init_tracing(); + + // Leaf covers `localhost` only, so dialing the address literal can only work + // if the TLS server name is decoupled from the dial target. + let (ca_root, chain, key) = make_ca_chain_for(vec!["localhost".into()])?; + let (addr, server) = spawn_server(chain, key).await?; + + let session = ClientBuilder::default() + .with_bind(loopback_for(addr))? + .with_root_certificates(vec![ca_root]) + .with_server_name("localhost") + .connect(ip_url_for(addr)?) + .await? + .established() + .await + .context("handshake should succeed when SNI names a host the cert covers")?; + + session.close(0, "bye"); + session.closed().await; + server.abort(); + Ok(()) +} + +/// The control for `server_name_override_accept`: without the override the +/// dial target is the server name, and the cert doesn't cover it. +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn server_name_override_required() -> Result<()> { + init_tracing(); + + let (ca_root, chain, key) = make_ca_chain_for(vec!["localhost".into()])?; + let (addr, server) = spawn_server(chain, key).await?; + + let url = ip_url_for(addr)?; + let client_bind = loopback_for(addr); + + let result = tokio::time::timeout(Duration::from_secs(5), async move { + ClientBuilder::default() + .with_bind(client_bind)? + .with_root_certificates(vec![ca_root]) + .connect(url) + .await? + .established() + .await + }) + .await + .context("handshake neither succeeded nor failed within the timeout")?; + + assert!( + result.is_err(), + "handshake must fail when the cert does not cover the dialed host" + ); + + server.abort(); + Ok(()) +} + +/// `connect_to` skips DNS but still takes the server name from the URL. +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn connect_to_resolved_addr() -> Result<()> { + init_tracing(); + + let (ca_root, chain, key) = make_ca_chain()?; + let (addr, server) = spawn_server(chain, key).await?; + + let session = ClientBuilder::default() + .with_bind(loopback_for(addr))? + .with_root_certificates(vec![ca_root]) + .connect_to(url_for(addr)?, addr) + .await? + .established() + .await + .context("handshake should succeed against an already-resolved address")?; + + session.close(0, "bye"); + session.closed().await; + server.abort(); + Ok(()) +}