diff --git a/Cargo.lock b/Cargo.lock index 8ad6c3ba018..18b6155f02b 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -4943,6 +4943,7 @@ dependencies = [ "futures", "hex", "http", + "httparse", "humantime", "humantime-serde", "jni", @@ -5043,7 +5044,6 @@ dependencies = [ "rind-rustls", "rustc-hash 2.1.3", "rustls", - "rustls-platform-verifier", "send_wrapper", "serde_json", "slab", diff --git a/Cargo.toml b/Cargo.toml index 8f7a2c69b83..c2aa1e318ba 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -113,6 +113,7 @@ hashbrown = { version = "0.17.1", default-features = false } hashlink = "0.12.2" hex = "0.4" http-body = "1.0.1" +httparse = "1.10" humantime = "2.3" humantime-serde = "1.1.1" hyper = "1.11" diff --git a/quest/m2/api-shape/README.md b/quest/m2/api-shape/README.md index 35e56ef6430..c32ca6acaf8 100644 --- a/quest/m2/api-shape/README.md +++ b/quest/m2/api-shape/README.md @@ -60,5 +60,4 @@ Rejected in planning: ## Required - [moq-net's route stream goes public](/quest/m2/api-shape/net.md) - un-hide `origin::Consumer::routes` and reshape `RouteEvent`, now that #8063 merged -- [Transports take shared types](/quest/m2/api-shape/transport.md) - moq-uring takes a congestion factory and rustls configs, and the tokio server request stops exposing its QUIC internals - [JS Key.sign takes exactly Claims](/quest/m2/api-shape/auth-sign.md) - `@moq/auth` signs only `Claims`, with no legacy `put`/`get` input, like Rust diff --git a/quest/m2/api-shape/transport.md b/quest/m2/api-shape/transport.md deleted file mode 100644 index 1f324ac3ef2..00000000000 --- a/quest/m2/api-shape/transport.md +++ /dev/null @@ -1,46 +0,0 @@ -# [S] Transports take shared types - -## Goal - -- moq-uring stops defining its own congestion and TLS types. Its QUIC - configs take a `moq_congestion` factory (BBR3 by default, matching - moq-tokio; today `moq_uring::quic::Congestion` defaults to `Loss`) and an - `Arc` or `Arc`, as moq-tokio's - do. `Identity`, `ClientAuth`, and the client TLS fields are gone. - `moq_uring::quic::Congestion` stays as a convenience that converts into a - factory, so the built-in families keep their names while the config holds - only a factory. The relay's io_uring runtime installs the controller a - `moq_tokio::quic::Config::controller` carries instead of refusing it - (`rs/moq-tokio/src/quic.rs`), so an embedder gets the same controller on - either runtime. -- qmux names no tungstenite type: TLS is qmux's own `tls::Client` config, - `WebSocketConfig` folds into qmux's record-size limits, `Upgraded::new` - takes the raw upgraded IO plus a role, `Error::WebSocket` is opaque, and - `pub use tokio_tungstenite` goes (`rs/qmux/src/ws.rs`), so a tungstenite - major is not a qmux break. -- moq-tokio's `server::Request::quic()` is removed, since it exposes the QUIC - backend's connection through the session API. Its one outside caller, the - relay's `pacer()` (`rs/moq-relay/src/incoming.rs`), clones it to apply - changed bitrate caps after acceptance, which `with_quic_limits` cannot do. - It gets a narrow replacement: a cloneable live-limits handle (for example - `Request::quic_limits() -> Option` with `set(moq_quic::Limits)`), - `None` on qmux streams as today, so the relay keeps updating caps on a - running session without naming the connection. - -## Plan - -Decided 2026-11-07 (maintainer, /quest-plan from the 2026-11-07 API sweep of open PRs and remaining quests): ✅ the -`Congestion` enum converts to a factory, and the config takes only a factory -(rejected: delete the enum; keep the enum with a factory variant). This -reconciles the 2026-11-01 decision that the enum keeps its built-in families -with one factory both runtimes install, and absorbs the deleted m3 -uring-controller quest. qmux's tungstenite exposure is from -[the stream codec bench](/quest/m4/perf/stream/codec.md), which stays -perf-only. - -Found by the 2026-11-07 API shape audit (transport); see the -[epic](/quest/m2/api-shape/README.md). The relay's io_uring path builds its -configs the way the tokio path does, so it loses its conversion code. The -shared sans-IO driver is in [quic-sansio-driver](/quest/m3/quic-sansio-driver.md). -Run the uring benches before and after, since the default controller -changes. diff --git a/quest/m4/perf/stream/codec.md b/quest/m4/perf/stream/codec.md index 1b9ad8c34c4..3243106086c 100644 --- a/quest/m4/perf/stream/codec.md +++ b/quest/m4/perf/stream/codec.md @@ -51,11 +51,8 @@ The subset to implement, from the 2026-10-22 audit: A maintained crate was weighed against this. The codec is small, sits on the hot path, and must write into the driver's buffers, so qmux owns it. -Public API: the `tokio_tungstenite` and `tungstenite` types in -`Client::with_config`, `Client::with_connector`, `Upgraded`, and -`Error::WebSocket` go away. The Impact bullet says how to -migrate in `qmux` and `moq-tokio`. The existing qmux interop tests (the `qmux06` +Public API: none, since qmux already names no tungstenite type. The existing qmux interop tests (the `qmux06` dev-dependency) and the browser path in `js/qmux` must keep passing unchanged. -Changed 2026-11-07 (maintainer, /quest-plan from the API sweep of open PRs and remaining quests): qmux's tungstenite exposure is removed before the cut in [api-shape/transport](/quest/m2/api-shape/transport.md), whatever this bench finds; this quest is perf-only. +Changed 2026-11-07 (maintainer, /quest-plan from the API sweep of open PRs and remaining quests): qmux's tungstenite exposure is removed before the cut (done in [moq-dev/moq#8300](https://github.com/moq-dev/moq/pull/8300)), whatever this bench finds; this quest is perf-only. diff --git a/rs/moq-relay/Cargo.toml b/rs/moq-relay/Cargo.toml index 4628c94cbcf..6c2f71110c9 100644 --- a/rs/moq-relay/Cargo.toml +++ b/rs/moq-relay/Cargo.toml @@ -48,7 +48,7 @@ websocket = ["moq-tokio/websocket", "dep:qmux", "dep:hyper", "dep:tokio-tungsten uds = ["moq-tokio/uds"] # Serve QUIC from io_uring workers (`--runtime-io-uring`), Linux-only. # This is off by default: the relay is useful without io_uring. -io-uring = ["_uring", "moq-uring/quic"] +io-uring = ["_uring", "moq-tokio/driver", "moq-uring/quic"] # The listener machinery shared by the feature above. _uring = ["dep:moq-sock", "dep:moq-uring", "dep:web-transport-proto"] @@ -92,7 +92,7 @@ sysinfo = { version = "0.40", default-features = false, features = ["system"] } thiserror = { workspace = true } tokio = { workspace = true, features = ["full"] } tokio-rustls = { workspace = true } -# Reclaims the socket under a WebSocket upgrade as a WebSocket stream. +# The upgrade's accept key; qmux runs the frames after it. tokio-tungstenite = { workspace = true, features = ["connect", "handshake"], optional = true } toml = { workspace = true } tower-service = "0.3" diff --git a/rs/moq-relay/src/incoming.rs b/rs/moq-relay/src/incoming.rs index 962691c1c08..92e053b195e 100644 --- a/rs/moq-relay/src/incoming.rs +++ b/rs/moq-relay/src/incoming.rs @@ -123,9 +123,9 @@ impl Sealed for moq_tokio::server::Request { #[cfg(feature = "quic")] fn pacer(&self) -> Option { - let quic = moq_tokio::server::Request::quic(self)?.clone(); + let quic = moq_tokio::server::Request::quic_limits(self)?; Some(auth::Pacer::new(move |caps| { - if let Err(err) = quic.set_limits(caps.limits()) { + if let Err(err) = quic.set(caps.limits()) { tracing::warn!(%err, "failed to apply bitrate caps"); } })) diff --git a/rs/moq-relay/src/uring.rs b/rs/moq-relay/src/uring.rs index f7890910a0e..90f2bcd1a7f 100644 --- a/rs/moq-relay/src/uring.rs +++ b/rs/moq-relay/src/uring.rs @@ -484,11 +484,6 @@ fn transport(quic: &moq_tokio::quic::Resolved) -> anyhow::Result anyhow::Result moq_uring::quic::Congestion::Loss, + transport.congestion = match (&quic.controller, quic.congestion_control) { + // The application's own controller wins over a family, as on the tokio path. + (Some(controller), _) => controller.0.clone(), + (None, Some(moq_tokio::quic::CongestionControl::Loss)) => moq_uring::quic::Congestion::Loss.into(), // Unset means the backend's own default, and live media wants a steady // send rate an encoder can track over CUBIC's sawtooth. - Some(moq_tokio::quic::CongestionControl::Delay) | None => moq_uring::quic::Congestion::Delay, - Some(moq_tokio::quic::CongestionControl::None) => moq_uring::quic::Congestion::None, + (None, Some(moq_tokio::quic::CongestionControl::Delay) | None) => moq_uring::quic::Congestion::Delay.into(), + (None, Some(moq_tokio::quic::CongestionControl::None)) => moq_uring::quic::Congestion::None.into(), // A family added to the tokio stack later, rather than a silent // substitution the operator would never learn about. - Some(other) => anyhow::bail!("io_uring workers cannot run the {other:?} congestion controller"), + (None, Some(other)) => anyhow::bail!("io_uring workers cannot run the {other:?} congestion controller"), }; Ok(transport) } diff --git a/rs/moq-relay/src/web.rs b/rs/moq-relay/src/web.rs index 04366628f08..4fae775c457 100644 --- a/rs/moq-relay/src/web.rs +++ b/rs/moq-relay/src/web.rs @@ -2857,13 +2857,7 @@ mod tests { "{}", String::from_utf8_lossy(&response) ); - let socket = tokio_tungstenite::WebSocketStream::from_raw_socket( - stream, - tokio_tungstenite::tungstenite::protocol::Role::Client, - None, - ) - .await; - let (transport, driver) = qmux::ws::Upgraded::new(socket, "qmux-01.moq-lite-06") + let (transport, driver) = qmux::ws::Upgraded::new(stream, "qmux-01.moq-lite-06") .connect() .await .unwrap(); diff --git a/rs/moq-relay/src/websocket.rs b/rs/moq-relay/src/websocket.rs index e2cadb83cca..f5f68ee4b4f 100644 --- a/rs/moq-relay/src/websocket.rs +++ b/rs/moq-relay/src/websocket.rs @@ -2,8 +2,7 @@ use bytes::Bytes; use hyper_util::rt::TokioIo; use std::sync::Arc; use tokio::io::{AsyncRead, AsyncWrite}; -use tokio_tungstenite::WebSocketStream; -use tokio_tungstenite::tungstenite::{self, protocol::Role}; +use tokio_tungstenite::tungstenite; use axum::{ extract::{ConnectInfo, Extension, OriginalUri, Request, State}, @@ -118,15 +117,6 @@ fn switching_protocols(key: &HeaderValue, protocol: &str) -> Response { .expect("every header is a valid value") } -/// The socket config for an upgraded connection: a message is one qmux record, so the socket -/// buffers no more than one before qmux sees it. -fn socket_config() -> tungstenite::protocol::WebSocketConfig { - let max = moq_tokio::websocket::max_message_size(); - tungstenite::protocol::WebSocketConfig::default() - .max_message_size(Some(max)) - .max_frame_size(Some(max)) -} - /// How many upgrades reclaimed the listener's socket, so a test can tell the direct path ran. #[cfg(test)] pub(crate) static RECLAIMED: std::sync::atomic::AtomicUsize = std::sync::atomic::AtomicUsize::new(0); @@ -153,13 +143,23 @@ async fn serve_upgraded( } } -/// Run a MoQ session over the WebSocket on `io`, whose first bytes, `read`, were already read. +/// Run a MoQ session over the WebSocket on `io`, after `read`, the bytes hyper already +/// read past the request. +/// +/// A client sends no frame until it has the 101, so bytes already read past the request +/// came from a peer that broke the handshake, and the connection is dropped. async fn serve_socket(io: S, read: Bytes, options: moq_tokio::websocket::Options, state: Arc) where S: AsyncRead + AsyncWrite + Unpin + Send + 'static, { - let socket = WebSocketStream::from_partially_read(io, read.to_vec(), Role::Server, Some(socket_config())).await; - let handshake = moq_tokio::websocket::accept(socket, options); + if !read.is_empty() { + tracing::debug!( + bytes = read.len(), + "dropped a WebSocket client that sent frames before the upgrade" + ); + return; + } + let handshake = moq_tokio::websocket::accept(io, options); let Some(request) = state.shutdown.until_deadline(handshake).await else { return; }; @@ -253,7 +253,6 @@ fn supported_subprotocols(versions: &moq_net::Versions) -> Vec { mod tests { use super::*; use axum::{Router, routing::any}; - use futures::{Sink, Stream}; use std::{ pin::Pin, sync::atomic::{AtomicBool, Ordering}, @@ -483,8 +482,7 @@ mod tests { let response = switching_protocols(&key, &wire); tokio::spawn(async move { let upgraded = upgrade.await.expect("upgrade"); - let socket = WebSocketStream::from_raw_socket(TokioIo::new(upgraded), Role::Server, None).await; - let (session, driver) = qmux::ws::Upgraded::new(socket, &wire) + let (session, driver) = qmux::ws::Upgraded::new(TokioIo::new(upgraded), &wire) .accept() .await .expect("the upgrade selected a QMux subprotocol"); @@ -617,63 +615,52 @@ mod tests { drop(session); } - /// One leg of an in-memory WebSocket pair. + /// One end of an in-memory connection. /// /// Reading can be frozen: once the flag is set the stream parks forever - /// instead of yielding `None`. That is the failure this is here to model -- + /// instead of yielding data or EOF. That is the failure this is here to model -- /// a peer whose host or network vanished sends no close frame and no FIN, so /// the socket stays readable-but-silent indefinitely. struct Pipe { - incoming: mpsc::UnboundedReceiver, - outgoing: mpsc::UnboundedSender, + io: tokio::io::DuplexStream, frozen: Arc, } impl Pipe { - fn new( - incoming: mpsc::UnboundedReceiver, - outgoing: mpsc::UnboundedSender, - frozen: Arc, - ) -> Self { - Self { - incoming, - outgoing, - frozen, - } + fn new(io: tokio::io::DuplexStream, frozen: Arc) -> Self { + Self { io, frozen } } } - impl Stream for Pipe { - type Item = Result; - - fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll> { + impl AsyncRead for Pipe { + fn poll_read( + mut self: Pin<&mut Self>, + cx: &mut Context<'_>, + buf: &mut tokio::io::ReadBuf<'_>, + ) -> Poll> { if self.frozen.load(Ordering::Relaxed) { return Poll::Pending; } - self.incoming.poll_recv(cx).map(|msg| msg.map(Ok)) + Pin::new(&mut self.io).poll_read(cx, buf) } } - impl Sink for Pipe { - type Error = tungstenite::Error; - - fn poll_ready(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll> { - Poll::Ready(Ok(())) - } - - fn start_send(self: Pin<&mut Self>, message: tungstenite::Message) -> Result<(), Self::Error> { - // A receiver that went away is the peer hanging up, not an error we - // need to surface: the read half reports the close. - let _ = self.outgoing.send(message); - Ok(()) + impl AsyncWrite for Pipe { + fn poll_write(mut self: Pin<&mut Self>, cx: &mut Context<'_>, buf: &[u8]) -> Poll> { + // A peer that went away is hanging up, not an error to surface here: the + // read half reports the close it sent first. + match Pin::new(&mut self.io).poll_write(cx, buf) { + Poll::Ready(Err(err)) if err.kind() == std::io::ErrorKind::BrokenPipe => Poll::Ready(Ok(buf.len())), + other => other, + } } - fn poll_flush(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll> { - Poll::Ready(Ok(())) + fn poll_flush(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll> { + Pin::new(&mut self.io).poll_flush(cx) } - fn poll_close(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll> { - Poll::Ready(Ok(())) + fn poll_shutdown(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll> { + Pin::new(&mut self.io).poll_shutdown(cx) } } @@ -688,8 +675,7 @@ mod tests { #[tokio::test(start_paused = true)] async fn keep_alive_reaps_a_silent_peer() { let alpn = format!("{}{}", preferred_qmux_prefix(), newest_moq_alpn()); - let (client_to_server, server_incoming) = mpsc::unbounded_channel(); - let (server_to_client, client_incoming) = mpsc::unbounded_channel(); + let (server_io, client_io) = tokio::io::duplex(1 << 16); let frozen = Arc::new(AtomicBool::new(false)); let admission = moq_net::server::Admission::default(); @@ -702,7 +688,7 @@ mod tests { ); options.link.alpn = Some(alpn.clone()); let server = tokio::spawn(moq_tokio::websocket::accept( - Pipe::new(server_incoming, server_to_client, frozen.clone()), + Pipe::new(server_io, frozen.clone()), options, )); @@ -710,13 +696,10 @@ mod tests { // timeout is out of the picture before we go silent. It never speaks moq, // so the server is parked awaiting SETUP -- exactly where an idle // publisher's session sits between groups. - let (client, driver) = qmux::ws::Upgraded::new( - Pipe::new(client_incoming, client_to_server, Arc::new(AtomicBool::new(false))), - &alpn, - ) - .connect() - .await - .unwrap(); + let (client, driver) = qmux::ws::Upgraded::new(Pipe::new(client_io, Arc::new(AtomicBool::new(false))), &alpn) + .connect() + .await + .unwrap(); tokio::spawn(driver); // Paused time only advances once every task is idle, so this resolves @@ -742,8 +725,7 @@ mod tests { use moq_net::transport::Error as _; let alpn = format!("{}{}", preferred_qmux_prefix(), newest_moq_alpn()); - let (client_to_server, server_incoming) = mpsc::unbounded_channel(); - let (server_to_client, client_incoming) = mpsc::unbounded_channel(); + let (server_io, client_io) = tokio::io::duplex(1 << 16); let timeout = Duration::from_secs(10); let admission = moq_net::server::Admission::default(); @@ -757,18 +739,15 @@ mod tests { options.link.alpn = Some(alpn.clone()); let start = kio::clock::now(); let server = tokio::spawn(moq_tokio::websocket::accept( - Pipe::new(server_incoming, server_to_client, Arc::new(AtomicBool::new(false))), + Pipe::new(server_io, Arc::new(AtomicBool::new(false))), options, )); // A live qmux peer that answers keep-alives but never speaks moq. - let (client, driver) = qmux::ws::Upgraded::new( - Pipe::new(client_incoming, client_to_server, Arc::new(AtomicBool::new(false))), - &alpn, - ) - .connect() - .await - .unwrap(); + let (client, driver) = qmux::ws::Upgraded::new(Pipe::new(client_io, Arc::new(AtomicBool::new(false))), &alpn) + .connect() + .await + .unwrap(); tokio::spawn(driver); let err = server diff --git a/rs/moq-tokio/Cargo.toml b/rs/moq-tokio/Cargo.toml index 1537866f4e8..7702744cf5e 100644 --- a/rs/moq-tokio/Cargo.toml +++ b/rs/moq-tokio/Cargo.toml @@ -39,7 +39,7 @@ qlog = ["moq-quic?/qlog"] # LAN peer discovery over mDNS (the `mdns` module). mdns = ["dep:mdns-sd"] jemalloc = ["dep:tikv-jemallocator", "dep:tikv-jemalloc-ctl"] -websocket = ["qmux", "watch", "dep:dns-lookup", "dep:tokio-tungstenite"] +websocket = ["qmux", "watch", "dep:dns-lookup", "dep:httparse", "dep:tokio-rustls", "dep:tokio-tungstenite"] # Qmux over TCP: plaintext `tcp://`, and `tls://`. Server-side and client-side. tcp = ["qmux", "dep:dns-lookup", "dep:tokio-rustls"] # MoQ over a qmux session established elsewhere, such as moq-webrtc's data channel: @@ -63,6 +63,8 @@ dns-lookup = { version = "4", optional = true } futures = "0.3" hex = { workspace = true } http = { version = "1", optional = true } +# Parses the WebSocket upgrade response without reading past it. +httparse = { workspace = true, optional = true } humantime = { workspace = true } humantime-serde = { workspace = true } kio = { workspace = true, features = ["async", "runtime"] } @@ -95,8 +97,8 @@ time = { workspace = true } tokio = { workspace = true, features = ["full"] } # The TLS stream `tcp::TlsAcceptor` accepts from a listener sharing its port. tokio-rustls = { workspace = true, optional = true } -# The WebSocket fallback dials and accepts with its own rustls config. -tokio-tungstenite = { workspace = true, features = ["handshake", "rustls-tls-native-roots"], optional = true } +# The WebSocket upgrade's HTTP handshake; qmux runs the frames after it. +tokio-tungstenite = { workspace = true, features = ["handshake"], optional = true } toml = { workspace = true } tracing = "0.1" tracing-subscriber = { version = "0.3", features = ["env-filter"] } diff --git a/rs/moq-tokio/src/quic.rs b/rs/moq-tokio/src/quic.rs index 56249f3f498..88d30fe90ac 100644 --- a/rs/moq-tokio/src/quic.rs +++ b/rs/moq-tokio/src/quic.rs @@ -253,7 +253,7 @@ pub struct Config { /// [`congestion_control`](Self::congestion_control)'s families. /// /// Code-only, since a controller has no text form: no flag or setting reaches it. - /// The QUIC backend runs it, and the io_uring runtime refuses it. + /// Both the QUIC backend and the relay's io_uring runtime run it. #[cfg(feature = "driver")] #[serde(skip)] #[usage(skip)] @@ -721,23 +721,6 @@ impl Resolved { self.qlog.as_deref() } - /// Whether an application's own congestion controller is installed. - /// - /// Compiled in every feature set, unlike the `controller` field, so a runtime - /// that cannot run one, such as the relay's io_uring workers, can refuse it - /// whichever features unify onto this crate. - #[cfg(feature = "driver")] - pub fn has_controller(&self) -> bool { - self.controller.is_some() - } - - /// Whether an application's own congestion controller is installed: never, - /// since a build without the `driver` feature has no way to install one. - #[cfg(not(feature = "driver"))] - pub fn has_controller(&self) -> bool { - false - } - /// The keep-alive interval against a peer whose idle timeout is no shorter than /// ours, using moq-quic's 30s default (a 9s interval) when neither peer sets one. #[cfg(feature = "_transport")] diff --git a/rs/moq-tokio/src/server.rs b/rs/moq-tokio/src/server.rs index 0551fcf2cee..1834a1b800d 100644 --- a/rs/moq-tokio/src/server.rs +++ b/rs/moq-tokio/src/server.rs @@ -1493,6 +1493,29 @@ pub struct Link { pub alpn: Option, } +/// Changes one accepted QUIC connection's limits while its session runs, from +/// [`Request::quic_limits`]. Cloning shares the connection. +#[cfg(feature = "quic")] +#[derive(Clone)] +pub struct QuicLimits(crate::quic::driver::connection::Connection); + +#[cfg(feature = "quic")] +impl QuicLimits { + /// Update the connection's limits, as [`Request::with_quic_limits`] does before SETUP. + /// + /// A `None` field keeps its current limit; lift a rate cap with `moq_quic::Rate::Unlimited`. + pub fn set(&self, limits: moq_quic::Limits) -> Result<(), moq_quic::LimitsError> { + self.0.set_limits(limits) + } +} + +#[cfg(feature = "quic")] +impl std::fmt::Debug for QuicLimits { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + f.debug_struct("QuicLimits").finish_non_exhaustive() + } +} + /// An incoming MoQ session that can be accepted or rejected. /// /// The transport connection and the MoQ SETUP are already complete, so [`path`](Self::path), @@ -1608,12 +1631,13 @@ impl Request { self } - /// The QUIC connection under this request, to change its limits once the session runs. - /// `None` on a transport moq-quic does not drive (qmux streams). + /// A handle that changes this connection's QUIC limits while the session runs, such as + /// bitrate caps that change after acceptance. `None` on a transport moq-quic does not + /// drive (qmux streams). #[cfg(feature = "quic")] - pub fn quic(&self) -> Option<&crate::quic::driver::connection::Connection> { + pub fn quic_limits(&self) -> Option { match &self.under { - Under::Quic(quic) => Some(quic), + Under::Quic(quic) => Some(QuicLimits(quic.clone())), #[allow(unreachable_patterns, reason = "a QUIC-only build has no other connection")] _ => None, } @@ -2183,12 +2207,8 @@ mod tests { /// paused clock. #[cfg(feature = "websocket")] async fn session_pair() -> (qmux::Session, qmux::Session) { - use tokio_tungstenite::{WebSocketStream, tungstenite::protocol::Role}; - const ALPN: &str = "qmux-01.moq-lite-06"; let (client, server) = tokio::io::duplex(64 * 1024); - let client = WebSocketStream::from_raw_socket(client, Role::Client, None).await; - let server = WebSocketStream::from_raw_socket(server, Role::Server, None).await; let (client, server) = tokio::join!( qmux::ws::Upgraded::new(client, ALPN).connect(), qmux::ws::Upgraded::new(server, ALPN).accept(), @@ -2201,11 +2221,7 @@ mod tests { #[tokio::test(start_paused = true)] async fn upgraded_websocket_preserves_request_metadata() { use moq_net::transport::poll::{SendOptions, SendStream as _, Session as _}; - use tokio_tungstenite::{WebSocketStream, tungstenite::protocol::Role}; - let (client, server) = tokio::io::duplex(64 * 1024); - let client = WebSocketStream::from_raw_socket(client, Role::Client, None).await; - let server = WebSocketStream::from_raw_socket(server, Role::Server, None).await; let admission = moq_net::server::Admission::default(); let url = "wss://example.com/room?jwt=token".parse().unwrap(); let mut options = crate::websocket::Options::new(url, moq_net::Server::new(), admission.enter().unwrap()); @@ -2245,13 +2261,9 @@ mod tests { #[tokio::test(start_paused = true)] async fn upgraded_websocket_refuses_setup_path_presence() { use moq_net::transport::poll::{SendOptions, SendStream as _, Session as _}; - use tokio_tungstenite::{WebSocketStream, tungstenite::protocol::Role}; - for protocol in ["qmux-01.moq-lite-06", "qmux-01.moqt-18", "qmux-01.moqt-22"] { for path in ["", "/room?jwt=url-token", "/other?jwt=setup-token"] { let (client, server) = tokio::io::duplex(64 * 1024); - let client = WebSocketStream::from_raw_socket(client, Role::Client, None).await; - let server = WebSocketStream::from_raw_socket(server, Role::Server, None).await; let admission = moq_net::server::Admission::default(); let mut options = crate::websocket::Options::new( "wss://example.com/room?jwt=url-token".parse().unwrap(), @@ -2923,7 +2935,10 @@ mod tests { let _connection = client.with_reconnect(false).connect(url); let mut request = listener.accept().await.expect("no incoming connection"); - let quic = request.quic().expect("a QUIC request").clone(); + let Under::Quic(quic) = &request.under else { + panic!("a QUIC request"); + }; + let quic = quic.clone(); assert_eq!( quic.receive_window().connection, std::num::NonZeroU64::new(moq_net::transport::SETUP_RECEIVE_WINDOW) diff --git a/rs/moq-tokio/src/tls.rs b/rs/moq-tokio/src/tls.rs index ce64a285044..d95fedc769e 100644 --- a/rs/moq-tokio/src/tls.rs +++ b/rs/moq-tokio/src/tls.rs @@ -3669,6 +3669,27 @@ mod tests { assert!(client_verifier.verify_client_cert(&client_b, &[], now).is_ok()); } + /// A root file holding several CAs trusts every one of them, not only the first. + #[cfg(all(feature = "watch", feature = "quic"))] + #[test] + fn a_root_bundle_is_loaded_whole() { + let (unrelated, _, _, _, _) = signed_certificates(); + let (signer, server, _, _, _) = signed_certificates(); + let dir = tempfile::TempDir::new().unwrap(); + let bundle = dir.path().join("bundle.pem"); + std::fs::write(&bundle, format!("{unrelated}{signer}")).unwrap(); + + let roots = read_roots(&[bundle]).unwrap(); + assert_eq!(roots.len(), 2); + let verifier = Connect::build_root_server_verifier(&roots, false, &crypto::provider().unwrap()).unwrap(); + let name = ServerName::try_from("localhost").unwrap(); + assert!( + verifier + .verify_server_cert(&server, &[], &name, &[], UnixTime::now()) + .is_ok() + ); + } + #[test] fn build_uses_platform_verifier_by_default() { // No custom roots, system trust on: resolves to the OS platform verifier diff --git a/rs/moq-tokio/src/websocket.rs b/rs/moq-tokio/src/websocket.rs index f51c61acbe5..6a535f5873f 100644 --- a/rs/moq-tokio/src/websocket.rs +++ b/rs/moq-tokio/src/websocket.rs @@ -10,6 +10,7 @@ use std::collections::HashSet; use std::sync::{Arc, LazyLock, Mutex}; use std::{net, time}; +use tokio::io::{AsyncRead, AsyncReadExt, AsyncWrite, AsyncWriteExt}; use tokio_tungstenite::tungstenite::{self, client::IntoClientRequest, http}; use url::Url; @@ -141,32 +142,14 @@ impl Options { } } -/// The largest WebSocket message or frame [`accept`] reads: one qmux record. +/// Read SETUP on a connection an HTTP server upgraded to WebSocket, leaving admission +/// to the returned request. /// -/// Cap the socket at this before upgrading it, so a peer cannot make it buffer more. -pub fn max_message_size() -> usize { - qmux::ws::max_message_size(&crate::qmux_default()) -} - -/// A socket config capped at [`max_message_size`], for the sockets upgraded here. -fn socket_config() -> tungstenite::protocol::WebSocketConfig { - let max = max_message_size(); - tungstenite::protocol::WebSocketConfig::default() - .max_message_size(Some(max)) - .max_frame_size(Some(max)) -} - -/// Read SETUP on an upgraded WebSocket, leaving admission to the returned request. -/// -/// The accept deadline runs from here, set by `options`' server -/// ([`moq_net::Server::with_timeout`]). Cap the socket at [`max_message_size`] before the upgrade. -pub async fn accept(socket: T, options: Options) -> crate::Result +/// `io` is the raw upgraded connection, past the 101 response. The accept deadline runs +/// from here, set by `options`' server ([`moq_net::Server::with_timeout`]). +pub async fn accept(io: T, options: Options) -> crate::Result where - T: futures::Stream> - + futures::Sink - + Send - + Unpin - + 'static, + T: AsyncRead + AsyncWrite + Send + Unpin + 'static, { crate::runtime::scope(async move { let deadline = options.server.deadline(kio::clock::now()); @@ -174,7 +157,7 @@ where // lifetime that QUIC's idle timeout bounds on the native path. // The subprotocol names the QMux version; an upgrade without one has no wire format. let alpn = options.link.alpn.as_deref().unwrap_or_default(); - let mut upgraded = qmux::ws::Upgraded::new(socket, alpn) + let mut upgraded = qmux::ws::Upgraded::new(io, alpn) .with_keep_alive(qmux::ws::KeepAlive::default()) .with_session_config(crate::qmux_config(&options.quic)); if let Some(stats) = &options.socket_stats { @@ -446,21 +429,15 @@ pub(crate) async fn connect( .await? } _ => { - // Use the existing TLS config (which respects tls-disable-verify) for secure connections. - let connector = if needs_tls { - tokio_tungstenite::Connector::Rustls(Arc::new(tls.clone())) - } else { - tokio_tungstenite::Connector::Plain - }; - // Most moq ALPNs can ride on any QMux draft (`&[]` lets the polyfill expand // to every version it knows). `qmux_versions_for` pins the few that the spec // restricts. qmux also offers the bare ALPNs (`qmux-02`, `qmux-01`, // `qmux-00`) by default so we still interop with relays that only know a // wire-format version. + // The existing TLS config (which respects tls-disable-verify) for secure connections. qmux::ws::Client::new() .with_protocols(alpns.iter().map(|&a| (a, qmux_versions_for(a)))) - .with_connector(connector) + .with_tls(Arc::new(tls.clone())) .with_keep_alive(qmux::ws::KeepAlive::default()) // 5s ping / 30s deadline; TCP backs off retransmits too far for QUIC's 10s .with_session_config(session.clone()) .connect(url.as_str()) @@ -485,46 +462,27 @@ async fn connect_tls_override( alpns: &[&str], session: &qmux::Config, ) -> Result { - let original_request = url - .as_str() - .into_client_request() - .map_err(|err| Error::BuildRequest(crate::error::message(err)))?; - let original_host = original_request - .headers() - .get(http::header::HOST) - .cloned() - .ok_or(Error::MissingHostname)?; - // tokio-tungstenite takes the TLS name from the URL host, so a bare IPv6 override - // needs the URL brackets that `set_host` refuses to add. - let mut tls_url = url.clone(); - match tls_host_name.parse::() { - Ok(ip) => tls_url.set_ip_host(ip), - Err(_) => tls_url.set_host(Some(tls_host_name)).map_err(|_| ()), - } - .map_err(|_| Error::BuildRequest(format!("invalid TLS host name: {tls_host_name}")))?; - let mut request = tls_url + let mut request = url .as_str() .into_client_request() .map_err(|err| Error::BuildRequest(crate::error::message(err)))?; - request.headers_mut().insert(http::header::HOST, original_host); let protocols = supported_subprotocols(alpns).join(", "); request.headers_mut().insert( http::header::SEC_WEBSOCKET_PROTOCOL, http::HeaderValue::from_str(&protocols).map_err(|err| Error::ProtocolHeader(crate::error::message(err)))?, ); - let connector = tokio_tungstenite::Connector::Rustls(Arc::new(tls.clone())); - let (websocket, response) = - tokio_tungstenite::client_async_tls_with_config(request, stream, Some(socket_config()), Some(connector)) - .await - .map_err(qmux::Error::from) - .map_err(Error::connect)?; + // URL brackets delimit IPv6 literals, but aren't part of the TLS server name. + let bare = tls_host_name.trim_start_matches('[').trim_end_matches(']'); + let name = rustls::pki_types::ServerName::try_from(bare.to_string()) + .map_err(|_| Error::BuildRequest(format!("invalid TLS host name: {tls_host_name}")))?; + let mut stream = tokio_rustls::TlsConnector::from(Arc::new(tls.clone())) + .connect(name, stream) + .await + .map_err(|err| Error::WebSocketConnect(crate::error::message(err)))?; + let negotiated = upgrade(&mut stream, &request).await?; - let negotiated = response - .headers() - .get(http::header::SEC_WEBSOCKET_PROTOCOL) - .and_then(|value| value.to_str().ok()); - qmux::ws::Upgraded::new(websocket, negotiated.unwrap_or_default()) + qmux::ws::Upgraded::new(stream, &negotiated) .with_keep_alive(qmux::ws::KeepAlive::default()) .with_session_config(session.clone()) .connect() @@ -533,6 +491,96 @@ async fn connect_tls_override( .map_err(Error::connect) } +/// The most response header bytes [`upgrade`] reads before giving up on the server. +const MAX_RESPONSE: usize = 16 * 1024; + +/// Send the WebSocket upgrade `request` on `stream` and read the 101 that answers it, +/// returning the subprotocol the server chose from those the request offered. +/// +/// The response is read a byte at a time so nothing past it is consumed: the server may +/// send QMux frames right behind its 101, and those belong to the session. +async fn upgrade(stream: &mut S, request: &http::Request<()>) -> Result { + let key = request + .headers() + .get(http::header::SEC_WEBSOCKET_KEY) + .expect("a client request carries its key") + .clone(); + let path = request.uri().path_and_query().map_or("/", |path| path.as_str()); + let mut head = format!("GET {path} HTTP/1.1\r\n").into_bytes(); + for (name, value) in request.headers() { + head.extend_from_slice(name.as_str().as_bytes()); + head.extend_from_slice(b": "); + head.extend_from_slice(value.as_bytes()); + head.extend_from_slice(b"\r\n"); + } + head.extend_from_slice(b"\r\n"); + stream.write_all(&head).await?; + stream.flush().await?; + + let mut response = Vec::new(); + while !response.ends_with(b"\r\n\r\n") { + if response.len() == MAX_RESPONSE { + return Err(Error::WebSocketConnect("upgrade response headers too large".into())); + } + response.push(stream.read_u8().await?); + } + + let mut headers = [httparse::EMPTY_HEADER; tungstenite::handshake::headers::MAX_HEADERS]; + let mut parsed = httparse::Response::new(&mut headers); + parsed + .parse(&response) + .map_err(|err| Error::WebSocketConnect(format!("invalid upgrade response: {err}")))?; + let status = parsed.code.unwrap_or_default(); + if status != 101 { + return Err(Error::Connect { + message: format!("upgrade refused with HTTP {status}"), + status: Some(status), + }); + } + let invalid = |reason: &str| Err(Error::WebSocketConnect(format!("invalid upgrade response: {reason}"))); + if parsed.version != Some(1) { + return invalid("not HTTP/1.1"); + } + let header = |name: http::header::HeaderName| { + parsed + .headers + .iter() + .find(|header| header.name.eq_ignore_ascii_case(name.as_str())) + .and_then(|header| std::str::from_utf8(header.value).ok()) + }; + if !header(http::header::UPGRADE).is_some_and(|value| value.trim().eq_ignore_ascii_case("websocket")) { + return invalid("no Upgrade: websocket"); + } + // `Connection` may arrive as several fields; the upgrade token can sit in any of them. + let upgrades = parsed + .headers + .iter() + .filter(|header| header.name.eq_ignore_ascii_case(http::header::CONNECTION.as_str())) + .filter_map(|header| std::str::from_utf8(header.value).ok()) + .flat_map(|value| value.split(',')) + .any(|token| token.trim().eq_ignore_ascii_case("upgrade")); + if !upgrades { + return invalid("no Connection: upgrade"); + } + let accept = tungstenite::handshake::derive_accept_key(key.as_bytes()); + if header(http::header::SEC_WEBSOCKET_ACCEPT) != Some(accept.as_str()) { + return invalid("wrong accept key"); + } + + // The subprotocol pairs the QMux draft with the moq version, so one the request never + // offered could pair them in a way we refuse, such as moqt-22 over qmux-00. + let offered = request + .headers() + .get(http::header::SEC_WEBSOCKET_PROTOCOL) + .and_then(|value| value.to_str().ok()) + .unwrap_or_default(); + match header(http::header::SEC_WEBSOCKET_PROTOCOL).map(str::trim) { + Some(protocol) if offered.split(',').any(|offer| offer.trim() == protocol) => Ok(protocol.to_string()), + Some(_) => invalid("subprotocol was not offered"), + None => invalid("no subprotocol"), + } +} + /// The QMux drafts a moq ALPN is allowed to ride on, for `qmux::ws::Client::with_protocols`. /// /// moq-transport-18 and newer require qmux-01, so we never pair them with qmux-00. @@ -734,16 +782,18 @@ impl Listener { }; async move { - let websocket = tokio_tungstenite::accept_hdr_async_with_config(stream, callback, Some(socket_config())) + // The client waits for the 101 before it sends a frame, so the handshake reads + // nothing past the request and the stream underneath holds only frames. + let stream = tokio_tungstenite::accept_hdr_async(stream, callback) .await - .map_err(qmux::Error::from) - .map_err(Error::accept)?; + .map_err(|err| Error::Accept(crate::error::message(err)))? + .into_inner(); let (protocol, url) = accepted .lock() .unwrap() .take() .expect("successful upgrade selected a protocol"); - let session = qmux::ws::Upgraded::new(websocket, &protocol) + let session = qmux::ws::Upgraded::new(stream, &protocol) .with_keep_alive(qmux::ws::KeepAlive::default()) .with_session_config(session) .accept() @@ -980,10 +1030,13 @@ mod tests { } Ok(response) }; - let websocket = tokio_tungstenite::accept_hdr_async(tls, callback).await.unwrap(); + let tls = tokio_tungstenite::accept_hdr_async(tls, callback) + .await + .unwrap() + .into_inner(); let protocol = protocol.lock().unwrap().take().unwrap(); // The client returns once QMux is established, so answer with its parameters. - let (session, driver) = qmux::ws::Upgraded::new(websocket, &protocol).accept().await.unwrap(); + let (session, driver) = qmux::ws::Upgraded::new(tls, &protocol).accept().await.unwrap(); tokio::spawn(driver); let host = host.lock().unwrap().take(); (server_name, host, session) @@ -1019,14 +1072,10 @@ mod tests { /// the `[quic]` windows its [`Options`] carry rather than the defaults. #[tokio::test] async fn an_upgraded_session_starts_at_the_options_windows() { - use tungstenite::protocol::Role; const TIMEOUT: time::Duration = time::Duration::from_secs(10); let alpn = supported_subprotocols(moq_net::ALPNS).remove(0); - let (server_io, client_io) = tokio::io::duplex(1 << 20); - let server_ws = - tokio_tungstenite::WebSocketStream::from_raw_socket(server_io, Role::Server, Some(socket_config())).await; - let client_ws = tokio_tungstenite::WebSocketStream::from_raw_socket(client_io, Role::Client, None).await; + let (server_ws, client_ws) = tokio::io::duplex(1 << 20); let quic = crate::quic::Config { receive_window: Some(moq_net::transport::MIN_RECEIVE_WINDOW * 2), @@ -1067,6 +1116,99 @@ mod tests { client.abort(); } + /// Answer an upgrade offering only `qmux-01.moqt-22` with `response`, followed by + /// `trailer`, and return what [`upgrade`] made of it plus the bytes left on the stream. + async fn check_upgrade(response: impl Fn(&str) -> String, trailer: &[u8]) -> (Result, Vec) { + let (result, rest, _) = check_upgrade_request(response, trailer).await; + (result, rest) + } + + /// [`check_upgrade`], also returning the request head [`upgrade`] wrote. + async fn check_upgrade_request( + response: impl Fn(&str) -> String, + trailer: &[u8], + ) -> (Result, Vec, String) { + let mut request = "ws://relay.example/room?jwt=token".into_client_request().unwrap(); + request.headers_mut().insert( + http::header::SEC_WEBSOCKET_PROTOCOL, + http::HeaderValue::from_static("qmux-01.moqt-22"), + ); + let key = request.headers()[http::header::SEC_WEBSOCKET_KEY].clone(); + let accept = tungstenite::handshake::derive_accept_key(key.as_bytes()); + + let (mut client, mut server) = tokio::io::duplex(1 << 16); + server.write_all(response(&accept).as_bytes()).await.unwrap(); + server.write_all(trailer).await.unwrap(); + let result = upgrade(&mut client, &request).await; + let mut rest = vec![0; trailer.len()]; + client.read_exact(&mut rest).await.unwrap(); + drop(client); + let mut head = String::new(); + server.read_to_string(&mut head).await.unwrap(); + (result, rest, head) + } + + fn switching(accept: &str, protocol: &str) -> String { + format!( + "HTTP/1.1 101 Switching Protocols\r\nUpgrade: websocket\r\nConnection: Upgrade\r\nSec-WebSocket-Accept: {accept}\r\nSec-WebSocket-Protocol: {protocol}\r\n\r\n" + ) + } + + #[tokio::test] + async fn upgrade_leaves_frames_behind_the_101() { + let (result, rest) = check_upgrade(|accept| switching(accept, "qmux-01.moqt-22"), b"frames").await; + assert_eq!(result.unwrap(), "qmux-01.moqt-22"); + assert_eq!(rest, b"frames"); + } + + #[tokio::test] + async fn upgrade_requests_the_path_and_query() { + let (result, _, head) = check_upgrade_request(|accept| switching(accept, "qmux-01.moqt-22"), b"").await; + result.unwrap(); + assert!(head.starts_with("GET /room?jwt=token HTTP/1.1\r\n"), "{head}"); + } + + #[tokio::test] + async fn upgrade_finds_the_token_in_a_later_connection_field() { + let (result, _) = check_upgrade( + |accept| { + switching(accept, "qmux-01.moqt-22") + .replace("Connection: Upgrade", "Connection: keep-alive\r\nConnection: Upgrade") + }, + b"", + ) + .await; + assert_eq!(result.unwrap(), "qmux-01.moqt-22"); + } + + #[tokio::test] + async fn upgrade_refuses_an_unoffered_subprotocol() { + let (result, _) = check_upgrade(|accept| switching(accept, "qmux-00.moqt-22"), b"").await; + assert!(result.is_err()); + } + + #[tokio::test] + async fn upgrade_refuses_a_missing_subprotocol() { + let (result, _) = check_upgrade( + |accept| { + format!("HTTP/1.1 101 Switching Protocols\r\nUpgrade: websocket\r\nConnection: Upgrade\r\nSec-WebSocket-Accept: {accept}\r\n\r\n") + }, + b"", + ) + .await; + assert!(result.is_err()); + } + + #[tokio::test] + async fn upgrade_refuses_a_101_that_is_not_a_websocket() { + let (result, _) = check_upgrade( + |accept| switching(accept, "qmux-01.moqt-22").replace("Upgrade: websocket", "Upgrade: h2c"), + b"", + ) + .await; + assert!(result.is_err()); + } + #[test] fn moqt_18_and_newer_pin_to_qmux01() { // The literals in `qmux_versions_for` must stay the IETF draft ALPNs; @@ -1169,9 +1311,6 @@ mod timeout_tests { ); options.link.alpn = Some(ALPN.into()); let (server, _peer) = tokio::io::duplex(4096); - let server = - tokio_tungstenite::WebSocketStream::from_raw_socket(server, tungstenite::protocol::Role::Server, None) - .await; let start = kio::clock::now(); let result = accept(server, options).await; assert!( @@ -1197,12 +1336,6 @@ mod timeout_tests { // Complete QMux over an in-memory WebSocket, then withhold MoQ SETUP. // A silent transport would expire QMux's own parameter deadline first. let (server, peer) = tokio::io::duplex(4096); - let server = - tokio_tungstenite::WebSocketStream::from_raw_socket(server, tungstenite::protocol::Role::Server, None) - .await; - let peer = - tokio_tungstenite::WebSocketStream::from_raw_socket(peer, tungstenite::protocol::Role::Client, None) - .await; let mut accept = Box::pin(accept(server, options)); // Each end waits for the other's QMux parameters, so the accept runs meanwhile. let peer = tokio::select! { diff --git a/rs/moq-uring/Cargo.toml b/rs/moq-uring/Cargo.toml index 3dc6bb2ae69..38daa99a52c 100644 --- a/rs/moq-uring/Cargo.toml +++ b/rs/moq-uring/Cargo.toml @@ -21,7 +21,7 @@ doc = false [features] default = ["quic"] # QUIC and WebTransport on moq-quic, with TLS through rustls. -quic = ["dep:moq-congestion", "dep:moq-crypto", "dep:moq-quic", "dep:rustls", "dep:rustls-platform-verifier"] +quic = ["dep:moq-congestion", "dep:moq-crypto", "dep:moq-quic", "dep:rustls"] # Capture qlog traces (`quic::Transport::qlog`). Off by default: it compiles the # qlog crate and its serde_json encoder into moq-quic, which a production build has # no use for. @@ -49,9 +49,6 @@ moq-quic = { workspace = true, features = [], optional = true } moq-sock = { workspace = true } rustc-hash = { workspace = true } rustls = { version = "0.23", default-features = false, features = ["std"], optional = true } -# OS-native server-cert verifier for the rustls backends. Version matched to -# the copy moq-tokio pulls in, to avoid duplication. -rustls-platform-verifier = { version = "0.7", optional = true } # Runtime thread checks for the session moq-transport's `Send` drivers capture. send_wrapper = { workspace = true } slab = "0.4" diff --git a/rs/moq-uring/benches/echo_quic.rs b/rs/moq-uring/benches/echo_quic.rs index d67d5770e53..62558c0aa20 100644 --- a/rs/moq-uring/benches/echo_quic.rs +++ b/rs/moq-uring/benches/echo_quic.rs @@ -50,12 +50,7 @@ mod linux { let mut worker = Worker::new(Config::default()).expect("worker"); let handle = worker.handle(); let certs = support::certs().expect("certificates"); - let server = quic::server::Config::new( - quic::Identity::open(&certs.cert, &certs.key) - .expect("identity") - .server(&[moq_tokio::quic::web::ALPN], quic::server::ClientAuth::None) - .expect("tls"), - ); + let server = quic::server::Config::new(certs.server(&[moq_tokio::quic::web::ALPN])); let socket = handle .udp(UdpSocket::bind("127.0.0.1:0").expect("bind"), udp_config) .expect("socket"); diff --git a/rs/moq-uring/benches/session_lite.rs b/rs/moq-uring/benches/session_lite.rs index 5b11bf9cad5..483e4ff772b 100644 --- a/rs/moq-uring/benches/session_lite.rs +++ b/rs/moq-uring/benches/session_lite.rs @@ -124,12 +124,7 @@ mod linux { .expect("create track"); let certs = support::certs().expect("certificates"); - let server_config = quic::server::Config::new( - quic::Identity::open(&certs.cert, &certs.key) - .expect("identity") - .server(&[ALPN], quic::server::ClientAuth::None) - .expect("tls"), - ); + let server_config = quic::server::Config::new(certs.server(&[ALPN])); let server_sock = handle .udp(UdpSocket::bind("127.0.0.1:0").expect("bind"), ablation.config.clone()) @@ -138,9 +133,7 @@ mod linux { let client_sock = handle .udp(UdpSocket::bind("127.0.0.1:0").expect("bind"), ablation.config.clone()) .expect("client socket"); - let mut dial = quic::client::Config::default(); - dial.alpn = vec![ALPN.to_string()]; - dial.verify = false; + let dial = quic::client::Config::new(support::client_tls(&[ALPN])); let server_handle = handle.clone(); handle.spawn(async move { diff --git a/rs/moq-uring/benches/unreachable_dials.rs b/rs/moq-uring/benches/unreachable_dials.rs index 2a7907cf8af..d118d3dd85c 100644 --- a/rs/moq-uring/benches/unreachable_dials.rs +++ b/rs/moq-uring/benches/unreachable_dials.rs @@ -23,11 +23,12 @@ mod linux { type Dial<'a> = Pin> + 'a>>; fn dial_config() -> quic::client::Config { - let mut config = quic::client::Config::default(); - config.alpn = vec!["moq-uring-bench".to_string()]; - config.verify = false; - config.system_roots = false; - config + // Nothing ever answers, so nothing is verified. + let mut tls = moq_tokio::tls::Connect::default(); + tls.insecure = Some(true); + let mut tls = tls.build().expect("tls"); + tls.alpn_protocols = vec![b"moq-uring-bench".to_vec()]; + quic::client::Config::new(std::sync::Arc::new(tls)) } /// An IPv6-only socket, which Linux refuses to send to a v4-mapped address diff --git a/rs/moq-uring/benches/worker_turns.rs b/rs/moq-uring/benches/worker_turns.rs index c34772da99d..dc01c9cd66c 100644 --- a/rs/moq-uring/benches/worker_turns.rs +++ b/rs/moq-uring/benches/worker_turns.rs @@ -23,10 +23,7 @@ fn request_reply(c: &mut criterion::Criterion) { const ALPN: &str = "worker-turns"; let _ = rind_rustls::default_provider().unwrap().install_default(); let certs = support::certs().expect("certificates"); - let tls = quic::Identity::open(&certs.cert, &certs.key) - .expect("identity") - .server(&[ALPN], quic::server::ClientAuth::None) - .expect("tls"); + let tls = certs.server(&[ALPN]); let mut group = c.benchmark_group("worker_request_reply"); for connections in [1, 16] { for bytes in [64, 256 * 1024] { @@ -43,9 +40,7 @@ fn request_reply(c: &mut criterion::Criterion) { ) .expect("server"); let client = quic::Endpoint::new(quic::endpoint::Config::default(), socket()).expect("client"); - let mut dial = quic::client::Config::default(); - dial.alpn = vec![ALPN.to_owned()]; - dial.verify = false; + let dial = quic::client::Config::new(support::client_tls(&[ALPN])); for _ in 0..connections { let server = server.clone(); handle.spawn(async move { diff --git a/rs/moq-uring/src/quic/client.rs b/rs/moq-uring/src/quic/client.rs index 387e9f4f572..fbede02fdf7 100644 --- a/rs/moq-uring/src/quic/client.rs +++ b/rs/moq-uring/src/quic/client.rs @@ -1,44 +1,31 @@ //! Dialing: everything one outgoing connection needs. use std::net::SocketAddr; +use std::sync::Arc; -use super::{Connection, Error, Identity}; +use super::{Connection, Error}; use crate::udp; -/// How to dial: as whom, and who to trust. +/// How to dial: the TLS policy and the transport settings. /// /// Where to dial is an argument to [`connect`] and /// [`Endpoint::connect`](super::Endpoint::connect), so one config serves every peer. #[derive(Clone, Debug)] #[non_exhaustive] pub struct Config { - /// ALPN protocols to offer, in preference order. Required: QUIC without an - /// application protocol is refused. - pub alpn: Vec, - /// Verify the server's certificate against the roots below (default). - /// Turn it off only for tests or a pinned local deployment. - pub verify: bool, - /// PEM root certificate files to trust, on top of the system store. - pub roots: Vec, - /// Trust the platform's root store as well, on by default. Turn it off to - /// trust only [`roots`](Self::roots), which is a real restriction here: - /// the trust store is built from scratch rather than added to. - pub system_roots: bool, - /// A certificate to present when the server asks for one (mTLS). - pub identity: Option, + /// The TLS policy every dial uses: who to trust, any client certificate, + /// and the ALPNs to offer. It must allow TLS 1.3 and name at least one ALPN. + pub tls: Arc, /// The per-connection transport settings (timeouts, stream limits, /// congestion control). pub transport: super::transport::Config, } -impl Default for Config { - fn default() -> Self { +impl Config { + /// Dial with `tls` and the default transport settings. + pub fn new(tls: Arc) -> Self { Self { - alpn: Vec::new(), - verify: true, - roots: Vec::new(), - system_roots: true, - identity: None, + tls, transport: super::transport::Config::default(), } } diff --git a/rs/moq-uring/src/quic/driver/connection.rs b/rs/moq-uring/src/quic/driver/connection.rs index 71eb2753bc3..d82b6f6f7d5 100644 --- a/rs/moq-uring/src/quic/driver/connection.rs +++ b/rs/moq-uring/src/quic/driver/connection.rs @@ -295,9 +295,8 @@ impl Connection { /// The certificate chain the peer presented, end entity first, or `None` /// if it presented none. /// - /// A server only sees one when it asked - /// ([`ClientAuth`](crate::quic::server::ClientAuth)), and TLS already - /// validated it against the configured roots by the time this connection + /// A server only sees one when its TLS policy asked, and TLS already + /// validated it against that policy's roots by the time this connection /// exists: an invalid chain fails the handshake instead. So a `Some` here /// is an authenticated peer, and the chain is what names it. pub fn peer_identity(&self) -> Option>> { diff --git a/rs/moq-uring/src/quic/driver/mod.rs b/rs/moq-uring/src/quic/driver/mod.rs index a4586830e9e..e537c10d1ed 100644 --- a/rs/moq-uring/src/quic/driver/mod.rs +++ b/rs/moq-uring/src/quic/driver/mod.rs @@ -21,12 +21,9 @@ pub(crate) use connection::Shared; use std::{num::NonZeroU64, sync::Arc}; -use rustls::pki_types::pem::PemObject; -use rustls::pki_types::{CertificateDer, PrivateKeyDer}; - #[cfg(feature = "qlog")] use super::qlog; -use super::{Congestion, Error, Identity, client, server, transport}; +use super::{Error, client, server, transport}; use crate::udp; /// Streams of each kind a peer may have open at once. @@ -117,50 +114,7 @@ pub(crate) fn endpoint_config( /// Dial as `config` says. pub(crate) fn client_config(config: &client::Config) -> Result { - // Built explicitly rather than read from the process-wide default, so a - // consumer that never installed one still gets working TLS. - let provider = moq_crypto::provider::provider()?; - let builder = rustls::ClientConfig::builder_with_provider(provider.clone()) - .with_protocol_versions(&[&rustls::version::TLS13]) - .map_err(|err| Error::Tls(err.to_string()))?; - - // Nothing is checked with verification off, so nothing is loaded: a root - // path that does not exist must not fail a connection that was never - // going to look at it. - let verifier: Arc = match (config.verify, config.system_roots) { - (false, _) => Arc::new(NoVerify(provider.clone())), - // The platform verifier reads the OS trust store the way the OS means - // it, so extra roots go through it rather than around it. - (true, true) if config.roots.is_empty() => Arc::new( - rustls_platform_verifier::Verifier::new(provider.clone()).map_err(|err| Error::Tls(err.to_string()))?, - ), - (true, true) => Arc::new( - rustls_platform_verifier::Verifier::new_with_extra_roots(read_roots(&config.roots)?, provider.clone()) - .map_err(|err| Error::Tls(err.to_string()))?, - ), - // Trusting only the configured roots means a store built from - // scratch, never the platform's with ours added on top. - (true, false) => rustls::client::WebPkiServerVerifier::builder_with_provider( - Arc::new(root_store(&config.roots)?), - provider.clone(), - ) - .build() - .map_err(|err| Error::Tls(err.to_string()))?, - }; - let builder = builder.dangerous().with_custom_certificate_verifier(verifier); - - let mut tls = match &config.identity { - Some(identity) => { - let (chain, key) = keypair(identity)?; - builder - .with_client_auth_cert(chain, key) - .map_err(|err| Error::Tls(err.to_string()))? - } - None => builder.with_no_client_auth(), - }; - tls.alpn_protocols = alpn(&config.alpn); - - let mut client = moq_quic::client::Config::new(Arc::new(tls)); + let mut client = moq_quic::client::Config::new(config.tls.clone()); client.transport = transport_config(&config.transport)?; #[cfg(feature = "qlog")] with_qlog(&mut client.transport, &config.transport, qlog::Side::Client); @@ -177,50 +131,6 @@ pub(crate) fn server_config(config: &server::Config) -> Result>, - auth: &server::ClientAuth, -) -> Result, Error> { - // Built explicitly rather than read from the process-wide default, so a - // consumer that never installed one still gets working TLS. - let provider = moq_crypto::provider::provider()?; - let builder = rustls::ServerConfig::builder_with_provider(provider.clone()) - .with_protocol_versions(&[&rustls::version::TLS13]) - .map_err(|err| Error::Tls(err.to_string()))?; - - let verifier = match auth.roots() { - None => rustls::server::WebPkiClientVerifier::no_client_auth(), - // An empty store rejects every chain, so asking for a certificate with - // nothing to check it against refuses exactly the clients that obey. - Some(([], _)) => { - return Err(Error::Tls( - "client authentication needs at least one root certificate".to_string(), - )); - } - Some((roots, required)) => { - // Client certificates chain to the roots configured here, never to - // the platform store for public sites. - let builder = - rustls::server::WebPkiClientVerifier::builder_with_provider(Arc::new(root_store(roots)?), provider); - let builder = match required { - true => builder, - false => builder.allow_unauthenticated(), - }; - builder.build().map_err(|err| Error::Tls(err.to_string()))? - } - }; - - let (chain, key) = keypair(identity)?; - let mut tls = builder - .with_client_cert_verifier(verifier) - .with_single_cert(chain, key) - .map_err(|err| Error::Tls(err.to_string()))?; - tls.alpn_protocols = alpn; - Ok(Arc::new(tls)) -} - /// Attach the configured qlog sink, if any. /// /// moq-quic holds one sink per transport config, which is per endpoint on the @@ -255,105 +165,12 @@ fn transport_config(config: &transport::Config) -> Result { - Arc::new(moq_congestion::cubic::Config::default()) as Arc - } - Congestion::Delay => Arc::new(moq_congestion::bbr3::Config::default()), - Congestion::None => Arc::new(moq_congestion::Unlimited), - }; + transport.congestion = config.congestion.clone(); // A value that cannot go on the wire is an error: clamping it would ignore what was set. transport.validate()?; Ok(transport) } -/// ALPN protocols on the wire, which is a length-prefixed list of byte -/// strings rather than the `String`s a caller configures. -fn alpn(protocols: &[String]) -> Vec> { - protocols.iter().map(|proto| proto.as_bytes().to_vec()).collect() -} - -/// Split an [`Identity`] into the chain and key rustls wants. -fn keypair(identity: &Identity) -> Result<(Vec>, PrivateKeyDer<'static>), Error> { - let chain = CertificateDer::pem_slice_iter(identity.cert()) - .collect::, _>>() - .map_err(|err| Error::Tls(format!("certificate: {err}")))?; - if chain.is_empty() { - return Err(Error::Tls("certificate chain holds no certificates".to_string())); - } - let key = PrivateKeyDer::from_pem_slice(identity.key()).map_err(|err| Error::Tls(format!("key: {err}")))?; - Ok((chain, key)) -} - -/// Read every PEM certificate in each root file, naming the one that fails. -/// -/// A root is routinely a bundle of several CAs, and taking only the first -/// would reject a peer chaining to any of the others while looking configured. -/// A file holding none is an error rather than a store that trusts nothing. -fn read_roots(paths: &[std::path::PathBuf]) -> Result>, Error> { - let mut roots = Vec::new(); - for path in paths { - let pem = std::fs::read(path).map_err(|err| Error::Tls(format!("{}: {err}", path.display())))?; - let certs = CertificateDer::pem_slice_iter(&pem) - .collect::, _>>() - .map_err(|err| Error::Tls(format!("{}: {err}", path.display())))?; - if certs.is_empty() { - return Err(Error::Tls(format!("{}: no certificates", path.display()))); - } - roots.extend(certs); - } - Ok(roots) -} - -/// A store holding exactly the roots `paths` names, and nothing else. -fn root_store(paths: &[std::path::PathBuf]) -> Result { - let mut store = rustls::RootCertStore::empty(); - for root in read_roots(paths)? { - store.add(root).map_err(|err| Error::Tls(err.to_string()))?; - } - Ok(store) -} - -/// Accepts any server certificate, for -/// [`verify`](client::Config::verify) turned off. -#[derive(Debug)] -struct NoVerify(Arc); - -impl rustls::client::danger::ServerCertVerifier for NoVerify { - fn verify_server_cert( - &self, - _end_entity: &CertificateDer<'_>, - _intermediates: &[CertificateDer<'_>], - _server_name: &rustls::pki_types::ServerName<'_>, - _ocsp: &[u8], - _now: rustls::pki_types::UnixTime, - ) -> Result { - Ok(rustls::client::danger::ServerCertVerified::assertion()) - } - - fn verify_tls12_signature( - &self, - message: &[u8], - cert: &CertificateDer<'_>, - dss: &rustls::DigitallySignedStruct, - ) -> Result { - rustls::crypto::verify_tls12_signature(message, cert, dss, &self.0.signature_verification_algorithms) - } - - fn verify_tls13_signature( - &self, - message: &[u8], - cert: &CertificateDer<'_>, - dss: &rustls::DigitallySignedStruct, - ) -> Result { - rustls::crypto::verify_tls13_signature(message, cert, dss, &self.0.signature_verification_algorithms) - } - - fn supported_verify_schemes(&self) -> Vec { - self.0.signature_verification_algorithms.supported_schemes() - } -} - #[cfg(test)] mod tests { use std::num::NonZeroU64; diff --git a/rs/moq-uring/src/quic/mod.rs b/rs/moq-uring/src/quic/mod.rs index 3049fcb39a5..a5b73da1f89 100644 --- a/rs/moq-uring/src/quic/mod.rs +++ b/rs/moq-uring/src/quic/mod.rs @@ -42,82 +42,14 @@ pub use moq_quic::{Limits, LimitsError, Rate}; /// Stream identity, and the varint every QUIC code is. pub use moq_quic::{VarInt, stream}; -/// A TLS certificate chain and the private key that signs for it, as PEM. -/// One value, so neither half can be configured alone. -/// -/// The bytes are read once and held, not re-read per connection: a worker -/// group builds one of these before it spawns, so replacing the files on disk -/// afterwards cannot leave two workers serving different identities. -#[derive(Clone)] -pub struct Identity { - cert: Vec, - key: Vec, -} - -impl Identity { - /// Read the PEM chain at `cert` and the PEM key at `key`. - pub fn open(cert: impl AsRef, key: impl AsRef) -> Result { - let read = |path: &std::path::Path| { - std::fs::read(path).map_err(|err| Error::Tls(format!("{}: {err}", path.display()))) - }; - Ok(Self { - cert: read(cert.as_ref())?, - key: read(key.as_ref())?, - }) - } - - /// The same, from PEM already in hand. - pub fn from_pem(cert: impl Into>, key: impl Into>) -> Self { - Self { - cert: cert.into(), - key: key.into(), - } - } - - /// A TLS 1.3 server policy presenting this identity, speaking `alpn` in - /// preference order, and checking clients as `client_auth` says. - pub fn server( - &self, - alpn: &[impl AsRef], - client_auth: server::ClientAuth, - ) -> Result, Error> { - let alpn = alpn.iter().map(|proto| proto.as_ref().as_bytes().to_vec()).collect(); - driver::identity_server_config(self, alpn, &client_auth) - } - - /// The PEM certificate chain being presented, for a caller that publishes - /// its fingerprint. The key is deliberately not readable back out. - pub fn cert(&self) -> &[u8] { - &self.cert - } - - /// The PEM private key, for the backend loading it into its TLS stack. - pub(crate) fn key(&self) -> &[u8] { - &self.key - } -} - -impl std::fmt::Debug for Identity { - fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { - // Whatever else gets logged, the private key does not. - f.debug_struct("Identity") - .field("cert", &format_args!("{} PEM bytes", self.cert.len())) - .finish_non_exhaustive() - } -} - -/// The congestion control family a connection runs. +/// A built-in congestion control family, which converts into the factory +/// [`transport::Config::congestion`] holds. /// /// CUBIC is used for loss-based control and BBRv3 for delay-based control. -/// -/// The default is [`Loss`](Self::Loss), which is what moq-quic runs -/// unasked. An application carrying live media wants [`Delay`](Self::Delay) -/// and should say so; the relay does. -#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)] +#[derive(Clone, Copy, Debug, PartialEq, Eq)] #[non_exhaustive] pub enum Congestion { /// Loss-based: grows until it drops packets, so the send rate sawtooths. - #[default] Loss, /// Delay-based: tracks the measured delivery rate and RTT instead of /// waiting for loss, which keeps queues short and the send rate steady @@ -127,6 +59,16 @@ pub enum Congestion { None, } +impl From for std::sync::Arc { + fn from(family: Congestion) -> Self { + match family { + Congestion::Loss => std::sync::Arc::new(moq_congestion::cubic::Config::default()), + Congestion::Delay => std::sync::Arc::new(moq_congestion::bbr3::Config::default()), + Congestion::None => std::sync::Arc::new(moq_congestion::Unlimited), + } + } +} + /// Why a connection could not be set up or has ended. /// /// A [`web::Session`] reports it as [`web::Error::Connection`], decoding the @@ -168,9 +110,6 @@ pub enum Error { /// A lease wanted more than the connection receive window can hold. #[error("want exceeds the receive window")] TooLarge, - /// The TLS material a connection needs could not be loaded. - #[error("tls error: {0}")] - Tls(String), /// This CPU lacks a feature that rind, MoQ's cryptography, requires. #[error(transparent)] UnsupportedCpu(#[from] moq_crypto::cpu::UnsupportedCpu), diff --git a/rs/moq-uring/src/quic/server.rs b/rs/moq-uring/src/quic/server.rs index b7701ca468c..d21812754e4 100644 --- a/rs/moq-uring/src/quic/server.rs +++ b/rs/moq-uring/src/quic/server.rs @@ -3,44 +3,13 @@ use super::{Connection, Error}; use crate::udp; -/// Whether connecting clients are asked for a certificate, and against what. -/// -/// Asking for a certificate and validating one that arrives still admits a -/// client that presents none; only [`Required`](Self::Required) turns a -/// missing certificate into a failed handshake. -#[derive(Clone, Debug, Default)] -#[non_exhaustive] -pub enum ClientAuth { - /// Don't ask for a client certificate. - #[default] - None, - /// Ask, and verify against these roots if one is presented. A client that - /// presents none is still accepted. - Optional(Vec), - /// Require every client to present a certificate chaining to these roots. - Required(Vec), -} - -impl ClientAuth { - /// The roots a presented certificate is checked against, and whether one - /// is mandatory. - pub(crate) fn roots(&self) -> Option<(&[std::path::PathBuf], bool)> { - match self { - Self::None => None, - Self::Optional(roots) => Some((roots, false)), - Self::Required(roots) => Some((roots, true)), - } - } -} - /// What to present, what to speak, and who may connect. #[derive(Clone, Debug)] #[non_exhaustive] pub struct Config { /// The TLS policy every handshake uses: identity, ALPNs, and client - /// authentication. Build one from an [`Identity`](super::Identity) with - /// [`Identity::server`](super::Identity::server), or share a resolver with - /// other listeners. It must allow TLS 1.3 and name at least one ALPN. + /// authentication, shareable with other listeners. It must allow TLS 1.3 + /// and name at least one ALPN. pub tls: std::sync::Arc, /// The per-connection transport settings (timeouts, stream limits, /// congestion control). diff --git a/rs/moq-uring/src/quic/transport.rs b/rs/moq-uring/src/quic/transport.rs index 213d658b3a8..8e62eb400f8 100644 --- a/rs/moq-uring/src/quic/transport.rs +++ b/rs/moq-uring/src/quic/transport.rs @@ -10,7 +10,7 @@ use crate::quic::Limits; /// Separate from the role configs so a caller that already has these /// settings (a relay applying its `--quic-*` section, say) sets them once and /// hands the same value to a dial and a listener. -#[derive(Clone, Debug)] +#[derive(Clone)] #[non_exhaustive] pub struct Config { /// Close the connection after this long without activity. Must fit a QUIC @@ -30,8 +30,9 @@ pub struct Config { /// stream. /// - 4 MiB of unacknowledged outgoing data, whatever the peer allows. pub limits: Limits, - /// Which congestion controller to run. - pub congestion: Congestion, + /// The congestion controller each connection runs, BBRv3 by default. A + /// [`Congestion`] family converts into one with `.into()`. + pub congestion: std::sync::Arc, /// How often to send an ack-eliciting packet on an otherwise idle /// connection, `None` (the default) to derive it from the negotiated idle /// timeout, or zero to send none and let the idle timeout decide. See @@ -58,7 +59,7 @@ impl Default for Config { Self { idle_timeout: std::time::Duration::from_secs(10), limits: Limits::default(), - congestion: Congestion::default(), + congestion: Congestion::Delay.into(), keep_alive: None, mtu_discovery: false, #[cfg(feature = "qlog")] @@ -66,3 +67,19 @@ impl Default for Config { } } } + +// By hand, since a congestion factory carries no description of its own. +impl std::fmt::Debug for Config { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + let mut debug = f.debug_struct("Config"); + debug + .field("idle_timeout", &self.idle_timeout) + .field("limits", &self.limits) + .field("congestion", &format_args!("..")) + .field("keep_alive", &self.keep_alive) + .field("mtu_discovery", &self.mtu_discovery); + #[cfg(feature = "qlog")] + debug.field("qlog", &self.qlog); + debug.finish() + } +} diff --git a/rs/moq-uring/tests/integration/credit.rs b/rs/moq-uring/tests/integration/credit.rs index cc399e257c8..2f28a6aa58c 100644 --- a/rs/moq-uring/tests/integration/credit.rs +++ b/rs/moq-uring/tests/integration/credit.rs @@ -91,12 +91,7 @@ fn endpoints( .udp(UdpSocket::bind("127.0.0.1:0").expect("bind"), udp::Config::default()) .expect("socket") }; - let mut server_config = quic::server::Config::new( - quic::Identity::open(&certs.cert, &certs.key) - .expect("identity") - .server(&[ALPN], quic::server::ClientAuth::None) - .expect("tls"), - ); + let mut server_config = quic::server::Config::new(certs.server(&[ALPN])); server_config.transport.limits.receive_window = std::num::NonZeroU64::new(window as u64); server_config.transport.limits.stream_receive_window = std::num::NonZeroU64::new(stream_window as u64); let server = quic::Endpoint::new( @@ -105,9 +100,7 @@ fn endpoints( ) .expect("server endpoint"); let client = quic::Endpoint::new(quic::endpoint::Config::default(), socket(handle)).expect("dial-only endpoint"); - let mut dial = quic::client::Config::default(); - dial.alpn = vec![ALPN.to_string()]; - dial.verify = false; + let dial = quic::client::Config::new(support::client_tls(&[ALPN])); (server, client, dial) } @@ -236,12 +229,7 @@ fn chunked_writes_stop_at_the_window() { .udp(UdpSocket::bind("127.0.0.1:0").expect("bind"), udp::Config::default()) .expect("socket") }; - let mut server_config = quic::server::Config::new( - quic::Identity::open(&certs.cert, &certs.key) - .expect("identity") - .server(&[ALPN], quic::server::ClientAuth::None) - .expect("tls"), - ); + let mut server_config = quic::server::Config::new(certs.server(&[ALPN])); server_config.transport.limits.stream_receive_window = std::num::NonZeroU64::new(STREAM_WINDOW as u64); let server = quic::Endpoint::new( quic::endpoint::Config::default().with_server(server_config), @@ -249,9 +237,7 @@ fn chunked_writes_stop_at_the_window() { ) .expect("server endpoint"); let client = quic::Endpoint::new(quic::endpoint::Config::default(), socket(&handle)).expect("dial-only endpoint"); - let mut dial = quic::client::Config::default(); - dial.alpn = vec![ALPN.to_string()]; - dial.verify = false; + let dial = quic::client::Config::new(support::client_tls(&[ALPN])); let payload: Bytes = (0..3 * STREAM_WINDOW).map(|i| (i * 31 % 251) as u8).collect(); diff --git a/rs/moq-uring/tests/integration/echo.rs b/rs/moq-uring/tests/integration/echo.rs index 454cebb46f6..bcdbc462c0d 100644 --- a/rs/moq-uring/tests/integration/echo.rs +++ b/rs/moq-uring/tests/integration/echo.rs @@ -50,12 +50,7 @@ fn echo_tokio_peer() { let handle = worker.handle(); let certs = support::certs().expect("certificates"); - let server = quic::server::Config::new( - quic::Identity::open(&certs.cert, &certs.key) - .expect("identity") - .server(&[moq_tokio::quic::web::ALPN], quic::server::ClientAuth::None) - .expect("tls"), - ); + let server = quic::server::Config::new(certs.server(&[moq_tokio::quic::web::ALPN])); let socket = handle .udp(UdpSocket::bind("127.0.0.1:0").expect("bind"), udp::Config::default()) .expect("socket"); diff --git a/rs/moq-uring/tests/integration/endpoint.rs b/rs/moq-uring/tests/integration/endpoint.rs index fc30a1b1edd..f48c4c836e1 100644 --- a/rs/moq-uring/tests/integration/endpoint.rs +++ b/rs/moq-uring/tests/integration/endpoint.rs @@ -32,19 +32,11 @@ const CLOSE_CODE: u32 = 42; const CLOSE_REASON: &str = "bye"; fn server_config(certs: &support::Certs) -> quic::server::Config { - quic::server::Config::new( - quic::Identity::open(&certs.cert, &certs.key) - .expect("identity") - .server(&[ALPN], quic::server::ClientAuth::None) - .expect("tls"), - ) + quic::server::Config::new(certs.server(&[ALPN])) } fn dial_config() -> quic::client::Config { - let mut config = quic::client::Config::default(); - config.alpn = vec![ALPN.to_string()]; - config.verify = false; - config + quic::client::Config::new(support::client_tls(&[ALPN])) } /// One socket carries dials and accepts at once: two endpoints dial each @@ -768,54 +760,6 @@ fn accept_needs_a_server_config() { assert!(matches!(result, Err(quic::Error::NotServer)), "got {result:?}"); } -/// One [`quic::Identity`] serves every endpoint built from it, without ever -/// touching the files again. -/// -/// A worker group builds its endpoints on the worker threads, so re-reading -/// the certificate per endpoint means a certificate replaced on disk while -/// those threads are starting leaves earlier and later workers presenting -/// different identities, and a client pinning either one fails depending on -/// which worker accepts it. -#[test] -fn an_identity_is_read_once() { - let Some(mut worker) = worker() else { return }; - let handle = worker.handle(); - let certs = support::certs().expect("certificates"); - let config = server_config(&certs); - - // Whatever the endpoints below present, it did not come from here. - std::fs::remove_file(&certs.cert).expect("remove cert"); - std::fs::remove_file(&certs.key).expect("remove key"); - - let listener = |handle: &moq_uring::Handle| { - let sock = handle - .udp(UdpSocket::bind("127.0.0.1:0").expect("bind"), udp::Config::default()) - .expect("socket"); - quic::Endpoint::new(quic::endpoint::Config::default().with_server(config.clone()), sock).expect("endpoint") - }; - - // Two endpoints, as two workers would build them, then a real handshake - // against the second: the identity has to survive being cloned as well as - // being read. - let _first = listener(&handle); - let second = listener(&handle); - let server = second.local_addr(); - - worker - .block_on(async move { - let sock = handle - .udp(UdpSocket::bind("127.0.0.1:0").expect("bind"), udp::Config::default()) - .expect("client socket"); - handle.spawn(async move { - quic::client::connect(sock, &dial_config(), server, "localhost") - .await - .expect("dial"); - }); - second.accept().await.expect("accepted connection"); - }) - .expect("worker"); -} - /// Cancelling a dial after it created a connection closes that connection, /// even though its driver and the endpoint remain alive for sibling dials. #[test] diff --git a/rs/moq-uring/tests/integration/identity.rs b/rs/moq-uring/tests/integration/identity.rs index 03c9d4d4176..2a4671e93ad 100644 --- a/rs/moq-uring/tests/integration/identity.rs +++ b/rs/moq-uring/tests/integration/identity.rs @@ -28,19 +28,11 @@ fn worker() -> Option { const ALPN: &str = "moq-uring-identity"; fn server_config(certs: &support::Certs) -> quic::server::Config { - quic::server::Config::new( - quic::Identity::open(&certs.cert, &certs.key) - .expect("identity") - .server(&[ALPN], quic::server::ClientAuth::None) - .expect("tls"), - ) + quic::server::Config::new(certs.server(&[ALPN])) } fn dial_config() -> quic::client::Config { - let mut config = quic::client::Config::default(); - config.alpn = vec![ALPN.to_string()]; - config.verify = false; - config + quic::client::Config::new(support::client_tls(&[ALPN])) } fn socket(handle: &moq_uring::Handle) -> udp::Socket { diff --git a/rs/moq-uring/tests/integration/local_addr.rs b/rs/moq-uring/tests/integration/local_addr.rs index 0421e663694..56cabf9a44e 100644 --- a/rs/moq-uring/tests/integration/local_addr.rs +++ b/rs/moq-uring/tests/integration/local_addr.rs @@ -153,12 +153,7 @@ fn a_wildcard_endpoint_handshakes_on_a_secondary_address() { let Some(mut worker) = worker() else { return }; let handle = worker.handle(); let certs = crate::support::certs().expect("certificates"); - let server = quic::server::Config::new( - quic::Identity::open(&certs.cert, &certs.key) - .expect("identity") - .server(&[moq_tokio::quic::web::ALPN], quic::server::ClientAuth::None) - .expect("tls"), - ); + let server = quic::server::Config::new(certs.server(&[moq_tokio::quic::web::ALPN])); let socket = handle .udp(UdpSocket::bind("0.0.0.0:0").expect("bind"), udp::Config::default()) .expect("socket"); diff --git a/rs/moq-uring/tests/integration/mtu.rs b/rs/moq-uring/tests/integration/mtu.rs index 8db0ff4dee5..09a902470e6 100644 --- a/rs/moq-uring/tests/integration/mtu.rs +++ b/rs/moq-uring/tests/integration/mtu.rs @@ -195,12 +195,7 @@ fn endpoints( .udp(UdpSocket::bind("127.0.0.1:0").expect("bind"), udp::Config::default()) .expect("socket") }; - let mut server_config = quic::server::Config::new( - quic::Identity::open(&certs.cert, &certs.key) - .expect("identity") - .server(&[ALPN], quic::server::ClientAuth::None) - .expect("tls"), - ); + let mut server_config = quic::server::Config::new(certs.server(&[ALPN])); server_config.transport = transport.clone(); let server = quic::Endpoint::new( quic::endpoint::Config::default().with_server(server_config), @@ -209,9 +204,7 @@ fn endpoints( .expect("server endpoint"); let client = quic::Endpoint::new(quic::endpoint::Config::default(), socket(handle)).expect("client endpoint"); let path = Path::new(server.local_addr(), mtu); - let mut dial = quic::client::Config::default(); - dial.alpn = vec![ALPN.to_string()]; - dial.verify = false; + let mut dial = quic::client::Config::new(support::client_tls(&[ALPN])); dial.transport = transport.clone(); (server, client, dial, path) } diff --git a/rs/moq-uring/tests/integration/qlog.rs b/rs/moq-uring/tests/integration/qlog.rs index eca0a1240bb..680ce500b6f 100644 --- a/rs/moq-uring/tests/integration/qlog.rs +++ b/rs/moq-uring/tests/integration/qlog.rs @@ -49,12 +49,7 @@ fn a_connection_writes_a_trace() { let sink = quic::qlog::Sink::directory(dir.path()).expect("qlog sink"); - let mut server = quic::server::Config::new( - quic::Identity::open(&certs.cert, &certs.key) - .expect("identity") - .server(&[ALPN], quic::server::ClientAuth::None) - .expect("tls"), - ); + let mut server = quic::server::Config::new(certs.server(&[ALPN])); server.transport.qlog = Some(sink.clone()); let server_sock = handle @@ -65,9 +60,7 @@ fn a_connection_writes_a_trace() { .udp(UdpSocket::bind("127.0.0.1:0").expect("bind"), udp::Config::default()) .expect("client socket"); - let mut dial = quic::client::Config::default(); - dial.alpn = vec![ALPN.to_string()]; - dial.verify = false; + let mut dial = quic::client::Config::new(support::client_tls(&[ALPN])); dial.transport.qlog = Some(sink.clone()); handle.spawn(async move { diff --git a/rs/moq-uring/tests/integration/receive_credit.rs b/rs/moq-uring/tests/integration/receive_credit.rs index 8e67c32d350..023b05e2b0a 100644 --- a/rs/moq-uring/tests/integration/receive_credit.rs +++ b/rs/moq-uring/tests/integration/receive_credit.rs @@ -96,12 +96,7 @@ fn partial_frame_gives_way() { let high = broadcast.broadcast().create_track("high", timed()).expect("high track"); let certs = support::certs().expect("certificates"); - let server_config = quic::server::Config::new( - quic::Identity::open(&certs.cert, &certs.key) - .expect("identity") - .server(&[ALPN], quic::server::ClientAuth::None) - .expect("tls"), - ); + let server_config = quic::server::Config::new(certs.server(&[ALPN])); let socket = |handle: &moq_uring::Handle| { handle .udp(UdpSocket::bind("127.0.0.1:0").expect("bind"), udp::Config::default()) @@ -109,9 +104,7 @@ fn partial_frame_gives_way() { }; let server_sock = socket(&handle); let server_addr = server_sock.local_addr().expect("server addr"); - let mut dial = quic::client::Config::default(); - dial.alpn = vec![ALPN.to_string()]; - dial.verify = false; + let mut dial = quic::client::Config::new(support::client_tls(&[ALPN])); dial.transport.limits.receive_window = std::num::NonZeroU64::new(WINDOW as u64); let client_sock = socket(&handle); @@ -250,12 +243,7 @@ fn javascript_partial_frame_gives_way() { }); let certs = support::certs().expect("certificates"); - let mut server = quic::server::Config::new( - quic::Identity::open(&certs.cert, &certs.key) - .expect("identity") - .server(&["h3"], quic::server::ClientAuth::None) - .expect("tls"), - ); + let mut server = quic::server::Config::new(certs.server(&["h3"])); server.transport.limits.receive_window = std::num::NonZeroU64::new(WINDOW as u64); let sock = handle .udp(UdpSocket::bind("127.0.0.1:0").expect("bind"), udp::Config::default()) diff --git a/rs/moq-uring/tests/integration/session.rs b/rs/moq-uring/tests/integration/session.rs index 8e3188bc642..51fa6601c12 100644 --- a/rs/moq-uring/tests/integration/session.rs +++ b/rs/moq-uring/tests/integration/session.rs @@ -70,12 +70,7 @@ fn lite_session_over_the_worker() { group.finish().expect("finish group"); let certs = support::certs().expect("certificates"); - let server_config = quic::server::Config::new( - quic::Identity::open(&certs.cert, &certs.key) - .expect("identity") - .server(&[ALPN], quic::server::ClientAuth::None) - .expect("tls"), - ); + let server_config = quic::server::Config::new(certs.server(&[ALPN])); let server_sock = handle .udp(UdpSocket::bind("127.0.0.1:0").expect("bind"), udp::Config::default()) @@ -84,10 +79,8 @@ fn lite_session_over_the_worker() { let client_sock = handle .udp(UdpSocket::bind("127.0.0.1:0").expect("bind"), udp::Config::default()) .expect("client socket"); - let mut dial = quic::client::Config::default(); - dial.alpn = vec![ALPN.to_string()]; // The server's certificate is our own self-signed one; this is loopback. - dial.verify = false; + let dial = quic::client::Config::new(support::client_tls(&[ALPN])); // The publisher session runs as a worker task for the life of the test. let server_handle = handle.clone(); @@ -207,12 +200,7 @@ fn two_lite_sessions_share_the_server_socket() { group.finish().expect("finish group"); let certs = support::certs().expect("certificates"); - let server_config = quic::server::Config::new( - quic::Identity::open(&certs.cert, &certs.key) - .expect("identity") - .server(&[ALPN], quic::server::ClientAuth::None) - .expect("tls"), - ); + let server_config = quic::server::Config::new(certs.server(&[ALPN])); let server_sock = handle .udp(UdpSocket::bind("127.0.0.1:0").expect("bind"), udp::Config::default()) @@ -245,9 +233,7 @@ fn two_lite_sessions_share_the_server_socket() { } }); - let mut dial = quic::client::Config::default(); - dial.alpn = vec![ALPN.to_string()]; - dial.verify = false; + let dial = quic::client::Config::new(support::client_tls(&[ALPN])); let subs = [sub_a.clone(), sub_b.clone()]; worker @@ -308,168 +294,3 @@ fn two_lite_sessions_share_the_server_socket() { drop(sub_b); origins.join().expect("origin drivers"); } - -/// The client verifies for real, trusting only the certificate the server -/// presents: nothing handshakes unless the configured roots reach moq-quic. -#[test] -fn configured_roots_verify_the_server() { - let Some(mut worker) = worker() else { return }; - let handle = worker.handle(); - let certs = support::certs().expect("certificates"); - - let server_config = quic::server::Config::new( - quic::Identity::open(&certs.cert, &certs.key) - .expect("identity") - .server(&[ALPN], quic::server::ClientAuth::None) - .expect("tls"), - ); - - let server_sock = handle - .udp(UdpSocket::bind("127.0.0.1:0").expect("bind"), udp::Config::default()) - .expect("server socket"); - let server_addr = server_sock.local_addr().expect("server addr"); - let client_sock = handle - .udp(UdpSocket::bind("127.0.0.1:0").expect("bind"), udp::Config::default()) - .expect("client socket"); - let mut dial = quic::client::Config::default(); - dial.alpn = vec![ALPN.to_string()]; - // Verification on, trusting nothing but the certificate this server - // presents. The platform store is off, so if it leaked in, an unrelated - // public CA would be trusted too. - dial.system_roots = false; - dial.roots = vec![certs.cert.clone()]; - - handle.spawn(async move { - quic::server::accept(server_sock, &server_config) - .await - .expect("quic accept"); - }); - - worker - .block_on(async move { - let conn = quic::client::connect(client_sock, &dial, server_addr, "localhost") - .await - .expect("quic connect"); - assert_eq!(conn.protocol(), Some(ALPN), "negotiated ALPN"); - }) - .expect("worker"); -} - -/// `ClientAuth::Required` is the only setting that actually demands a client -/// certificate. `SSL_VERIFY_PEER` on its own validates one that arrives and -/// waves through a client that presents none, so a server meaning mTLS has to -/// say `Required`. -/// -/// A failed handshake never reaches `accept` (the endpoint keeps listening), -/// so the assertion is double-sided: the server accepts nothing, and the -/// client's connection dies. Under TLS 1.3 the client may finish its own -/// handshake before the server's verdict on the certificate it never sent -/// comes back, so `connect` returning `Ok` proves nothing; the refusal -/// arrives as the connection's terminal error. -#[test] -fn required_client_auth_refuses_an_anonymous_client() { - let Some(mut worker) = worker() else { return }; - let handle = worker.handle(); - let certs = support::certs().expect("certificates"); - - // Any root will do: the client presents nothing, so it fails before the - // chain is ever checked. - let server_config = quic::server::Config::new( - quic::Identity::open(&certs.cert, &certs.key) - .expect("identity") - .server(&[ALPN], quic::server::ClientAuth::Required(vec![certs.cert.clone()])) - .expect("tls"), - ); - - let server_sock = handle - .udp(UdpSocket::bind("127.0.0.1:0").expect("bind"), udp::Config::default()) - .expect("server socket"); - let client_sock = handle - .udp(UdpSocket::bind("127.0.0.1:0").expect("bind"), udp::Config::default()) - .expect("client socket"); - - let endpoint = quic::Endpoint::new( - quic::endpoint::Config::default().with_server(server_config), - server_sock, - ) - .expect("endpoint"); - let server_addr = endpoint.local_addr(); - - let mut dial = quic::client::Config::default(); - dial.alpn = vec![ALPN.to_string()]; - dial.verify = false; - // No identity: this client is anonymous. - - let accepted = std::rc::Rc::new(std::cell::Cell::new(false)); - let accept_flag = accepted.clone(); - handle.spawn(async move { - if endpoint.accept().await.is_ok() { - accept_flag.set(true); - } - }); - - worker - .block_on(async move { - match quic::client::connect(client_sock, &dial, server_addr, "localhost").await { - // The dial itself was refused; done. - Err(_) => {} - // Established locally, but the server's refusal closes it. - Ok(conn) => { - conn.closed().await; - } - } - }) - .expect("worker"); - - assert!( - !accepted.get(), - "a server requiring a certificate must refuse an anonymous client" - ); -} - -/// A root file is routinely a bundle of several CAs, and it has to be loaded -/// whole. -/// -/// Only the second CA in the bundle below signed the server, so a loader that -/// stops at the first certificate rejects a peer that is perfectly valid while -/// still looking configured. The relay reaches this through `listen.tls.root`. -#[test] -fn a_root_bundle_is_loaded_whole() { - let Some(mut worker) = worker() else { return }; - let handle = worker.handle(); - let bundle = support::bundle().expect("bundle"); - - let server_config = quic::server::Config::new( - quic::Identity::open(&bundle.cert, &bundle.key) - .expect("identity") - .server(&[ALPN], quic::server::ClientAuth::None) - .expect("tls"), - ); - - let server_sock = handle - .udp(UdpSocket::bind("127.0.0.1:0").expect("bind"), udp::Config::default()) - .expect("server socket"); - let server_addr = server_sock.local_addr().expect("server addr"); - let client_sock = handle - .udp(UdpSocket::bind("127.0.0.1:0").expect("bind"), udp::Config::default()) - .expect("client socket"); - let mut dial = quic::client::Config::default(); - dial.alpn = vec![ALPN.to_string()]; - dial.system_roots = false; - dial.roots = vec![bundle.roots.clone()]; - - handle.spawn(async move { - quic::server::accept(server_sock, &server_config) - .await - .expect("quic accept"); - }); - - worker - .block_on(async move { - let conn = quic::client::connect(client_sock, &dial, server_addr, "localhost") - .await - .expect("quic connect"); - assert_eq!(conn.protocol(), Some(ALPN), "negotiated ALPN"); - }) - .expect("worker"); -} diff --git a/rs/moq-uring/tests/integration/support.rs b/rs/moq-uring/tests/integration/support.rs index 09554c578ee..41b36a7faa2 100644 --- a/rs/moq-uring/tests/integration/support.rs +++ b/rs/moq-uring/tests/integration/support.rs @@ -31,59 +31,24 @@ pub fn certs() -> anyhow::Result { Ok(Certs { dir, cert, key }) } -/// A CA-signed server certificate plus a root file holding two CAs, only the -/// second of which signed it. -pub struct Bundle { - pub dir: tempfile::TempDir, - pub cert: std::path::PathBuf, - pub key: std::path::PathBuf, - /// Both CAs concatenated, in that order. - pub roots: std::path::PathBuf, +impl Certs { + /// A server TLS policy presenting this certificate and speaking `alpn`. + pub fn server(&self, alpn: &[&str]) -> std::sync::Arc { + let mut tls = moq_tokio::tls::Listen::default(); + tls.cert = vec![self.cert.clone()]; + tls.key = vec![self.key.clone()]; + tls.server_config(alpn.iter().map(|proto| proto.as_bytes().to_vec()).collect()) + .expect("server tls") + } } -/// Build one, for a test that a root file is loaded whole rather than to its -/// first certificate. -pub fn bundle() -> anyhow::Result { - let ca = |name: &str| -> anyhow::Result> { - let key = rind_rcgen::KeyPair::generate(&rcgen::PKCS_ECDSA_P256_SHA256)?; - let mut params = rcgen::CertificateParams::new(Vec::new()).map(|mut params| { - params.serial_number = Some(1u64.into()); - params - })?; - params.is_ca = rcgen::IsCa::Ca(rcgen::BasicConstraints::Unconstrained); - params.key_usages = vec![rcgen::KeyUsagePurpose::KeyCertSign, rcgen::KeyUsagePurpose::CrlSign]; - params.distinguished_name.push(rcgen::DnType::CommonName, name); - Ok(rcgen::CertifiedIssuer::self_signed(params, key)?) - }; - let unrelated = ca("moq unrelated ca")?; - let signer = ca("moq signing ca")?; - - let key = rind_rcgen::KeyPair::generate(&rcgen::PKCS_ECDSA_P256_SHA256)?; - let mut params = rcgen::CertificateParams::new(vec!["localhost".into()]).map(|mut params| { - params.serial_number = Some(1u64.into()); - params - })?; - params.use_authority_key_identifier_extension = true; - let cert = params.signed_by(&key, &signer)?; - - let dir = tempfile::tempdir()?; - let cert_path = dir.path().join("cert.pem"); - let key_path = dir.path().join("key.pem"); - let roots_path = dir.path().join("roots.pem"); - std::fs::write(&cert_path, cert.pem())?; - std::fs::write( - &key_path, - pem::encode(&pem::Pem::new("PRIVATE KEY", key.to_pkcs8().secret_pkcs8_der())), - )?; - // The signer is second on purpose: reading only the first certificate - // leaves the store unable to verify anything this server presents. - std::fs::write(&roots_path, format!("{}{}", unrelated.pem(), signer.pem()))?; - Ok(Bundle { - dir, - cert: cert_path, - key: key_path, - roots: roots_path, - }) +/// A client TLS policy offering `alpn` and trusting whatever certificate the server presents. +pub fn client_tls(alpn: &[&str]) -> std::sync::Arc { + let mut tls = moq_tokio::tls::Connect::default(); + tls.insecure = Some(true); + let mut tls = tls.build().expect("client tls"); + tls.alpn_protocols = alpn.iter().map(|proto| proto.as_bytes().to_vec()).collect(); + std::sync::Arc::new(tls) } /// Open a WebTransport session to `request` from tokio, through the driver's diff --git a/rs/moq-uring/tests/integration/teardown.rs b/rs/moq-uring/tests/integration/teardown.rs index 54bf3abb6de..ca4bdabef2e 100644 --- a/rs/moq-uring/tests/integration/teardown.rs +++ b/rs/moq-uring/tests/integration/teardown.rs @@ -69,19 +69,11 @@ fn a_send_staged_on_the_way_out_still_goes() { } fn server_config(certs: &support::Certs) -> quic::server::Config { - quic::server::Config::new( - quic::Identity::open(&certs.cert, &certs.key) - .expect("identity") - .server(&[ALPN], quic::server::ClientAuth::None) - .expect("tls"), - ) + quic::server::Config::new(certs.server(&[ALPN])) } fn dial_config() -> quic::client::Config { - let mut config = quic::client::Config::default(); - config.alpn = vec![ALPN.to_string()]; - config.verify = false; - config + quic::client::Config::new(support::client_tls(&[ALPN])) } /// The one-shot client shape: dial, close, wait for the close to be published, diff --git a/rs/moq-uring/tests/integration/web.rs b/rs/moq-uring/tests/integration/web.rs index fff6f13be99..577b44b8dae 100644 --- a/rs/moq-uring/tests/integration/web.rs +++ b/rs/moq-uring/tests/integration/web.rs @@ -40,12 +40,7 @@ fn timed() -> moq_net::track::Info { /// Build the uring server endpoint serving `h3`. fn h3_endpoint(handle: &moq_uring::Handle, certs: &support::Certs) -> quic::Endpoint { - let server = quic::server::Config::new( - quic::Identity::open(&certs.cert, &certs.key) - .expect("identity") - .server(&["h3"], quic::server::ClientAuth::None) - .expect("tls"), - ); + let server = quic::server::Config::new(certs.server(&["h3"])); let sock = handle .udp(UdpSocket::bind("127.0.0.1:0").expect("bind"), udp::Config::default()) .expect("socket"); @@ -1045,9 +1040,7 @@ fn dropped_session_delivers_close(explicit: bool) { .udp(UdpSocket::bind("127.0.0.1:0").expect("bind"), udp::Config::default()) .expect("client socket"); let client = quic::Endpoint::new(quic::endpoint::Config::default(), sock).expect("client endpoint"); - let mut dial = quic::client::Config::default(); - dial.alpn = vec!["h3".to_string()]; - dial.verify = false; + let dial = quic::client::Config::new(support::client_tls(&["h3"])); worker .block_on(async { @@ -1164,9 +1157,7 @@ fn cancelling_a_pending_web_handshake(send_settings: bool) { .udp(UdpSocket::bind("127.0.0.1:0").expect("bind"), udp::Config::default()) .expect("client socket"); let client = quic::Endpoint::new(quic::endpoint::Config::default(), sock).expect("client endpoint"); - let mut dial = quic::client::Config::default(); - dial.alpn = vec!["h3".to_string()]; - dial.verify = false; + let dial = quic::client::Config::new(support::client_tls(&["h3"])); worker .block_on(async { diff --git a/rs/moq-uring/tests/integration/web_headers.rs b/rs/moq-uring/tests/integration/web_headers.rs index d8e0e3f9ae1..f46204e9fde 100644 --- a/rs/moq-uring/tests/integration/web_headers.rs +++ b/rs/moq-uring/tests/integration/web_headers.rs @@ -19,12 +19,7 @@ pub fn run(pending: usize, ready: usize, iterations: u64) -> Duration { let mut worker = Worker::new(Config::default()).expect("worker"); let handle = worker.handle(); let certs = support::certs().expect("certificates"); - let mut server = quic::server::Config::new( - quic::Identity::open(&certs.cert, &certs.key) - .expect("identity") - .server(&["h3"], quic::server::ClientAuth::None) - .expect("TLS"), - ); + let mut server = quic::server::Config::new(certs.server(&["h3"])); server.transport.limits.max_streams = std::num::NonZeroU64::new((pending + ready + 3) as u64); let socket = handle .udp(UdpSocket::bind("127.0.0.1:0").expect("bind"), udp::Config::default()) diff --git a/rs/moq-uring/tests/integration/workers.rs b/rs/moq-uring/tests/integration/workers.rs index f630deab11c..1ba366db60c 100644 --- a/rs/moq-uring/tests/integration/workers.rs +++ b/rs/moq-uring/tests/integration/workers.rs @@ -98,8 +98,7 @@ fn a_steered_group_serves_a_shared_port() { .map(|(member, stop)| { let accepted = accepted.clone(); let stop = stop.clone(); - let cert = certs.cert.clone(); - let key = certs.key.clone(); + let tls = certs.server(&[ALPN]); let ready = ready.clone(); std::thread::spawn(move || { let shard = member.shard(); @@ -109,12 +108,7 @@ fn a_steered_group_serves_a_shared_port() { // steers with, and there is no other way to hand one over. let socket = handle.udp(member, udp::Config::default()).expect("socket"); - let server = quic::server::Config::new( - quic::Identity::open(cert, key) - .expect("identity") - .server(&[ALPN], quic::server::ClientAuth::None) - .expect("tls"), - ); + let server = quic::server::Config::new(tls); let endpoint = quic::Endpoint::new(quic::endpoint::Config::default().with_server(server), socket) .expect("endpoint"); @@ -140,9 +134,7 @@ fn a_steered_group_serves_a_shared_port() { // accepted is the steering assertion (see the module docs). let mut client_worker = client_worker; let handle = client_worker.handle(); - let mut dial = quic::client::Config::default(); - dial.alpn = vec![ALPN.to_string()]; - dial.verify = false; + let dial = quic::client::Config::new(support::client_tls(&[ALPN])); client_worker .block_on(async { diff --git a/rs/qmux/benches/receive.rs b/rs/qmux/benches/receive.rs index aee4dfe477b..a549547c11f 100644 --- a/rs/qmux/benches/receive.rs +++ b/rs/qmux/benches/receive.rs @@ -95,9 +95,6 @@ impl Receivers { let (client, server) = tokio::io::duplex(64 << 10); let (client, server) = if websocket { use qmux::ws::Upgraded; - use tokio_tungstenite::{WebSocketStream, tungstenite::protocol::Role}; - let client = WebSocketStream::from_raw_socket(client, Role::Client, None).await; - let server = WebSocketStream::from_raw_socket(server, Role::Server, None).await; let (client, server) = tokio::join!( Upgraded::new(client, version.alpn()) .with_session_config(config.clone()) diff --git a/rs/qmux/examples/bench.rs b/rs/qmux/examples/bench.rs index 26df9d26775..108e1378d47 100644 --- a/rs/qmux/examples/bench.rs +++ b/rs/qmux/examples/bench.rs @@ -191,15 +191,12 @@ async fn connect(transport: Transport, tls: &Tls, addr: SocketAddr) -> anyhow::R .await?, ), Transport::Ws => spawn(qmux::ws::Client::new().connect(&format!("ws://{addr}")).await?), - Transport::Wss => { - let connector = tokio_tungstenite::Connector::Rustls(tls.client.clone()); - spawn( - qmux::ws::Client::new() - .with_connector(connector) - .connect(&format!("wss://localhost:{}", addr.port())) - .await?, - ) - } + Transport::Wss => spawn( + qmux::ws::Client::new() + .with_tls(tls.client.clone()) + .connect(&format!("wss://localhost:{}", addr.port())) + .await?, + ), }) } diff --git a/rs/qmux/src/error.rs b/rs/qmux/src/error.rs index b602ced964b..47edb84a54d 100644 --- a/rs/qmux/src/error.rs +++ b/rs/qmux/src/error.rs @@ -105,7 +105,7 @@ pub enum Error { /// The WebSocket failed. #[cfg(feature = "ws")] #[error(transparent)] - WebSocket(Arc), + WebSocket(WebSocketError), /// Datagrams are disabled on this session, or the peer does not accept them. #[error("datagrams not supported")] @@ -168,16 +168,36 @@ impl From for Error { } } +/// A WebSocket failure, opaque so the WebSocket implementation is not part of this API. #[cfg(feature = "ws")] -impl From for Error { - fn from(err: tokio_tungstenite::tungstenite::Error) -> Self { +#[derive(Clone, Debug)] +pub struct WebSocketError(Arc); + +#[cfg(feature = "ws")] +impl std::fmt::Display for WebSocketError { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + self.0.fmt(f) + } +} + +#[cfg(feature = "ws")] +impl std::error::Error for WebSocketError { + fn source(&self) -> Option<&(dyn std::error::Error + 'static)> { + self.0.source() + } +} + +#[cfg(feature = "ws")] +impl Error { + /// Classify a WebSocket failure. + pub(crate) fn websocket(err: tokio_tungstenite::tungstenite::Error) -> Self { // A non-101 upgrade response carries the HTTP status; surface it as a // distinct, transport-agnostic variant so callers can act on terminal // rejections (e.g. auth) without downcasting tungstenite. if let tokio_tungstenite::tungstenite::Error::Http(response) = &err { return Self::Http(response.status().as_u16()); } - Self::WebSocket(Arc::new(err)) + Self::WebSocket(WebSocketError(Arc::new(err))) } } @@ -217,13 +237,13 @@ mod tests { .status(http::StatusCode::UNAUTHORIZED) .body(None::>) .unwrap(); - let err: Error = tungstenite::Error::Http(Box::new(response)).into(); + let err = Error::websocket(tungstenite::Error::Http(Box::new(response))); assert!(matches!(err, Error::Http(401))); } #[test] fn non_http_preserves_source() { - let err: Error = tungstenite::Error::ConnectionClosed.into(); + let err = Error::websocket(tungstenite::Error::ConnectionClosed); assert!(matches!(err, Error::WebSocket(_))); } } diff --git a/rs/qmux/src/lib.rs b/rs/qmux/src/lib.rs index 704fa9f6de0..935455690a9 100644 --- a/rs/qmux/src/lib.rs +++ b/rs/qmux/src/lib.rs @@ -63,6 +63,8 @@ pub mod ws; pub use config::{Config, Protocol}; pub use driver::Driver; pub use error::Error; +#[cfg(feature = "ws")] +pub use error::WebSocketError; pub use moq_quic::mux::Version; pub use moq_quic::{Limits, LimitsError}; pub use session::{RecvStream, SendGroup, SendStream, Session}; diff --git a/rs/qmux/src/transport/websocket.rs b/rs/qmux/src/transport/websocket.rs index 2a6c6180f45..4ee4d8e3821 100644 --- a/rs/qmux/src/transport/websocket.rs +++ b/rs/qmux/src/transport/websocket.rs @@ -123,7 +123,7 @@ impl Transport for WsTransport { let message = match message { Some(Ok(message)) => message, // Keep the WebSocket error, so callers can still tell a protocol fault apart. - Some(Err(err)) => return Poll::Ready(Err(err.into())), + Some(Err(err)) => return Poll::Ready(Err(Error::websocket(err))), None => return Poll::Ready(Err(Error::Closed)), }; match message { diff --git a/rs/qmux/src/ws.rs b/rs/qmux/src/ws.rs index 4f4c11ecc88..b85b4162f4e 100644 --- a/rs/qmux/src/ws.rs +++ b/rs/qmux/src/ws.rs @@ -6,31 +6,28 @@ use crate::protocol::validate_protocol; use crate::transport::WsTransport; use moq_net::transport::Address; use moq_quic::Side; -use tokio_tungstenite::{MaybeTlsStream, WebSocketStream, tungstenite}; +use tokio_tungstenite::tungstenite; +use tokio_tungstenite::{MaybeTlsStream, WebSocketStream}; use crate::{Config, Driver, Error, Session, Version, alpn}; /// The largest WebSocket message or frame a session under `config` reads. /// -/// Each message carries one record, or one frame on draft-00, so cap a socket upgraded -/// elsewhere (see [`Upgraded`]) at this: a larger one is refused anyway, but only after -/// the socket buffered it whole. -pub fn max_message_size(config: &Config) -> usize { +/// Each message carries one record, or one frame on draft-00, so the socket is capped +/// at this: a larger one would be refused anyway, but only after buffering it whole. +fn max_message_size(config: &Config) -> usize { usize::try_from(config.max_record_size) .unwrap_or(usize::MAX) .max(crate::MAX_FRAME_SIZE) } -/// `config` with its message and frame sizes held to [`max_message_size`] of `session`. -fn capped( - config: Option, - session: &Config, -) -> tungstenite::protocol::WebSocketConfig { +/// The socket settings for a session under `session`: messages and frames held to +/// [`max_message_size`]. +fn capped(session: &Config) -> tungstenite::protocol::WebSocketConfig { let max = max_message_size(session); - let mut config = config.unwrap_or_default(); - config.max_message_size = Some(config.max_message_size.map_or(max, |size| size.min(max))); - config.max_frame_size = Some(config.max_frame_size.map_or(max, |size| size.min(max))); - config + tungstenite::protocol::WebSocketConfig::default() + .max_message_size(Some(max)) + .max_frame_size(Some(max)) } /// Keep-alive configuration for WebSocket transports. @@ -69,31 +66,29 @@ impl Default for KeepAlive { } } -/// Wrap an already-upgraded WebSocket as a QMux session. +/// Run a QMux session over a connection whose HTTP upgrade to WebSocket already +/// finished. /// -/// Use this when the WebSocket handshake was performed by an external -/// framework (e.g. axum) and the caller already has the negotiated -/// `Sec-WebSocket-Protocol` value, which names the QMux wire-format version and -/// application protocol. Disable Nagle on the TCP socket underneath, as with -/// [`Server::accept`]. +/// Use this when an HTTP server (e.g. hyper or axum) answered the upgrade and the +/// caller has the raw upgraded IO and the negotiated `Sec-WebSocket-Protocol` +/// value, which names the QMux wire-format version and application protocol. +/// [`connect`](Self::connect) speaks the WebSocket client role and +/// [`accept`](Self::accept) the server role. The IO must hold nothing but +/// WebSocket frames: no handshake bytes, and none already read past it. Disable +/// Nagle on the TCP socket underneath, as with [`Server::accept`]. pub struct Upgraded { - ws: T, + io: T, alpn: String, keep_alive: Option, socket_stats: Option, session: Config, } -impl Upgraded -where - T: futures::Stream> - + futures::Sink - + Unpin, -{ - /// Wrap `ws`, whose handshake negotiated the `Sec-WebSocket-Protocol` value `alpn`. - pub fn new(ws: T, alpn: &str) -> Self { +impl Upgraded { + /// Wrap `io`, whose upgrade negotiated the `Sec-WebSocket-Protocol` value `alpn`. + pub fn new(io: T, alpn: &str) -> Self { Self { - ws, + io, alpn: alpn.to_string(), keep_alive: None, socket_stats: None, @@ -109,9 +104,6 @@ where /// Use `config` for every session setting except the wire-format version /// and application protocol, which the handshake negotiates. - /// - /// Cap the socket's message and frame sizes at [`max_message_size`] of `config`, - /// since each record is one message. pub fn with_session_config(mut self, config: Config) -> Self { self.session = config; self @@ -136,32 +128,33 @@ where /// arrived, so its stream and datagram limits are known. Fails when the subprotocol /// names no QMux version, or with [`Error::HandshakeTimeout`] if the parameters never /// arrive. - pub async fn connect(self) -> Result<(Session, Driver>), Error> { + pub async fn connect(self) -> Result<(Session, Driver>>), Error> { self.start(Side::Client).await } /// Wrap as a server-side session once it is established, as with /// [`connect`](Self::connect). - pub async fn accept(self) -> Result<(Session, Driver>), Error> { + pub async fn accept(self) -> Result<(Session, Driver>>), Error> { self.start(Side::Server).await } - async fn start(self, side: Side) -> Result<(Session, Driver>), Error> { + async fn start(self, side: Side) -> Result<(Session, Driver>>), Error> { let (version, protocol) = alpn::parse(Some(&self.alpn))?; let config = self.session.clone().with_negotiated(version, protocol); - let transport = self.into_transport()?; - Session::start(transport, side, config).await - } - - fn into_transport(self) -> Result, Error> { - let mut transport = WsTransport::new(self.ws, Address::Uri); + let role = match side { + Side::Client => tungstenite::protocol::Role::Client, + Side::Server => tungstenite::protocol::Role::Server, + }; + let ws = WebSocketStream::from_raw_socket(self.io, role, Some(capped(&config))).await; + let mut transport = WsTransport::new(ws, Address::Uri); if let Some(stats) = self.socket_stats { transport = transport.with_socket_stats(stats); } - Ok(match self.keep_alive { + let transport = match self.keep_alive { Some(ka) => transport.with_keep_alive(ka)?, None => transport, - }) + }; + Session::start(transport, side, config).await } } @@ -179,11 +172,10 @@ where pub struct Client { protocols: Vec<(String, Vec)>, require_protocol: bool, - config: Option, keep_alive: Option, session: Config, #[cfg(feature = "wss")] - connector: Option, + tls: Option>, } impl Client { @@ -216,12 +208,6 @@ impl Client { self } - /// Set the WebSocket configuration, its message and frame sizes held to [`max_message_size`]. - pub fn with_config(mut self, config: tungstenite::protocol::WebSocketConfig) -> Self { - self.config = Some(config); - self - } - /// Send periodic Pings and close the session if the peer goes silent. /// /// WebSocket has no built-in idle timeout, so without this a crashed peer @@ -237,16 +223,16 @@ impl Client { /// The WebSocket handshake already settled the protocol, so the session starts /// without awaiting the peer's transport parameters, failing later with /// [`Error::HandshakeTimeout`] if they never arrive. The socket's message and frame - /// sizes are held to [`max_message_size`] of `config`, since each record is one message. + /// sizes are held to one record of `config`. pub fn with_session_config(mut self, config: Config) -> Self { self.session = config; self } - /// Set the TLS connector for secure WebSocket connections. + /// Dial `wss://` URLs with this TLS policy instead of the platform's root store. #[cfg(feature = "wss")] - pub fn with_connector(mut self, connector: tokio_tungstenite::Connector) -> Self { - self.connector = Some(connector); + pub fn with_tls(mut self, tls: std::sync::Arc) -> Self { + self.tls = Some(tls); self } @@ -268,7 +254,7 @@ impl Client { validate_protocol(a)?; } - let mut request = url.into_client_request().map_err(Error::from)?; + let mut request = url.into_client_request().map_err(Error::websocket)?; let entries = self.protocols.iter().map(|(a, vs)| (a.as_str(), vs.as_slice())); let protocol_value = alpn::build(entries, self.require_protocol).join(", "); @@ -278,19 +264,19 @@ impl Client { http::HeaderValue::from_str(&protocol_value).map_err(|_| Error::InvalidProtocol(protocol_value))?, ); - let ws_config = capped(self.config, &self.session); + let ws_config = capped(&self.session); // `true` disables Nagle: records are already batched, see `Server::accept`. #[cfg(feature = "wss")] let (ws_stream, response) = { let connector = match ( - &self.connector, + &self.tls, request .uri() .scheme_str() .is_some_and(|scheme| scheme.eq_ignore_ascii_case("wss")), ) { - (Some(connector), _) => Some(connector.clone()), + (Some(tls), _) => Some(tokio_tungstenite::Connector::Rustls(tls.clone())), (None, true) => { let provider = moq_crypto::provider::provider()?; let native = rustls_native_certs::load_native_certs(); @@ -314,13 +300,13 @@ impl Client { }; tokio_tungstenite::connect_async_tls_with_config(request, Some(ws_config), true, connector) .await - .map_err(Error::from)? + .map_err(Error::websocket)? }; #[cfg(not(feature = "wss"))] let (ws_stream, response) = tokio_tungstenite::connect_async_with_config(request, Some(ws_config), true) .await - .map_err(Error::from)?; + .map_err(Error::websocket)?; let negotiated = response .headers() @@ -409,7 +395,7 @@ impl Server { /// The WebSocket handshake already settled the protocol, so the session starts /// without awaiting the peer's transport parameters, failing later with /// [`Error::HandshakeTimeout`] if they never arrive. The socket's message and frame - /// sizes are held to [`max_message_size`] of `config`, since each record is one message. + /// sizes are held to one record of `config`. pub fn with_session_config(mut self, config: Config) -> Self { self.session = config; self @@ -495,8 +481,9 @@ impl Server { .unwrap()) }; - let ws = tokio_tungstenite::accept_hdr_async_with_config(socket, callback, Some(capped(None, &self.session))) - .await?; + let ws = tokio_tungstenite::accept_hdr_async_with_config(socket, callback, Some(capped(&self.session))) + .await + .map_err(Error::websocket)?; let (version, protocol) = negotiated .lock()