1use crate::client::ClientConfig;
4use crate::quic::CongestionControl;
5use crate::quic::Resolved;
6use crate::quic::ServerId;
7use crate::server::ServerConfig;
8use crate::tls::{FingerprintVerifier, ServeCerts};
9use std::net;
10use std::sync::Arc;
11use std::time::Duration;
12use url::Url;
13
14pub use web_transport_quinn;
15
16fn apply_transport(transport: &mut quinn::TransportConfig, quic: Resolved) {
18 transport.max_idle_timeout(Some(quic.idle_timeout.try_into().expect("idle timeout out of range")));
19 transport.keep_alive_interval(quic.keep_alive);
20
21 if !quic.mtu_discovery {
23 transport.mtu_discovery_config(None);
24 }
25
26 let max_streams = quinn::VarInt::from_u64(quic.max_streams).unwrap_or(quinn::VarInt::MAX);
27 transport.max_concurrent_bidi_streams(max_streams);
28 transport.max_concurrent_uni_streams(max_streams);
29
30 if let Some(gso) = quic.gso {
32 transport.enable_segmentation_offload(gso);
33 }
34
35 if let Some(family) = quic.congestion_control {
37 transport.congestion_controller_factory(congestion_factory(family));
38 }
39}
40
41fn congestion_factory(family: CongestionControl) -> Arc<dyn quinn::congestion::ControllerFactory + Send + Sync> {
43 match family {
44 CongestionControl::Loss => Arc::new(quinn::congestion::CubicConfig::default()),
45 CongestionControl::Delay => Arc::new(quinn::congestion::BbrConfig::default()),
46 }
47}
48
49#[derive(Debug, thiserror::Error)]
51#[non_exhaustive]
52pub enum Error {
53 #[error("failed to bind UDP socket")]
55 BindSocket(#[source] std::io::Error),
56
57 #[error("failed to create QUIC endpoint")]
59 CreateEndpoint(#[source] std::io::Error),
60
61 #[error("no async runtime")]
63 NoRuntime,
64
65 #[error("failed to get local address")]
67 LocalAddr(#[source] std::io::Error),
68
69 #[error("failed to resolve bind address")]
71 ResolveBind(#[source] std::io::Error),
72
73 #[error("invalid DNS name")]
75 InvalidDnsName,
76
77 #[error("failed DNS lookup")]
79 DnsLookup(#[source] std::io::Error),
80
81 #[error("no DNS entries")]
83 NoDnsEntries,
84
85 #[error("failed to fetch fingerprint")]
87 FetchFingerprint(#[source] reqwest::Error),
88
89 #[error("fingerprint request failed")]
91 FingerprintStatus(#[source] reqwest::Error),
92
93 #[error("failed to read fingerprint")]
95 ReadFingerprint(#[source] reqwest::Error),
96
97 #[error("invalid fingerprint")]
99 InvalidFingerprint(#[from] hex::FromHexError),
100
101 #[error("url scheme must be 'https', 'moqt', or 'moql'")]
103 InvalidScheme,
104
105 #[error("unsupported URL scheme: {0}")]
107 UnsupportedScheme(String),
108
109 #[error("missing handshake data")]
111 MissingHandshake,
112
113 #[error("missing ALPN")]
115 MissingAlpn,
116
117 #[error("failed to decode ALPN")]
119 DecodeAlpn(#[from] std::string::FromUtf8Error),
120
121 #[error("unsupported ALPN: {0}")]
123 UnsupportedAlpn(String),
124
125 #[error("missing server name for raw QUIC connection")]
127 MissingServerName,
128
129 #[error("failed to construct URL from server name")]
131 BuildUrl(#[source] url::ParseError),
132
133 #[error("quic_lb_nonce must be at least 4")]
135 QuicLbNonceTooSmall,
136
137 #[error("connection ID length ({0}) exceeds maximum of 20")]
139 QuicLbCidTooLong(usize),
140
141 #[error("failed to build client certificate verifier")]
143 ClientVerifier(#[source] rustls::server::VerifierBuilderError),
144
145 #[error(transparent)]
147 NoInitialCipherSuite(#[from] quinn::crypto::rustls::NoInitialCipherSuite),
148
149 #[error(transparent)]
151 Connect(#[from] quinn::ConnectError),
152
153 #[error(transparent)]
155 Connection(#[from] quinn::ConnectionError),
156
157 #[error(transparent)]
159 Client(#[from] web_transport_quinn::ClientError),
160
161 #[error(transparent)]
163 ConnectRejected(#[from] crate::ConnectError),
164
165 #[error(transparent)]
167 Server(#[from] web_transport_quinn::ServerError),
168
169 #[error("failed to establish QUIC connection")]
171 Establish(#[source] quinn::ConnectionError),
172
173 #[error("failed to receive WebTransport request")]
175 RecvRequest(#[source] web_transport_quinn::ServerError),
176
177 #[error(transparent)]
179 Tls(#[from] crate::tls::Error),
180}
181
182type Result<T> = std::result::Result<T, Error>;
183
184#[derive(Clone)]
187pub(crate) struct QuinnClient {
188 pub quic: quinn::Endpoint,
189 pub transport: Arc<quinn::TransportConfig>,
190 pub http_bootstrap: bool,
192 pub host_name: Option<String>,
194}
195
196impl QuinnClient {
197 pub fn new(config: &ClientConfig) -> Result<Self> {
198 let socket = crate::bind::udp(config.bind).map_err(Error::BindSocket)?;
199
200 let mut transport = quinn::TransportConfig::default();
201 apply_transport(&mut transport, config.quic.resolve());
202 let transport = Arc::new(transport);
203
204 let runtime = quinn::default_runtime().ok_or(Error::NoRuntime)?;
206 let endpoint_config = quinn::EndpointConfig::default();
207
208 let quic = quinn::Endpoint::new(endpoint_config, None, socket, runtime).map_err(Error::CreateEndpoint)?;
210
211 Ok(Self {
212 quic,
213 transport,
214 http_bootstrap: config.tls.allows_http_bootstrap(),
215 host_name: config.tls.host_name.clone(),
216 })
217 }
218
219 pub async fn connect(
220 &self,
221 tls: &rustls::ClientConfig,
222 url: Url,
223 versions: &moq_net::Versions,
224 ) -> Result<web_transport_quinn::Session> {
225 let mut url = url;
226 let mut config = tls.clone();
227
228 let host = url.host().ok_or(Error::InvalidDnsName)?.to_string();
229 let port = url.port().unwrap_or(443);
230
231 let local = self.quic.local_addr().map_err(Error::LocalAddr)?;
237 let addrs = tokio::net::lookup_host((host.clone(), port))
238 .await
239 .map_err(Error::DnsLookup)?;
240 let ip = crate::util::pick_addr(addrs, local).ok_or(Error::NoDnsEntries)?;
241
242 if url.scheme() == "http" {
243 if self.http_bootstrap {
247 let mut fingerprint = url.clone();
249 fingerprint.set_path("/certificate.sha256");
250 fingerprint.set_query(None);
251 fingerprint.set_fragment(None);
252
253 tracing::warn!(url = %fingerprint, "performing insecure HTTP request for certificate");
254
255 let resp = reqwest::get(fingerprint.as_str())
256 .await
257 .map_err(Error::FetchFingerprint)?
258 .error_for_status()
259 .map_err(Error::FingerprintStatus)?;
260
261 let fingerprint = resp.text().await.map_err(Error::ReadFingerprint)?;
262 let fingerprint = hex::decode(fingerprint.trim())?;
263
264 let verifier = FingerprintVerifier::new(config.crypto_provider().clone(), vec![fingerprint]);
265 config.dangerous().set_certificate_verifier(Arc::new(verifier));
266 } else {
267 tracing::warn!(
268 "ignoring insecure http:// fingerprint bootstrap; using the configured TLS verification"
269 );
270 }
271
272 url.set_scheme("https").expect("failed to set scheme");
273 }
274
275 let alpns: Vec<Vec<u8>> = match url.scheme() {
276 "https" => vec![web_transport_quinn::ALPN.as_bytes().to_vec()],
277 "moqt" | "moql" => versions.alpns().iter().map(|alpn| alpn.as_bytes().to_vec()).collect(),
278 _ => return Err(Error::InvalidScheme),
279 };
280
281 config.alpn_protocols = alpns;
282 config.key_log = Arc::new(rustls::KeyLogFile::new());
283
284 let config: quinn::crypto::rustls::QuicClientConfig = config.try_into()?;
285 let mut config = quinn::ClientConfig::new(Arc::new(config));
286 config.transport_config(self.transport.clone());
287
288 tracing::debug!(%url, %ip, "connecting");
289
290 let host_name = self.host_name.clone().unwrap_or(host);
292
293 let connection = self.quic.connect_with(config, ip, &host_name)?.await?;
294 tracing::Span::current().record("id", connection.stable_id());
295
296 let mut request = web_transport_quinn::proto::ConnectRequest::new(url.clone());
297 for alpn in versions.alpns() {
298 request = request.with_protocol(alpn.to_string());
299 }
300
301 let session = match url.scheme() {
302 "https" => web_transport_quinn::Session::connect(connection, request)
303 .await
304 .map_err(map_client_error)?,
305 "moqt" | "moql" => {
306 let handshake = connection
307 .handshake_data()
308 .ok_or(Error::MissingHandshake)?
309 .downcast::<quinn::crypto::rustls::HandshakeData>()
310 .unwrap();
311
312 let alpn = handshake.protocol.ok_or(Error::MissingAlpn)?;
313 let alpn = String::from_utf8(alpn)?;
314
315 let response = web_transport_quinn::proto::ConnectResponse::OK.with_protocol(alpn);
316 web_transport_quinn::Session::raw(connection, request, response)
317 }
318 _ => return Err(Error::UnsupportedScheme(url.scheme().to_string())),
319 };
320
321 Ok(session)
322 }
323}
324
325impl Error {
326 pub(crate) fn connect_error(&self) -> Option<crate::ConnectError> {
327 match self {
328 Self::ConnectRejected(err) => Some(*err),
329 Self::Client(err) => classify_client_error(err),
330 _ => None,
331 }
332 }
333}
334
335fn map_client_error(err: web_transport_quinn::ClientError) -> Error {
336 if let Some(err) = classify_client_error(&err) {
337 return err.into();
338 }
339
340 err.into()
341}
342
343fn classify_client_error(err: &web_transport_quinn::ClientError) -> Option<crate::ConnectError> {
344 match err {
345 web_transport_quinn::ClientError::HttpError(err) => classify_connect_error(err),
346 _ => None,
347 }
348}
349
350fn classify_connect_error(err: &web_transport_quinn::ConnectError) -> Option<crate::ConnectError> {
351 match err {
352 web_transport_quinn::ConnectError::ErrorStatus(status) => crate::ConnectError::from_status_u16(status.as_u16()),
353 web_transport_quinn::ConnectError::ProtoError(err) => classify_proto_error(err),
354 _ => None,
355 }
356}
357
358fn classify_proto_error(err: &web_transport_quinn::proto::ConnectError) -> Option<crate::ConnectError> {
359 match err {
360 web_transport_quinn::proto::ConnectError::ErrorStatus(status)
361 | web_transport_quinn::proto::ConnectError::WrongStatus(Some(status)) => {
362 crate::ConnectError::from_status_u16(status.as_u16())
363 }
364 _ => None,
365 }
366}
367
368pub(crate) struct QuinnServer {
371 pub quic: quinn::Endpoint,
372 pub certs: Arc<ServeCerts>,
373}
374
375impl QuinnServer {
376 pub fn new(config: ServerConfig) -> Result<Self> {
377 let mut transport = quinn::TransportConfig::default();
378 apply_transport(&mut transport, config.quic.resolve());
379 let transport = Arc::new(transport);
380
381 let provider = crate::crypto::provider();
382
383 let certs = ServeCerts::new(provider.clone());
384 certs.load_certs(&config.tls)?;
385 let certs = Arc::new(certs);
386
387 let tls_builder = rustls::ServerConfig::builder_with_provider(provider.clone())
388 .with_protocol_versions(&[&rustls::version::TLS13])
389 .map_err(crate::tls::Error::from)?;
390
391 let mut tls = if config.tls.root.is_empty() {
392 tls_builder.with_no_client_auth().with_cert_resolver(certs.clone())
393 } else {
394 let roots = config.tls.load_roots()?;
395 let verifier = rustls::server::WebPkiClientVerifier::builder_with_provider(Arc::new(roots), provider)
396 .allow_unauthenticated()
397 .build()
398 .map_err(Error::ClientVerifier)?;
399 tls_builder
400 .with_client_cert_verifier(verifier)
401 .with_cert_resolver(certs.clone())
402 };
403
404 let mut alpns: Vec<Vec<u8>> = config
406 .versions()
407 .alpns()
408 .iter()
409 .map(|alpn| alpn.as_bytes().to_vec())
410 .collect();
411 alpns.push(web_transport_quinn::ALPN.as_bytes().to_vec());
412
413 tls.alpn_protocols = alpns;
414 tls.key_log = Arc::new(rustls::KeyLogFile::new());
415
416 let tls: quinn::crypto::rustls::QuicServerConfig = tls.try_into()?;
417 let mut tls = quinn::ServerConfig::with_crypto(Arc::new(tls));
418 tls.transport_config(transport);
419
420 if let Some(addr) = config.quic.preferred_v4 {
423 tls.preferred_address_v4(Some(addr));
424 }
425 if let Some(addr) = config.quic.preferred_v6 {
426 tls.preferred_address_v6(Some(addr));
427 }
428
429 let runtime = quinn::default_runtime().ok_or(Error::NoRuntime)?;
431
432 let listen =
433 crate::util::resolve(config.bind.as_deref(), crate::server::DEFAULT_BIND).map_err(Error::ResolveBind)?;
434
435 let mut endpoint_config = quinn::EndpointConfig::default();
437 if let Some(server_id) = config.quic.quic_lb_id {
438 let nonce_len = config.quic.quic_lb_nonce.unwrap_or(8);
439 if nonce_len < 4 {
440 return Err(Error::QuicLbNonceTooSmall);
441 }
442
443 let cid_len = 1 + server_id.len() + nonce_len;
444 if cid_len > 20 {
445 return Err(Error::QuicLbCidTooLong(cid_len));
446 }
447
448 tracing::info!(
449 ?server_id,
450 nonce_len,
451 "using QUIC-LB compatible connection ID generation"
452 );
453 endpoint_config.cid_generator(move || Box::new(ServerIdGenerator::new(server_id.clone(), nonce_len)));
454 }
455
456 let socket = crate::bind::udp(listen).map_err(Error::BindSocket)?;
457
458 let quic = quinn::Endpoint::new(endpoint_config, Some(tls), socket, runtime).map_err(Error::CreateEndpoint)?;
460
461 tokio::spawn(crate::tls::reload_certs(certs.clone(), config.tls.clone()));
464
465 Ok(Self { quic, certs })
466 }
467
468 pub fn accept(&self) -> impl std::future::Future<Output = Option<quinn::Incoming>> + '_ {
469 self.quic.accept()
470 }
471
472 pub fn certificates(&self) -> crate::tls::Certificates {
473 crate::tls::Certificates::new(self.certs.info.clone())
474 }
475
476 pub fn local_addr(&self) -> Result<net::SocketAddr> {
477 self.quic.local_addr().map_err(Error::LocalAddr)
478 }
479
480 pub fn close(&self) {
481 self.quic.close(quinn::VarInt::from_u32(0), b"server shutdown");
482 }
483}
484
485pub(crate) async fn accept(
492 conn: quinn::Incoming,
493 alpns: Vec<&'static str>,
494) -> Result<(
495 web_transport_quinn::Session,
496 Option<Url>,
497 Option<crate::tls::PeerIdentity>,
498)> {
499 let mut conn = conn.accept()?;
500
501 let handshake = conn
502 .handshake_data()
503 .await?
504 .downcast::<quinn::crypto::rustls::HandshakeData>()
505 .unwrap();
506
507 let alpn = handshake.protocol.ok_or(Error::MissingAlpn)?;
508 let alpn = String::from_utf8(alpn)?;
509 let host = handshake.server_name.unwrap_or_default();
510
511 tracing::debug!(%host, ip = %conn.remote_address(), %alpn, "accepting");
512
513 let conn = conn.await.map_err(Error::Establish)?;
515
516 let span = tracing::Span::current();
517 span.record("id", conn.stable_id()); tracing::debug!(%host, ip = %conn.remote_address(), %alpn, "accepted");
519
520 match alpn.as_str() {
521 web_transport_quinn::ALPN => {
522 let request = web_transport_quinn::Request::accept(conn)
525 .await
526 .map_err(Error::RecvRequest)?;
527 let url = Some(request.url.clone());
528 let identity = crate::tls::PeerIdentity::from_any(request.conn().peer_identity());
529
530 let mut response = web_transport_quinn::proto::ConnectResponse::OK;
531 if let Some(protocol) = request.protocols.iter().find(|p| alpns.contains(&p.as_str())) {
537 response = response.with_protocol(protocol);
538 }
539 let session = request.respond(response).await.map_err(Error::Server)?;
540 Ok((session, url, identity))
541 }
542 alpn if alpns.contains(&alpn) => {
547 let host_str = if host.contains(':') {
553 format!("[{}]", host)
554 } else {
555 host.clone()
556 };
557 let url = format!("moqt://{}", host_str).parse::<Url>().map_err(Error::BuildUrl)?;
558 let request = web_transport_quinn::proto::ConnectRequest::new(url);
559 let response = web_transport_quinn::proto::ConnectResponse::OK.with_protocol(alpn);
560 let identity = crate::tls::PeerIdentity::from_any(conn.peer_identity());
561 let session = web_transport_quinn::Session::raw(conn, request, response);
563 Ok((session, None, identity))
564 }
565 _ => Err(Error::UnsupportedAlpn(alpn)),
566 }
567}
568
569struct ServerIdGenerator {
572 server_id: ServerId,
573 nonce_len: usize,
574}
575
576impl ServerIdGenerator {
577 fn new(server_id: ServerId, nonce_len: usize) -> Self {
578 Self { server_id, nonce_len }
579 }
580}
581
582impl quinn::ConnectionIdGenerator for ServerIdGenerator {
583 fn generate_cid(&mut self) -> quinn::ConnectionId {
584 use rand::RngExt;
585 let cid_len = self.cid_len();
586 let mut cid = Vec::with_capacity(cid_len);
587 cid.push((cid_len - 1) as u8);
589 cid.extend(self.server_id.0.iter());
590 cid.extend(rand::rng().random_iter::<u8>().take(self.nonce_len));
591 quinn::ConnectionId::new(cid.as_slice())
592 }
593
594 fn cid_len(&self) -> usize {
595 1 + self.server_id.len() + self.nonce_len
596 }
597
598 fn cid_lifetime(&self) -> Option<Duration> {
599 None
600 }
601}
602
603#[cfg(test)]
604mod tests {
605 use super::*;
606
607 #[test]
610 fn congestion_factory_maps_each_family() {
611 let now = std::time::Instant::now();
612 let mtu = 1200;
613
614 let loss = congestion_factory(CongestionControl::Loss).build(now, mtu);
615 assert!(loss.into_any().downcast::<quinn::congestion::Cubic>().is_ok());
616
617 let delay = congestion_factory(CongestionControl::Delay).build(now, mtu);
618 assert!(delay.into_any().downcast::<quinn::congestion::Bbr>().is_ok());
619 }
620
621 #[tokio::test]
624 async fn delay_reaches_the_live_connection() {
625 let server_config = ServerConfig {
626 bind: Some("127.0.0.1:0".to_string()),
627 tls: crate::tls::Server {
628 generate: vec!["localhost".into()],
629 ..Default::default()
630 },
631 quic: crate::quic::Server {
632 congestion_control: Some(CongestionControl::Delay),
633 ..Default::default()
634 },
635 ..Default::default()
636 };
637
638 let server = QuinnServer::new(server_config).expect("server init");
639 let addr = server.local_addr().expect("local addr");
640
641 let accepted = tokio::spawn(async move {
642 let incoming = server.accept().await.expect("no incoming connection");
643 let conn = incoming.accept().expect("accept").await.expect("handshake");
644 conn.congestion_state()
645 .into_any()
646 .downcast::<quinn::congestion::Bbr>()
647 .is_ok()
648 });
649
650 let mut tls_config = crate::tls::Client::default();
652 tls_config.disable_verify = Some(true);
653
654 let client_config = ClientConfig {
655 bind: "127.0.0.1:0".parse().unwrap(),
656 tls: tls_config,
657 quic: crate::quic::Client {
658 congestion_control: Some(CongestionControl::Delay),
659 ..Default::default()
660 },
661 ..Default::default()
662 };
663
664 let tls = client_config.tls.build().expect("tls config");
665 let client = QuinnClient::new(&client_config).expect("client init");
666 let url: Url = format!("moqt://127.0.0.1:{}", addr.port()).parse().unwrap();
668
669 tokio::time::timeout(Duration::from_secs(5), async move {
672 let session = client
673 .connect(&tls, url, &moq_net::Versions::default())
674 .await
675 .expect("connect failed");
676
677 assert!(
679 session
680 .congestion_state()
681 .into_any()
682 .downcast::<quinn::congestion::Bbr>()
683 .is_ok(),
684 "client connection is not running BBR"
685 );
686 assert!(
687 accepted.await.expect("server task panicked"),
688 "server connection is not running BBR"
689 );
690 })
691 .await
692 .expect("test timed out");
693 }
694}